mito2/worker/
handle_catchup.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Handling catchup request.
16
17use std::sync::Arc;
18
19use common_telemetry::tracing::warn;
20use common_telemetry::{debug, error, info};
21use store_api::logstore::LogStore;
22use store_api::region_engine::{RegionRole, SettableRegionRoleState};
23use store_api::region_request::RegionCatchupRequest;
24use store_api::storage::RegionId;
25
26use crate::error::{self, Result};
27use crate::region::MitoRegion;
28use crate::region::catchup::RegionCatchupTask;
29use crate::region::opener::RegionOpener;
30use crate::request::OptionOutputTx;
31use crate::wal::entry_distributor::WalEntryReceiver;
32use crate::worker::RegionWorkerLoop;
33
34impl<S: LogStore> RegionWorkerLoop<S> {
35    pub(crate) async fn handle_catchup_request(
36        &mut self,
37        region_id: RegionId,
38        request: RegionCatchupRequest,
39        entry_receiver: Option<WalEntryReceiver>,
40        sender: OptionOutputTx,
41    ) {
42        let Some(region) = self.regions.get_region(region_id) else {
43            sender.send(Err(error::RegionNotFoundSnafu { region_id }.build()));
44            return;
45        };
46
47        if region.is_writable() {
48            debug!("Region {region_id} is writable, skip catchup");
49            sender.send(Ok(0));
50            return;
51        }
52
53        if self.catchup_regions.is_region_exists(region_id) {
54            warn!("Region {region_id} under catching up");
55            sender.send(Err(error::RegionBusySnafu { region_id }.build()));
56            return;
57        }
58
59        // If the memtable is not empty or the manifest has been updated, we need to reopen the region.
60        let region = match self.reopen_region_if_needed(region).await {
61            Ok(region) => region,
62            Err(e) => {
63                sender.send(Err(e));
64                return;
65            }
66        };
67
68        self.catchup_regions.insert_region(region_id);
69        let catchup_regions = self.catchup_regions.clone();
70        let wal = self.wal.clone();
71        let allow_stale_entries = self.config.allow_stale_entries;
72        common_runtime::spawn_global(async move {
73            let mut task = RegionCatchupTask::new(region.clone(), wal, allow_stale_entries)
74                .with_entry_receiver(entry_receiver)
75                .with_expected_last_entry_id(request.entry_id)
76                .with_replay_checkpoint_entry_id(request.checkpoint.map(|c| c.entry_id));
77
78            match task.run().await {
79                Ok(_) => {
80                    if request.set_writable {
81                        region.set_role(RegionRole::Leader);
82                        // Finalize leadership: persist backfilled metadata.
83                        if let Err(err) = region
84                            .set_role_state_gracefully(SettableRegionRoleState::Leader)
85                            .await
86                        {
87                            error!(err; "Failed to set region {region_id} to leader");
88                        }
89                    }
90                    catchup_regions.remove_region(region_id);
91                    sender.send(Ok(0));
92                }
93                Err(err) => {
94                    error!(err; "Failed to catchup region {region_id}");
95                    catchup_regions.remove_region(region_id);
96                    sender.send(Err(err));
97                }
98            }
99        });
100    }
101
102    /// Reopens a region.
103    pub(crate) async fn reopen_region(
104        &mut self,
105        region: &Arc<MitoRegion>,
106    ) -> Result<Arc<MitoRegion>> {
107        let region_id = region.region_id;
108        let manifest_version = region.manifest_ctx.manifest_version().await;
109        let flushed_entry_id = region.version_control.current().last_entry_id;
110        info!(
111            "Reopening the region: {region_id}, manifest version: {manifest_version}, flushed entry id: {flushed_entry_id}"
112        );
113        let reopened_region = RegionOpener::new(
114            region_id,
115            region.table_dir(),
116            region.access_layer.path_type(),
117            self.memtable_builder_provider.clone(),
118            self.object_store_manager.clone(),
119            self.purge_scheduler.clone(),
120            self.puffin_manager_factory.clone(),
121            self.intermediate_manager.clone(),
122            self.time_provider.clone(),
123            self.file_ref_manager.clone(),
124            self.partition_expr_fetcher.clone(),
125        )
126        .cache(Some(self.cache_manager.clone()))
127        .series_index_store(self.series_index_store.clone())
128        .series_index_purger(self.series_index_purger.clone())
129        .hook(self.plugins.get())
130        .options(region.version().options.clone())?
131        .skip_wal_replay(true)
132        .open(&self.config, &self.wal)
133        .await?;
134        debug_assert!(!reopened_region.is_writable());
135        self.register_region(reopened_region.clone());
136        if let Some(state) = &self.series_index_task_state {
137            state.wake();
138        }
139
140        Ok(reopened_region)
141    }
142
143    async fn reopen_region_if_needed(
144        &mut self,
145        region: Arc<MitoRegion>,
146    ) -> Result<Arc<MitoRegion>> {
147        // Note: Currently, We protect the split brain by ensuring the memtable table is empty.
148        // It's expensive to execute catch-up requests without `set_writable=true` multiple times.
149        let version = region.version();
150        let is_empty_memtable = version.memtables.is_empty();
151
152        // Utilizes the short circuit evaluation.
153        let region = if !is_empty_memtable || region.manifest_ctx.has_update().await? {
154            if !is_empty_memtable {
155                warn!(
156                    "Region {} memtables is not empty, which should not happen, manifest version: {}, last entry id: {}",
157                    region.region_id,
158                    region.manifest_ctx.manifest_version().await,
159                    region.version_control.current().last_entry_id
160                );
161            }
162            self.reopen_region(&region).await?
163        } else {
164            region
165        };
166
167        Ok(region)
168    }
169}
mito2/worker/handle_catchup.rs

mito2/worker/
handle_catchup.rs
