// Copyright 2023 PingCAP, Inc. use std::{ collections::HashSet, ops::RangeInclusive, sync::{ Arc, Mutex, atomic::{AtomicU64, Ordering}, }, thread::{JoinHandle, sleep}, time::Duration, }; use cloud_worker::native_br::{ RestoreProgressResponse, RestoreState, test_utils::NativeBrSvcClient, }; use engine_traits::ObjectCache; use kvengine::dfs::{Dfs, new_dfs_from_config}; use native_br::{ backup::BackupConfig, backup_worker::BackupWorker, common::send_request_to_store, error::Error, limiter::ThroughputLimiter, restore::{RestoreConfig, get_cluster_backup_meta, get_cluster_backup_meta_async}, restore_keyspace::{self, ReportRestoreStepTrait, RestoreStep}, }; use pd_client::{PdClient, pd_control::PdControl, util::get_all_stores_except_tiflash_async}; use rand::Rng; use rfengine::service_worker::DfsWorkerStatus; use security::{GetSecurityManager, HttpClient, SecurityManager}; use test_cloud_server::{ client::ClusterClient, keyspace::{ClusterKeyspaceClient, KeyspaceBackup, KeyspaceManager}, try_wait_result, }; use test_pd_client::TestPdClient; use tikv_util::{box_err, config::ReadableDuration, info, time::Instant, warn}; use tokio::runtime::Runtime; use crate::{ BACKUP_COUNTER, BACKUP_TOLERATED_ERR_COUNTER, BROKEN_BACKUP_COUNTER, ON_DEMAND_BACKUP_COUNTER, ONLINE_SNAPSHOT_RESTORE_COUNTER, RESTORE_COUNTER, RESTORE_TOLERATED_ERR_COUNTER, create_new_keyspace, remaining_timeout, }; const FETCH_WAL_TIMEOUT: ReadableDuration = ReadableDuration::secs(30); // Use small duration to cover scene of error tolerance more easily. const FETCH_WAL_TIMEOUT_TOLERATE_ERR: ReadableDuration = ReadableDuration::secs(1); const RFENGINE_HTTP_ERROR_RETRY_TIMES: usize = 20; const BACKUP_NAME_BOUNDARY: Duration = Duration::from_millis(1100); const DFS_WORKER_POLL_INTERVAL: Duration = Duration::from_secs(1); const ON_DEMAND_BACKUP_RETRY_TIMES: usize = 2; const ONLINE_SNAPSHOT_CAPTURE_STEP: &str = "capture online snapshot"; const ONLINE_SNAPSHOT_STATUS_POLL_INTERVAL: Duration = Duration::from_millis(100); static ONLINE_SNAPSHOT_RESTORE_TASK_ID: AtomicU64 = AtomicU64::new(1); pub(crate) fn do_restore_keyspace( pd_client: Arc, pd_control: Option, runtime: &Runtime, config: RestoreConfig, keyspace: u32, target_keyspace: u32, backup_name: &str, truncate_ts: u64, reporter: Arc, object_cache: Option, limiter: Option>, ) -> native_br::Result { let dfs_config = config.dfs.clone(); let dfs = new_dfs_from_config(dfs_config); restore_keyspace::restore_keyspace( keyspace, target_keyspace, backup_name, None, dfs, config, pd_client, pd_control, runtime, Some(truncate_ts), None, reporter, object_cache, limiter, ) } pub(crate) struct BackupWorkloadConfig { pub periodic_interval: Duration, pub batch_interval: Duration, pub instant_interval: Duration, pub on_demand_interval_range: RangeInclusive, pub operation_timeout: Duration, pub workload_timeout: Duration, } fn do_restore_keyspace_from_online_snapshot( br_client: &NativeBrSvcClient, runtime: &Runtime, keyspace_manager: &KeyspaceManager, source_keyspace: u32, target_keyspace: u32, restore_id: u64, timeout: Duration, ) -> crate::Result<(RestoreProgressResponse, KeyspaceBackup)> { let source_name = keyspace_manager .get_keyspace_meta(source_keyspace) .unwrap() .name(); let target_name = keyspace_manager .get_keyspace_meta(target_keyspace) .unwrap() .name(); // Quiesce source test workloads only until tikv-api has chosen branch_ts. // The production capture and restore continue after this guard is released. let source_lock = keyspace_manager.get_keyspace_lock(source_keyspace); let mut source_guard = Some(runtime.block_on(source_lock.mutex_lock())); let mut expected = keyspace_manager.backup_keyspace(source_keyspace, 0); runtime.block_on(br_client.restore_keyspace_from_online_snapshot( restore_id, source_name, target_name.clone(), ))?; let start = Instant::now(); while start.saturating_elapsed() < timeout { let progress = runtime.block_on(br_client.get_restore_progress(restore_id, target_name.clone()))?; let capture_started = progress.progress.step == ONLINE_SNAPSHOT_CAPTURE_STEP || progress.progress.progress > 10 || progress.status == RestoreState::Succeed; if capture_started { // All source writes that committed before branch_ts are now reflected // in `expected`; later writes are truncated from the branch. drop(source_guard.take()); } match progress.status { RestoreState::Succeed => { if progress.restore_ts == 0 { return Err(box_err!( "online snapshot restore {} succeeded without restore_ts", restore_id )); } expected.backup_ts = progress.restore_ts; return Ok((progress, expected)); } RestoreState::Error => { return Err(box_err!( "online snapshot restore {} failed: {}", restore_id, progress.error )); } RestoreState::Pending | RestoreState::Init | RestoreState::Running => {} } sleep(ONLINE_SNAPSHOT_STATUS_POLL_INTERVAL); } panic!( "online snapshot restore {} remained non-terminal after {:?}; the submitted task may still be running", restore_id, timeout ); } pub(crate) struct BackupWorkload { client: ClusterClient, pd_client: Arc, br_client: NativeBrSvcClient, keyspace_manager: KeyspaceManager, backup_config: BackupConfig, backup_worker: BackupWorker, dfs: Arc, security_mgr: Arc, http_client: HttpClient, config: BackupWorkloadConfig, } impl BackupWorkload { fn new( client: ClusterClient, pd_client: Arc, br_client: NativeBrSvcClient, keyspace_manager: KeyspaceManager, backup_config: BackupConfig, dfs: Arc, config: BackupWorkloadConfig, ) -> Self { let backup_worker = BackupWorker::new( backup_config.clone(), pd_client.clone(), config.periodic_interval, config.batch_interval, ); let security_mgr = pd_client.get_security_mgr(); let http_client = security_mgr.http_client(hyper::Client::builder()).unwrap(); Self { client, pd_client, br_client, keyspace_manager, backup_config, backup_worker, dfs, security_mgr, http_client, config, } } pub(crate) fn spawn( client: ClusterClient, pd_client: Arc, br_client: NativeBrSvcClient, keyspace_manager: KeyspaceManager, backup_config: BackupConfig, dfs: Arc, config: BackupWorkloadConfig, ) -> tokio::task::JoinHandle<()> { let workload = Self::new( client, pd_client, br_client, keyspace_manager, backup_config, dfs, config, ); tokio::spawn(workload.run()) } async fn run(mut self) { let start_time = Instant::now(); let instant_interval = self.config.instant_interval.max(BACKUP_NAME_BOUNDARY); let mut last_instant_backup_time = start_time - instant_interval; let mut last_on_demand_backup_time = start_time; let mut on_demand_interval = self.next_on_demand_interval(); while let Some(remaining) = remaining_timeout(start_time, self.config.workload_timeout) { if last_on_demand_backup_time.saturating_elapsed() >= on_demand_interval && remaining >= self.config.operation_timeout { if let Err(err) = self.run_on_demand_cycle().await { warn!("on-demand backup cycle failed, retry later"; "err" => err); } last_on_demand_backup_time = Instant::now(); last_instant_backup_time = Instant::now(); on_demand_interval = self.next_on_demand_interval(); continue; } if last_instant_backup_time.saturating_elapsed() >= instant_interval { match self.backup_instant().await { Ok(()) => last_instant_backup_time = Instant::now(), Err(err) if Self::is_backup_error_retryable(&err) => { warn!("backup failed, retry: {:?}", err); tokio::time::sleep(Duration::from_millis(500).min(remaining)).await; } Err(err) => panic!("backup failed: {:?}", err), } continue; } let instant_wait = instant_interval.saturating_sub(last_instant_backup_time.saturating_elapsed()); let on_demand_wait = if remaining >= self.config.operation_timeout { on_demand_interval.saturating_sub(last_on_demand_backup_time.saturating_elapsed()) } else { remaining }; tokio::time::sleep(instant_wait.min(on_demand_wait).min(remaining)).await; } self.backup_worker.stop(); info!("backup thread exit"); } fn next_on_demand_interval(&self) -> Duration { let interval = rand::thread_rng().gen_range(self.config.on_demand_interval_range.clone()); info!("scheduled next on-demand backup"; "interval" => ?interval); interval } async fn backup_instant(&self) -> Result<(), Error> { // Back up the RefStore of one random keyspace while TiKV still creates a // cluster-wide backup. Uniform selection intentionally leaves some large // keyspaces without a restorable backup in a run. let keyspace_id = { let mut rng = rand::thread_rng(); self.keyspace_manager.get_uniform_random_keyspace(&mut rng) }; let lock = self.keyspace_manager.get_keyspace_lock(keyspace_id); let guard = lock.mutex_lock().await; let backup_ts = self.client.get_ts().into_inner(); let mut keyspace_backup = self .keyspace_manager .backup_keyspace(keyspace_id, backup_ts); // Let writes continue after the captured timestamp, but keep restore from // replacing this keyspace until the cluster backup is durable. let shared_guard = guard.downgrade(); let backup_res = self.backup_worker.instant_backup().await?; keyspace_backup.backup_name = Some(backup_res.backup_file.name().to_string()); self.keyspace_manager.add_backup(keyspace_backup); drop(shared_guard); let backup_meta = get_cluster_backup_meta_async( self.dfs.as_ref(), backup_res.backup_file.name().to_string(), ) .await; info!( "instant backup success, keyspace {}, res {:?}, backup_meta {:?}", keyspace_id, backup_res, backup_meta, ); self.record_backup_result(&backup_meta); Ok(()) } async fn run_on_demand_cycle(&mut self) -> Result<(), String> { info!("stopping automatic backup for an on-demand backup cycle"); self.backup_worker.stop(); // Backup names have second granularity. Keep the last automatic backup and // the first on-demand attempt in different name buckets. tokio::time::sleep(BACKUP_NAME_BOUNDARY).await; let backup_result = self.backup_on_demand().await; self.restart_automatic_backup() .await .unwrap_or_else(|err| panic!("failed to restart automatic backup: {}", err)); backup_result } async fn backup_on_demand(&self) -> Result<(), String> { let keyspace_id = { let mut rng = rand::thread_rng(); self.keyspace_manager.get_uniform_random_keyspace(&mut rng) }; // Keep the RefStore snapshot aligned with the backup's actual backup_ts. // Other keyspaces remain available to concurrent workloads. let lock = self.keyspace_manager.get_keyspace_lock(keyspace_id); let _guard = lock.mutex_lock().await; self.set_dfs_worker_enabled(false).await?; self.backup_on_demand_with_retry(keyspace_id).await } async fn backup_on_demand_with_retry(&self, keyspace_id: u32) -> Result<(), String> { let mut last_error = None; for attempt in 0..ON_DEMAND_BACKUP_RETRY_TIMES { if attempt > 0 { // Reset the initial state before retrying after a partial failure. self.set_dfs_worker_enabled(false).await?; tokio::time::sleep(BACKUP_NAME_BOUNDARY).await; } let backup = async { self.set_dfs_worker_enabled(true).await?; self.wait_dfs_workers_ready().await?; self.br_client .create_lightweight_backup_async() .await .map_err(|err| err.to_string()) } .await; let cleanup = self.set_dfs_worker_enabled(false).await; let backup = match (backup, cleanup) { (Ok(response), Ok(())) => Ok(response), (Ok(_), Err(err)) | (Err(err), _) => Err(err), }; match backup { Ok(response) => { return self .record_on_demand_backup(keyspace_id, response.backup_name) .await; } Err(err) => { warn!("on-demand backup failed"; "keyspace" => keyspace_id, "attempt" => attempt + 1, "err" => %err); last_error = Some(err.to_string()); } } } Err(last_error.unwrap_or_else(|| "on-demand backup was not attempted".to_owned())) } async fn record_on_demand_backup( &self, keyspace_id: u32, backup_name: String, ) -> Result<(), String> { let backup_meta = get_cluster_backup_meta_async(self.dfs.as_ref(), backup_name.clone()).await; let mut keyspace_backup = self .keyspace_manager .backup_keyspace(keyspace_id, backup_meta.backup_ts); keyspace_backup.backup_name = Some(backup_name); self.keyspace_manager.add_backup(keyspace_backup); self.record_backup_result(&backup_meta); ON_DEMAND_BACKUP_COUNTER.fetch_add(1, Ordering::SeqCst); info!("on-demand backup success"; "keyspace" => keyspace_id, "backup_meta" => ?backup_meta); Ok(()) } fn record_backup_result(&self, backup_meta: &rfenginepb::ClusterBackupMeta) { BACKUP_COUNTER.fetch_add(1, Ordering::SeqCst); assert!(backup_meta.tolerated_err as usize <= self.backup_config.tolerate_err); if backup_meta.tolerated_err > 0 { BACKUP_TOLERATED_ERR_COUNTER .fetch_add(backup_meta.tolerated_err as usize, Ordering::Relaxed); } } async fn set_dfs_worker_enabled(&self, enabled: bool) -> Result<(), String> { let start_time = Instant::now(); let mut last_error = None; while let Some(remaining) = remaining_timeout(start_time, self.config.operation_timeout) { let stores = match get_all_stores_except_tiflash_async(self.pd_client.as_ref(), true).await { Ok(stores) if !stores.is_empty() => stores, Ok(_) => { last_error = Some("no TiKV stores found".to_owned()); tokio::time::sleep(DFS_WORKER_POLL_INTERVAL.min(remaining)).await; continue; } Err(err) => { last_error = Some(err.to_string()); tokio::time::sleep(DFS_WORKER_POLL_INTERVAL.min(remaining)).await; continue; } }; let mut error = None; for store in stores { let uri = self .security_mgr .build_uri(format!("{}/config", store.status_address)) .map_err(|err| err.to_string())?; let change = std::collections::HashMap::from([("rfengine.lightweight-backup", enabled)]); let request = hyper::Request::post(uri) .body(hyper::Body::from(serde_json::to_vec(&change).unwrap())) .map_err(|err| err.to_string())?; if let Err(err) = send_request_to_store( request, &store, &self.http_client, self.config.operation_timeout, ) .await { error = Some(format!("store {}: {err}", store.id)); } } if error.is_none() { info!("set DFS worker state"; "enabled" => enabled); return Ok(()); } last_error = error; tokio::time::sleep(DFS_WORKER_POLL_INTERVAL.min(remaining)).await; } Err(format!( "timed out setting DFS worker enabled={enabled}: {}", last_error.unwrap_or_else(|| "request was not sent".to_owned()) )) } async fn wait_dfs_workers_ready(&self) -> Result<(), String> { let start_time = Instant::now(); let mut last_status = None; while let Some(remaining) = remaining_timeout(start_time, self.config.operation_timeout) { let stores = match get_all_stores_except_tiflash_async(self.pd_client.as_ref(), true).await { Ok(stores) if !stores.is_empty() => stores, Ok(_) => { last_status = Some("no TiKV stores found".to_owned()); tokio::time::sleep(DFS_WORKER_POLL_INTERVAL.min(remaining)).await; continue; } Err(err) => { last_status = Some(err.to_string()); tokio::time::sleep(DFS_WORKER_POLL_INTERVAL.min(remaining)).await; continue; } }; let mut ready = true; let mut statuses = Vec::with_capacity(stores.len()); for store in stores { let status = async { let uri = self .security_mgr .build_uri(format!("{}/rfengine/dfs_worker", store.status_address)) .map_err(|err| err.to_string())?; let request = hyper::Request::get(uri) .body(hyper::Body::empty()) .map_err(|err| err.to_string())?; let (_, body) = send_request_to_store( request, &store, &self.http_client, self.config.operation_timeout, ) .await .map_err(|err| err.to_string())?; serde_json::from_slice::(&body).map_err(|err| err.to_string()) } .await; match status { Ok(status) => { ready &= status.enabled && status.running && status.healthy; statuses.push(format!("{}: {status:?}", store.id)); } Err(err) => { ready = false; statuses.push(format!("{}: {err}", store.id)); } } } if ready { info!("DFS workers are ready"; "statuses" => ?statuses); return Ok(()); } last_status = Some(format!("{statuses:?}")); tokio::time::sleep(DFS_WORKER_POLL_INTERVAL.min(remaining)).await; } Err(format!( "timed out waiting for DFS workers: {}", last_status.unwrap_or_else(|| "status was not queried".to_owned()) )) } async fn restart_automatic_backup(&mut self) -> Result<(), String> { self.set_dfs_worker_enabled(true).await?; self.wait_dfs_workers_ready().await?; // Separate the on-demand backup from the first backup produced by the // reconstructed worker. tokio::time::sleep(BACKUP_NAME_BOUNDARY).await; self.backup_worker = BackupWorker::new( self.backup_config.clone(), self.pd_client.clone(), self.config.periodic_interval, self.config.batch_interval, ); Ok(()) } // `Error::HttpError` is retryable even when `tolerate_err` > 0 because a // backup can overlap the restart of more than one store. fn is_backup_error_retryable(err: &Error) -> bool { match err { Error::MetaNotFound(_) | Error::HttpRequestError(_) | Error::RfengineDfsWorkerUnhealthy(_) => true, Error::BackupErrorOnStores(errs, _) => { errs.iter().all(|err| Self::is_backup_error_retryable(err)) } Error::SharedError(err) => Self::is_backup_error_retryable(err.inner()), _ => false, } } } pub(crate) fn check_br(online_snapshot_restore_ratio: f64) { let total_backup_count = BACKUP_COUNTER.load(Ordering::SeqCst); let total_on_demand_backup_count = ON_DEMAND_BACKUP_COUNTER.load(Ordering::SeqCst); let total_restore_count = RESTORE_COUNTER.load(Ordering::SeqCst); let online_snapshot_restore_count = ONLINE_SNAPSHOT_RESTORE_COUNTER.load(Ordering::SeqCst); assert!( // The total_backup_count is unstable and lower than expected. // TODO: investigate the reason. total_backup_count > 0, "backup count too small: {}", total_backup_count ); assert!( total_on_demand_backup_count > 0, "on-demand backup count too small: {}", total_on_demand_backup_count ); assert!( total_restore_count > 0, "restore count too small: {}", total_restore_count ); if online_snapshot_restore_ratio == 1.0 { assert!( online_snapshot_restore_count > 0, "online snapshot restore was not exercised when its ratio is 1" ); } } struct SourceRestoreGuard { source_keyspace: u32, active_sources: Arc>>, } impl SourceRestoreGuard { fn try_acquire(active_sources: Arc>>, source_keyspace: u32) -> Option { if !active_sources.lock().unwrap().insert(source_keyspace) { return None; } Some(Self { source_keyspace, active_sources, }) } } impl Drop for SourceRestoreGuard { fn drop(&mut self) { let removed = self .active_sources .lock() .unwrap() .remove(&self.source_keyspace); debug_assert!(removed); } } pub(crate) fn spawn_restore_keyspace( pd_client: Arc, pd_control: Option, mut client: ClusterKeyspaceClient, restore_config: RestoreConfig, keyspace_manager: KeyspaceManager, dfs: Arc, _enable_oss_chaos: bool, object_cache: Option, limiter: Option>, enable_cmek: bool, timeout: Duration, tikv_worker_endpoints: Vec, security_manager: Arc, online_snapshot_restore_ratio: f64, active_source_restores: Arc>>, ) -> JoinHandle<()> { let dfs = dfs.clone(); std::thread::spawn(move || { assert!((0.0..=1.0).contains(&online_snapshot_restore_ratio)); let mut rng = rand::thread_rng(); let runtime = tokio::runtime::Builder::new_multi_thread() .worker_threads(4) .enable_all() .thread_name("restore-keyspace") .build() .unwrap(); let reporter = Arc::new(DummyStepReporter::default()); let br_client = NativeBrSvcClient::new( pd_client.get_cluster_id().unwrap(), tikv_worker_endpoints, security_manager, ) .unwrap(); let start_time = Instant::now(); sleep(Duration::from_secs(3)); 'next_restore: while let Some(remaining) = remaining_timeout(start_time, timeout) { let backup = match keyspace_manager.get_random_backup(&mut rng) { Some(backup) => backup, None => { sleep(Duration::from_millis(500).min(remaining)); continue; } }; let source_keyspace = backup.keyspace_id; let online_snapshot = source_keyspace > 0 && rng.gen_bool(online_snapshot_restore_ratio); let branching = online_snapshot || (source_keyspace > 0 && rng.gen_bool(0.5)); // Only in-place restores mutate the source and need serialization. // Online snapshot branching validates each source Region again at // its file_refs observation and retries detected conflicts. let source_restore_guard = if source_keyspace > 0 && !branching { let Some(guard) = SourceRestoreGuard::try_acquire( active_source_restores.clone(), source_keyspace, ) else { sleep(Duration::from_millis(100).min(remaining)); continue; }; Some(guard) } else { None }; let (target_keyspace, target_keyspace_lock_guard) = if branching { let (new_keyspace, lock_guard) = runtime.block_on(create_new_keyspace( &pd_client, &keyspace_manager, dfs.as_ref(), 0, 0.0, enable_cmek, true, )); assert!(lock_guard.is_some()); info!( "branching restore, create new keyspace"; "online_snapshot" => online_snapshot, "source_keyspace" => source_keyspace, "target_keyspace" => new_keyspace, ); (new_keyspace, lock_guard) } else { (source_keyspace, None) }; let backup_name = backup.backup_name().to_string(); let backup_meta = (!online_snapshot) .then(|| get_cluster_backup_meta(dfs.as_ref(), backup_name.clone())); let tag = if online_snapshot { format!("{}->{}[online-snapshot]", source_keyspace, target_keyspace) } else { format!("{}->{}[{}]", source_keyspace, target_keyspace, backup_name) }; let mut config = restore_config.clone(); if let Some(backup_meta) = &backup_meta { config.timeout_fetch_wal = if config.tolerate_err > backup_meta.tolerated_err as usize { FETCH_WAL_TIMEOUT_TOLERATE_ERR } else { FETCH_WAL_TIMEOUT }; } let tolerated_err_limit = if online_snapshot { config .online_snapshot_max_unaligned_voter_peers .max(config.tolerate_err) } else { config.tolerate_err }; let online_restore_id = online_snapshot .then(|| ONLINE_SNAPSHOT_RESTORE_TASK_ID.fetch_add(1, Ordering::Relaxed)); let mut restored_keyspace = None; 'inner_retry: for retry in 0..RFENGINE_HTTP_ERROR_RETRY_TIMES { if remaining_timeout(start_time, timeout).is_none() { break 'next_restore; } let _guard = target_keyspace_lock_guard.is_none().then(|| { let lock = keyspace_manager.get_keyspace_lock(target_keyspace); runtime.block_on(lock.mutex_lock()) }); // The process of destroying ranges is not determined. So skip verifying the // destroying ranges. A failed online attempt may have restored a // partial target, so only verify its initial empty state. if !online_snapshot || retry == 0 { runtime .block_on(client.verify_keyspace_and_skip_destroyed_ranges(target_keyspace)) .unwrap_or_else(|err| { panic!( "{} verify_keyspace_with_ref_store (before restore): {:?}", tag, err ) }); } if online_snapshot { let restore_id = online_restore_id.unwrap(); match do_restore_keyspace_from_online_snapshot( &br_client, &runtime, &keyspace_manager, source_keyspace, target_keyspace, restore_id, // The restore task runs asynchronously in tikv-worker. Once it is // submitted, wait with a fresh timeout so the random workload does not // exit while the task is still mutating the target keyspace. timeout, ) { Ok((progress, expected)) => { restored_keyspace = Some(restore_keyspace::RestoredKeyspace { keyspace_id: source_keyspace, target_keyspace_id: target_keyspace, ts: progress.restore_ts, restore_bytes: progress.restore_bytes, resolved_ts: progress.restore_ts.into(), ..Default::default() }); keyspace_manager.restore_keyspace(&tag, expected, target_keyspace); } Err(err) => { warn!( "online snapshot attempt failed, retry from the beginning"; "tag" => &tag, "restore_id" => restore_id, "retry" => retry, "err" => ?err, ); let Some(remaining) = remaining_timeout(start_time, timeout) else { panic!( "{} online snapshot restore {} attempt failed after workload timeout: {:?}", tag, restore_id, err ); }; sleep(Duration::from_millis(250).min(remaining)); continue 'inner_retry; } } } else { restored_keyspace = match do_restore_keyspace( pd_client.clone(), pd_control.clone(), &runtime, config.clone(), source_keyspace, target_keyspace, &backup_name, backup.backup_ts, reporter.clone(), object_cache.clone(), limiter.clone(), ) { Ok(res) => Some(res), Err(Error::BackupEmptyForKeyspace(_)) => { // Empty backup will happen on newly created keyspace. Retry. warn!("{} backup is empty, retry", tag); continue 'next_restore; } Err(Error::WalChunkIntegrityError(msg)) => { // TODO: retry on next backup. warn!( "{} backup is broken: WAL chunk integrity error ({}), retry", tag, msg ); // TODO: assert!(enable_oss_chaos); BROKEN_BACKUP_COUNTER.fetch_add(1, Ordering::SeqCst); continue 'next_restore; } Err(Error::RejectReplayRestore) => { warn!("{} backup is rejected for replay restore, retry", tag); continue 'next_restore; } Err(Error::RfengineHttpRequestError(err)) => { // We would still meet `RfengineHttpError` even thought `tolerate_err` > // 0, as there are chances that restoration // is performed across the restart of // more than one store. warn!("{} meet RfengineHttpError, retry", tag; "err" => ?err); continue 'inner_retry; } Err(Error::SpitRegionsError(err)) => { // `split_regions_with_retry` is not stable itself. Retry. // TODO: improve the stability of `split_regions_with_retry`, and remove // this retry. warn!("{} meet SpitRegionsError, retry", tag; "err" => ?err); continue 'inner_retry; } Err(err) => panic!("{} restore failed: {:?}", tag, err), }; keyspace_manager.restore_keyspace(&tag, backup.clone(), target_keyspace); } // To find data corruption early, and generate read workload as well. // The retry should not be necessary. // TODO: Remove the retry after verification issue is addressed. let verify_res = try_wait_result( || { let verify_res = runtime.block_on( client.verify_keyspace_and_skip_destroyed_ranges(target_keyspace), ); if verify_res.is_err() { warn!( "{} verify_keyspace_with_ref_store failed (after restore): {:?}", tag, verify_res ); } verify_res }, 10, ); if verify_res.is_err() { info!("{} backup_meta: {:?}", tag, backup_meta); panic!( "{} verify_keyspace_with_ref_store (after restore): {:?}", tag, verify_res ); } break; } drop(target_keyspace_lock_guard); let restored_keyspace = restored_keyspace.unwrap_or_else(|| { panic!("{} restore retry limit exceeded", tag); }); info!("{} restore keyspace success", tag; "res" => ?restored_keyspace); RESTORE_COUNTER.fetch_add(1, Ordering::Relaxed); if online_snapshot { ONLINE_SNAPSHOT_RESTORE_COUNTER.fetch_add(1, Ordering::Relaxed); } assert!(restored_keyspace.tolerated_err <= tolerated_err_limit); if restored_keyspace.tolerated_err > 0 { RESTORE_TOLERATED_ERR_COUNTER .fetch_add(restored_keyspace.tolerated_err, Ordering::Relaxed); } drop(source_restore_guard); let Some(remaining) = remaining_timeout(start_time, timeout) else { break; }; sleep(Duration::from_secs(rng.gen_range(0..10)).min(remaining)); } info!("restore keyspace thread exit"); }) } #[derive(Default)] struct DummyStepReporter {} impl ReportRestoreStepTrait for DummyStepReporter { fn report_step(&self, _step: RestoreStep) {} }