// Copyright 2026 TiKV Project Authors. Licensed under Apache-2.0. use std::{sync::Arc, time::Duration}; use api_version::ApiV2; use fail::FailScenario; use kvengine::dfs::{DFSConfig, Dfs, FileType, new_dfs_from_config}; use native_br::{ archive, backup::{self, BackupType}, common::now, metrics::NATIVE_BR_RESTORE_ERROR, restore::RestoreConfig, restore_keyspace, step, }; use pd_client::PdClient; use rstest::rstest; use security::SecurityConfig; use tempfile::TempDir; use test_cloud_server::{ ServerCluster, ServerClusterBuilder, TikvWorkerOptions, client::{ ClusterClient, ClusterClientOptions, CommitAction, MutateOptions, RequestOptions, TxnWriteMethod, }, move_region_to_voter_stores, oss::{ObjectStorageService, prepare_dfs}, }; use test_pd_client::PdWrapper; use tikv::config::TikvConfig; use tikv_util::{ config::{ReadableDuration, ReadableSize}, info, warn, }; use tokio::runtime::Runtime; use txn_types::TimeStamp; use crate::{ TestEncryptionMode, alloc_node_id_vec, create_keyspace_meta, native_backup::{ DummyStepReporter, gen_keyspace_key, generate_backup_name, i_to_val, wait_all_dfs_workers_healthy, }, new_security_config, }; const BASIC_DATA_COUNT: usize = 10; const BASIC_DATA_LEN: usize = 80; const LONG_DATA_LEN: usize = 8 * 1024; const IMPORT_DATA_LEN: usize = 82; const FINAL_DATA_LEN: usize = 90; const KEYSPACE_COUNT: usize = 3; const DEFAULT_LOOP_COUNT: usize = 1; const NODES_COUNT: usize = 3; const BACKUP_DAYS: usize = 3; struct TestArchiveOptions { loop_count: usize, encryption_mode: TestEncryptionMode, expire_store_backup: bool, with_failpoints: bool, nodes_count: usize, with_missing_backup_meta_day: bool, } impl Default for TestArchiveOptions { fn default() -> Self { Self { loop_count: DEFAULT_LOOP_COUNT, encryption_mode: TestEncryptionMode::Disabled, expire_store_backup: false, with_failpoints: false, nodes_count: NODES_COUNT, with_missing_backup_meta_day: false, } } } #[rstest] #[case(TestEncryptionMode::Disabled)] #[case(TestEncryptionMode::Legacy)] #[case(TestEncryptionMode::Metadata)] fn test_restore_keyspace_with_archive(#[case] encryption_mode: TestEncryptionMode) { test_util::init_log_for_test(); test_restore_archived_keyspace_opt(TestArchiveOptions { encryption_mode, with_missing_backup_meta_day: matches!(encryption_mode, TestEncryptionMode::Disabled), ..Default::default() }); } #[test] fn test_restore_archived_keyspace_with_expiration() { test_restore_archived_keyspace_opt(TestArchiveOptions { expire_store_backup: true, nodes_count: 3, with_failpoints: true, ..Default::default() }); } #[test] fn test_archive_collects_files_from_store_without_keyspace_zero() { test_util::init_log_for_test(); const ARCHIVE_KEYSPACE_ID: u32 = 1; let (_temp_dir, mut oss, dfs_config) = prepare_dfs("test_archive_keyspace_without_keyspace_zero_"); let security_config = SecurityConfig::default(); let pd_wrapper = PdWrapper::new_test(1, &security_config, None); let mut cluster = ServerClusterBuilder::new(alloc_node_id_vec(6), |_, conf: &mut TikvConfig| { conf.dfs = dfs_config.clone(); conf.rocksdb.writecf.write_buffer_size = ReadableSize::kb(1); conf.coprocessor.region_split_size = ReadableSize::kb(128); conf.coprocessor.region_bucket_size = ReadableSize::kb(64); conf.rfengine.target_file_size = ReadableSize::kb(512); conf.rfengine.lightweight_backup = true; conf.rfengine.wal_chunk_target_file_size = ReadableSize::kb(32); }) .pd(pd_wrapper) .build(); cluster.start_tikv_workers(alloc_node_id_vec(1), TikvWorkerOptions::default()); cluster.wait_region_replicated(&[], 3); let runtime = Runtime::new().unwrap(); let pd_client = cluster.get_pd_client(); pd_client.disable_default_operator(); create_keyspace_meta(&cluster, ARCHIVE_KEYSPACE_ID, &runtime); let mut client = cluster.new_client(); client.split_keyspaces([0, ARCHIVE_KEYSPACE_ID]); cluster.wait_pd_region_min_count(4); client.clear_region_cache(); let mut stores = cluster.get_stores(); stores.sort_unstable(); assert_eq!(stores.len(), 6, "unexpected stores: {:?}", stores); let keyspace_zero_stores = stores[..3].to_vec(); let archive_keyspace_stores = stores[3..].to_vec(); let move_keyspace_regions = |client: &mut ClusterClient, keyspace_id: u32, target_stores: &[u64]| { let (start, end) = ApiV2::get_keyspace_range_by_id(keyspace_id); let regions = client.scan_regions(&start, &end).unwrap(); assert!( !regions.is_empty(), "keyspace {} should have at least one region", keyspace_id ); for region in regions { move_region_to_voter_stores( &cluster, &pd_client, region.id, target_stores, target_stores[0], ); client.clear_region_cache(); } }; move_keyspace_regions(&mut client, 0, &keyspace_zero_stores); move_keyspace_regions(&mut client, ARCHIVE_KEYSPACE_ID, &archive_keyspace_stores); let archive_keyspace_key = gen_keyspace_key(ARCHIVE_KEYSPACE_ID)(0); let archive_keyspace_region_id = client.get_region_by_key(&archive_keyspace_key).id; // Put data after moving peers so the backup records a store that has data // for the archived keyspace but no keyspace 0 shard. client.put_kv( 0..BASIC_DATA_COUNT, gen_keyspace_key(0), i_to_val(BASIC_DATA_LEN), ); client.put_kv( 0..BASIC_DATA_COUNT, gen_keyspace_key(ARCHIVE_KEYSPACE_ID), i_to_val(LONG_DATA_LEN), ); cluster.flush_memtable(archive_keyspace_region_id).unwrap(); cluster.wait_for_memtable_flushed(archive_keyspace_region_id, Duration::from_secs(10)); let backup_config = backup::BackupConfig { dfs: dfs_config.clone(), backup_delay: ReadableDuration::secs(1), skip_keyspace_meta: true, ..Default::default() }; let backup_ts = client.get_ts().into_inner(); let (_, mut backup_meta) = backup::backup_cluster_with_ts( backup_config, BackupType::Lightweight, generate_backup_name(), cluster.get_pd_client().as_ref(), backup_ts, None, ) .expect("backup::backup_cluster"); for node_id in cluster.get_nodes() { runtime.block_on(cluster.get_rfengine(node_id).upload_wal_chunk()); } // Model the production shape behind the archive bug: a store can have data // for a non-zero keyspace without a keyspace 0 entry in backup metadata. // The test cluster may keep zero-sized entries for empty keyspace 0 // regions, so normalize the metadata before exercising archive discovery. for store_id in &archive_keyspace_stores { let store_meta = backup_meta .stores .iter_mut() .find(|store| store.store_id == *store_id) .unwrap_or_else(|| panic!("store {} missing from backup meta", store_id)); assert!( store_meta.keyspace_size.contains_key(&ARCHIVE_KEYSPACE_ID), "test setup requires store {} to contain keyspace {}: {:?}", store_id, ARCHIVE_KEYSPACE_ID, store_meta.keyspace_size ); store_meta.keyspace_size.remove(&0); } let s3fs = new_dfs_from_config(dfs_config.clone()); let work_dir = TempDir::new().unwrap(); let restore_config = RestoreConfig { security: security_config, lower_memory: true, ..Default::default() }; let expected_cluster = restore_keyspace::BackupCluster::new( &backup_meta, work_dir.path().join("expected"), pd_client.clone(), s3fs.clone(), restore_config.clone(), ARCHIVE_KEYSPACE_ID, ARCHIVE_KEYSPACE_ID, backup_ts, false, None, false, None, ) .unwrap(); assert!( expected_cluster.shards_count() > 0, "test setup should load keyspace {} shards", ARCHIVE_KEYSPACE_ID ); drop(expected_cluster); let archive_cluster = restore_keyspace::BackupCluster::new( &backup_meta, work_dir.path().join("archive"), pd_client, s3fs, restore_config, 0, 0, backup_ts, true, None, false, None, ) .unwrap(); let archive_has_keyspace_shard = archive_cluster .get_shard(archive_keyspace_region_id) .is_some(); drop(archive_cluster); cluster.stop(); oss.shutdown(); assert!( archive_has_keyspace_shard, "archive missed keyspace {} region {} from store without keyspace 0", ARCHIVE_KEYSPACE_ID, archive_keyspace_region_id ); } #[rstest] #[case::lower_memory(true)] fn test_archive_retries_after_reject_replay_restore(#[case] lower_memory: bool) { test_util::init_log_for_test(); const KEYSPACE_ID: u32 = 1; const SPLIT_KEY: usize = 100; const LEFT_KEY_A: usize = 0; const RIGHT_KEY_C: usize = 200; const RFENGINE_TARGET_FILE_SIZE: usize = 16 * 1024; let (_temp_dir, mut oss, dfs_config) = prepare_dfs("test_archive_retries_after_reject_replay_restore_"); let s3fs = new_dfs_from_config(dfs_config.clone()); let reporter = Arc::new(DummyStepReporter::default()); let runtime = Runtime::new().unwrap(); let mut cluster = ServerCluster::new(alloc_node_id_vec(3), |_, conf: &mut TikvConfig| { conf.dfs = dfs_config.clone(); conf.rfengine.target_file_size = ReadableSize::kb(16); conf.rfengine.wal_chunk_target_file_size = ReadableSize::kb(128); conf.rfengine.lightweight_backup = true; conf.memory.enable_heap_profiling = false; }); cluster.wait_region_replicated(&[], 3); let pd_client = cluster.get_pd_client(); let mut client = cluster.new_client(); client.split_keyspace(KEYSPACE_ID); let i_to_key = gen_keyspace_key(KEYSPACE_ID); let backup_config = backup::BackupConfig { dfs: dfs_config.clone(), backup_delay: ReadableDuration::secs(1), skip_keyspace_meta: true, ..Default::default() }; let backup_time = chrono::NaiveDateTime::from_timestamp_millis(client.get_ts().physical() as i64).unwrap(); oss.set_as_if_date(Some(backup_time.and_utc())); let archive_backup_name = |days| { let date_time = backup_time .checked_add_days(chrono::Days::new(days)) .unwrap(); backup::IncrementalBackupFile::from_backup_ts( TimeStamp::compose(date_time.timestamp_millis() as u64, 0).into_inner(), ) .into_name() }; // Create backup A with only a small value in the left region. Splitting // after this backup makes the following restore leave a replayable restore // entry that is not yet covered by a fresh snapshot. client.put_kv( LEFT_KEY_A..(LEFT_KEY_A + 1), &i_to_key, i_to_val(BASIC_DATA_LEN), ); let archive_seed_key = gen_keyspace_key(0); client.put_kv(0..1, &archive_seed_key, i_to_val(BASIC_DATA_LEN)); let archive_seed_region_id = client.get_region_by_key(&archive_seed_key(0)).id; cluster.flush_memtable(archive_seed_region_id).unwrap(); cluster.wait_for_memtable_flushed(archive_seed_region_id, Duration::from_secs(10)); let ref_store_a = client.dump_ref_store(); wait_all_dfs_workers_healthy(&cluster, "dfs workers healthy before backup A"); let backup_a_name = archive_backup_name(0); let backup_a_ts = client.get_ts().into_inner(); let (_, backup_a_meta) = backup::backup_cluster_with_ts( backup_config.clone(), BackupType::Lightweight, backup_a_name.clone(), pd_client.as_ref(), backup_a_ts, None, ) .expect("backup::backup_cluster for backup A"); client.try_split(&i_to_key(SPLIT_KEY), 5).unwrap(); cluster.wait_pd_region_min_count(4); restore_keyspace::restore_keyspace( KEYSPACE_ID, KEYSPACE_ID, &backup_a_name, None, s3fs.clone(), RestoreConfig::default_for_test(), pd_client.clone(), None, &runtime, None, None, reporter.clone(), cluster.create_object_cache_randomly(), cluster.create_restore_limiter_randomly(runtime.handle().clone()), ) .unwrap(); client .verify_data_with_given_ref_store(&ref_store_a, None, &RequestOptions::default()) .unwrap(); // Backup B captures the restore entry before a fresh snapshot covers it. client.put_kv( LEFT_KEY_A + 1..(LEFT_KEY_A + 2), &i_to_key, i_to_val(BASIC_DATA_LEN), ); wait_all_dfs_workers_healthy(&cluster, "dfs workers healthy before backup B"); let backup_b_name = archive_backup_name(1); backup::backup_cluster_with_ts( backup_config.clone(), BackupType::Lightweight, backup_b_name, pd_client.as_ref(), client.get_ts().into_inner(), None, ) .expect("backup::backup_cluster for backup B"); // Backup C adds a fresh snapshot so archive can skip B and finish A using // C as the next valid backup candidate. for key in RIGHT_KEY_C..(RIGHT_KEY_C + 16) { client.put_kv( std::iter::once(key), &i_to_key, i_to_val(RFENGINE_TARGET_FILE_SIZE), ); } wait_all_dfs_workers_healthy(&cluster, "dfs workers healthy before backup C"); let backup_c_name = archive_backup_name(2); backup::backup_cluster_with_ts( backup_config, BackupType::Lightweight, backup_c_name, pd_client.as_ref(), client.get_ts().into_inner(), None, ) .expect("backup::backup_cluster for backup C"); for node_id in cluster.get_nodes() { runtime.block_on(cluster.get_rfengine(node_id).upload_wal_chunk()); } let archive_config = archive::ArchiveConfig { dfs: dfs_config, max_archive_file_size: 32 * 1024, ..Default::default() }; let backup_a_date = backup_time.date(); let backup_b_date = backup_time .checked_add_days(chrono::Days::new(1)) .unwrap() .date(); let reject_replay_restore = NATIVE_BR_RESTORE_ERROR.with_label_values(&["reject_replay_restore"]); reject_replay_restore.reset(); archive::archive_cluster_backup( archive_config, pd_client, s3fs.clone(), backup_a_date, backup_b_date, ) .unwrap(); let archive_reader = archive::ArchiveReader::new(s3fs.clone(), &backup_a_date).unwrap(); let archived_meta = archive_reader.read_meta_file().unwrap(); assert_eq!( archived_meta.backup_ts, backup_a_ts, "backup A should be archived after retrying with backup C" ); assert_eq!( archived_meta, backup_a_meta, "archived backup A data differs" ); // A concurrently persisted rfengine snapshot may already cover the restore // entry when backup B is evaluated, so replaying it and emitting this // metric is timing-dependent. if reject_replay_restore.get() == 0 { warn!("archive did not encounter RejectReplayRestore while evaluating backup B"); } else { assert!( archive::ArchiveReader::new(s3fs.clone(), &backup_b_date).is_err(), "rejected backup B should not produce an archive" ); } let regular_backup_a_meta_key = backup::backup_file_full_path(s3fs.get_prefix(), backup_a_name.clone(), None); runtime .block_on(s3fs.delete_object(regular_backup_a_meta_key.clone(), regular_backup_a_meta_key)) .unwrap(); restore_keyspace::restore_keyspace( KEYSPACE_ID, KEYSPACE_ID, &backup_a_name, None, s3fs.clone(), RestoreConfig { lower_memory, ..RestoreConfig::default_for_test() }, cluster.get_pd_client(), None, &runtime, None, None, reporter, cluster.create_object_cache_randomly(), cluster.create_restore_limiter_randomly(runtime.handle().clone()), ) .unwrap(); let verified_count = client .verify_data_with_given_ref_store(&ref_store_a, None, &RequestOptions::default()) .unwrap(); assert_eq!(verified_count, (2, 0)); cluster.stop(); oss.shutdown(); } fn test_restore_archived_keyspace_opt(options: TestArchiveOptions) { // Expiration exercises its own failure matrix. For the encryption matrix, // pair each mode with one data shape instead of running the full Cartesian // product; all modes and shapes still execute the complete archive flow. let cases = if options.expire_store_backup { // The ordinary archive matrix covers the other data shapes. Expiration // keeps the multi-region case that also covers unresolved normal writes. vec![(1, 1, 30, options.loop_count)] } else { match options.encryption_mode { TestEncryptionMode::Disabled => vec![(0, 1, 1, 1)], TestEncryptionMode::Legacy => vec![(1, 1, 100, options.loop_count)], TestEncryptionMode::Metadata => vec![(2, 2, 1, 1)], } }; let max_data_count = cases.iter().map(|x| x.2).max().unwrap(); let _scenario = options.with_failpoints.then(FailScenario::setup); let (_temp_dir, mut oss, dfs_config) = prepare_dfs("test_restore_keyspace_"); let security_config = new_security_config(); let pd_wrapper = PdWrapper::new_test(1, &security_config, None); let mut cluster = ServerClusterBuilder::new( alloc_node_id_vec(options.nodes_count), |_, conf: &mut TikvConfig| { conf.dfs = dfs_config.clone(); // Set small mem-table size to make data reach L1 and generate over bound // shards. conf.rocksdb.writecf.write_buffer_size = ReadableSize::kb(1); conf.coprocessor.region_split_size = ReadableSize::kb(128); // kv_opts.base_size = 8kb conf.coprocessor.region_bucket_size = ReadableSize::kb(64); conf.rfengine.target_file_size = ReadableSize::kb(512); conf.rfengine.lightweight_backup = true; conf.rfengine.wal_chunk_target_file_size = ReadableSize::kb(32); conf.rfengine.force_rotate_interval = Some(ReadableDuration::hours(10)); conf.security = security_config.clone(); }, ) .pd(pd_wrapper) .build(); cluster.start_tikv_workers(alloc_node_id_vec(1), TikvWorkerOptions::default()); cluster.wait_region_replicated(&[], 3); let pd_client = cluster.get_pd_client(); let mut client = cluster.new_client(); let runtime = Runtime::new().unwrap(); pd_client.set_tso(TimeStamp::compose( chrono::Utc::now().timestamp_millis() as u64, 0, )); // Split keyspaces. for keyspace_id in 0..KEYSPACE_COUNT as u32 { if keyspace_id != 0 { create_keyspace_meta(&cluster, keyspace_id, &runtime); } if options.encryption_mode.is_enabled() { pd_client .set_keyspace_encryption( keyspace_id, options.encryption_mode.keyspace_encryption_config(), ) .unwrap(); } } client.split_keyspaces(0..KEYSPACE_COUNT as u32); cluster.wait_pd_region_count(KEYSPACE_COUNT + 2); for keyspace_id in 0..KEYSPACE_COUNT as u32 { let i_to_key = gen_keyspace_key(keyspace_id); // To make secondary keys in another region. if max_data_count >= 10 { client.try_split(&i_to_key(max_data_count / 3), 5).unwrap(); client.try_split(&i_to_key(max_data_count / 2), 5).unwrap(); } // Import basic data. client.put_kv(0..BASIC_DATA_COUNT, &i_to_key, i_to_val(BASIC_DATA_LEN)); } let _enter = runtime.enter(); // Run cases. for &(case_idx, keyspace_id, data_count, loop_count) in &cases { step!("case group: {case_idx}"); pd_client.enable_default_operator(); // Remove learners by peer count check. for loop_idx in 0..loop_count { test_restore_archived_keyspace_impl( case_idx, loop_idx, &mut cluster, &dfs_config, security_config.clone(), &mut oss, keyspace_id, data_count, options.expire_store_backup, options.with_failpoints, options.with_missing_backup_meta_day && case_idx == 0 && loop_idx == 0, &runtime, ); } } cluster.stop(); oss.shutdown(); } fn test_restore_archived_keyspace_impl( case_idx: usize, loop_idx: usize, cluster: &mut ServerCluster, dfs_config: &DFSConfig, security_config: SecurityConfig, oss: &mut ObjectStorageService, keyspace_id: u32, data_count: usize, expire_store_backup: bool, with_failpoints: bool, with_missing_backup_meta_day: bool, runtime: &Runtime, ) { step!("case: {}:{}", case_idx, loop_idx); let mut client = cluster.new_client_opt(ClusterClientOptions { txn_file_max_chunk_size: Some(1024), ..Default::default() }); let i_to_key = gen_keyspace_key(keyspace_id); let backup_config = backup::BackupConfig { dfs: dfs_config.clone(), backup_delay: ReadableDuration::secs(1), security: security_config.clone(), skip_keyspace_meta: true, ..Default::default() }; let s3fs = new_dfs_from_config(dfs_config.clone()); let reporter = Arc::new(DummyStepReporter::default()); let pd_client = cluster.get_pd_client(); let mut archive_config = archive::ArchiveConfig { dfs: dfs_config.clone(), max_archive_file_size: 1024 * 32, // This legacy knob is ignored. Keep it explicit so the archive gap // below verifies automatic lookahead behavior with the deprecated value. skip_no_meta_days: 0, dry_run: false, security: security_config.clone(), ..Default::default() }; let _archive_temp_dir = archive_config.check_data_dir(); // Import, backup and archive every day. let begin_archive_date = chrono::NaiveDateTime::from_timestamp_millis(client.get_ts().physical() as i64).unwrap(); let mut date_time = begin_archive_date; let mut backup_ts_list = Vec::with_capacity(BACKUP_DAYS); let mut ref_store_list = Vec::with_capacity(BACKUP_DAYS); let mut last_ref_store = client.dump_ref_store(); for idx in 0..BACKUP_DAYS { // Import data. let write_method = if (case_idx + loop_idx + idx).is_multiple_of(2) { TxnWriteMethod::Normal } else { TxnWriteMethod::FileBased }; // Keep FileBased in the expiration matrix, but commit its secondary keys. // Normal writes still cover restoring expired snapshots with unresolved // secondary locks, while ordinary archive cases cover unresolved // FileBased transactions without the forced clock jump. let commit_action = if with_failpoints && write_method == TxnWriteMethod::FileBased { CommitAction::SyncCommit } else { CommitAction::AsyncCommitSecondaryKeys(Duration::MAX) }; client .try_put_kv( 0..data_count, &i_to_key, i_to_val(IMPORT_DATA_LEN + idx), MutateOptions { commit_action, write_method, ..Default::default() }, ) .unwrap(); step!("write done on {}", date_time.date()); client.verify_data_with_ref_store(); assert!( client .verify_data_with_given_ref_store(&last_ref_store, None, &RequestOptions::default()) .is_err(), "case: {}:{}:{}", case_idx, loop_idx, date_time.date(), ); last_ref_store = client.dump_ref_store(); step!("verify ok on {}", date_time.date()); // Perform snapshot backup. let backup_ts = client.get_ts().into_inner(); backup_ts_list.push(backup_ts); ref_store_list.push(client.dump_ref_store()); let (_, backup_meta) = backup::backup_cluster_with_ts( backup_config.clone(), BackupType::Lightweight, String::default(), cluster.get_pd_client().as_ref(), backup_ts, None, ) .expect("backup::backup_cluster"); info!("backup_cluster result: {:?}", backup_meta); for node_id in cluster.get_nodes() { // This wait can be done when archiving. futures::executor::block_on(cluster.get_rfengine(node_id).upload_wal_chunk()); } // Archive backup. archive::archive_cluster_backup( archive_config.clone(), cluster.get_pd_client(), s3fs.clone(), begin_archive_date.date(), date_time.date(), ) .unwrap(); step!("archive done on {}", date_time.date()); // The second backup is two calendar days after the first one. The // archive must skip the missing meta day and still archive the first // backup against this next real backup. if with_missing_backup_meta_day && idx == 1 { let first_backup_date = chrono::NaiveDateTime::from_timestamp_millis( TimeStamp::from(backup_ts_list[0]).physical() as i64, ) .unwrap() .date(); let archive_reader = archive::ArchiveReader::new(s3fs.clone(), &first_backup_date).unwrap(); let archived_meta = archive_reader.read_meta_file().unwrap(); assert_eq!(archived_meta.backup_ts, backup_ts_list[0]); } if idx > 0 && expire_store_backup { // Expire snapshots after archiving. Keep WAL chunks so replay can // still build continuous chunk ranges during restore. for node_id in cluster.get_nodes() { let store_id = cluster.get_store_id(node_id); let snapshot_prefix = format!( "{}/{}/store_backup/{:016x}/snapshots", dfs_config.s3_bucket, dfs_config.prefix, store_id ); let expired = oss .expire_path(&snapshot_prefix, Duration::from_secs(2 * 24 * 60 * 60)) .unwrap(); info!("expired snapshot objects after archiving"; "case_idx" => case_idx, "loop_idx" => loop_idx, "backup_idx" => idx, "expired" => expired, "day" => %date_time.date()); } } date_time = date_time.checked_add_days(chrono::Days::new(1)).unwrap(); pd_client.set_tso(TimeStamp::compose(date_time.timestamp_millis() as u64, 0)); oss.set_as_if_date(Some(date_time.and_utc())); if with_failpoints { fail::cfg( "rfengine::now_unix_millis", &format!("return({})", date_time.timestamp_millis()), ) .unwrap(); } if with_missing_backup_meta_day && idx == 0 { // Leave one calendar day without backup meta. let skipped_date = date_time.date(); date_time = date_time.checked_add_days(chrono::Days::new(1)).unwrap(); pd_client.set_tso(TimeStamp::compose(date_time.timestamp_millis() as u64, 0)); step!("skip backup meta on {}", skipped_date); } } // Delete old backup. { let snapshot_backup_ts = backup_ts_list[0]; let snapshot_backup_date = chrono::NaiveDateTime::from_timestamp_millis( TimeStamp::from(snapshot_backup_ts).physical() as i64, ) .unwrap() .date(); let archive_reader = archive::ArchiveReader::new(s3fs.clone(), &snapshot_backup_date).unwrap(); let last_archived_snapshot_backup_date = chrono::NaiveDateTime::from_timestamp_millis( TimeStamp::from(backup_ts_list[BACKUP_DAYS - 2]).physical() as i64, ) .unwrap() .date(); let last_archive_reader = archive::ArchiveReader::new(s3fs.clone(), &last_archived_snapshot_backup_date).unwrap(); let mut old_wal_rlog_keys = Vec::new(); let mut old_wal_chunks_len = 0; for node_id in cluster.get_nodes() { let store_id = cluster.get_store_id(node_id); let store_meta = archive_reader.get_store_wal_rlog_meta(store_id).unwrap(); let last_archived_store_meta = last_archive_reader .get_store_wal_rlog_meta(store_id) .unwrap(); for epoch_id in store_meta.snapshot_epoch_id..=last_archived_store_meta.snapshot_epoch_id { if epoch_id < last_archived_store_meta.snapshot_epoch_id { let old_meta_key = rfengine::snapshot_store_meta_key(store_id, epoch_id); old_wal_rlog_keys.push(format!("{}/{}", s3fs.get_prefix(), old_meta_key)); let old_rlog_key = rfengine::snapshot_rlog_key(store_id, epoch_id); old_wal_rlog_keys.push(format!("{}/{}", s3fs.get_prefix(), old_rlog_key)); } let old_wal_chunk_prefix = rfengine::wal_chunk_file_prefix(store_id, epoch_id); let old_wal_chunk_keys = Dfs::list_objects(s3fs.as_ref(), "", Some(&old_wal_chunk_prefix), None) .map(|(objects, _)| { objects .iter() .map(|object| object.key.clone()) .collect::>() }) .unwrap(); old_wal_chunks_len += old_wal_chunk_keys.len(); old_wal_rlog_keys.extend(old_wal_chunk_keys) } } let old_file_ids = archive_reader.get_file_ids(); let mut deleted_old_wal_rlog_cnt = 0; s3fs.get_runtime().block_on(async { for idx in 0..BACKUP_DAYS - 1 { let backup_ts = backup_ts_list[idx]; let backup_key = backup::backup_file_full_path( s3fs.get_prefix(), String::default(), Some(backup_ts), ); s3fs.delete_object(backup_key.clone(), backup_ts.to_string()) .await .unwrap(); } for wal_rlog_key in old_wal_rlog_keys { if s3fs .exist(wal_rlog_key.clone(), wal_rlog_key.clone()) .await .unwrap() { s3fs.delete_object(wal_rlog_key.clone(), wal_rlog_key) .await .unwrap(); deleted_old_wal_rlog_cnt += 1; } } for &file_id in &old_file_ids { let mut errs = vec![]; // There is no file type in archive, so we try all types. for ftype in [FileType::Sst, FileType::TxnChunk] { match s3fs .delete_object( s3fs.file_key(file_id, ftype), format!("{}.{}", file_id, ftype.suffix()), ) .await { Ok(_) => { errs.clear(); break; } Err(e) => errs.push(e), } } assert!(errs.is_empty(), "errs: {:?}", errs); } }); let old_rlog_len = (deleted_old_wal_rlog_cnt - old_wal_chunks_len) / 2; step!( "delete old backup meta, snap rlog {}, wal chunks {} and table files {:?} done. case: {}:{}", old_rlog_len, old_wal_chunks_len, old_file_ids, case_idx, loop_idx, ); } let object_cache = cluster.create_object_cache_randomly(); let limiter = cluster.create_restore_limiter_randomly(runtime.handle().clone()); for idx in 0..BACKUP_DAYS - 1 { // Restore keyspace. { let snapshot_backup_ts = backup_ts_list[idx]; let snapshot_backup_name = backup::IncrementalBackupFile::from_backup_ts(snapshot_backup_ts).into_name(); let restore_config = RestoreConfig { security: security_config.clone(), ..Default::default() }; restore_keyspace::restore_keyspace( keyspace_id, keyspace_id, &snapshot_backup_name, None, s3fs.clone(), restore_config, cluster.get_pd_client(), None, runtime, None, None, reporter.clone(), object_cache.clone(), limiter.clone(), ) .unwrap(); step!("restore done. case: {}:{}:{}", case_idx, loop_idx, idx); } // Verify restored data. { let verify_options: Vec<(Option<(&[u8], &[u8])>, RequestOptions)> = vec![(None, RequestOptions::default())]; let mut verify_res = vec![]; for (range, opt) in verify_options { verify_res.push( client .verify_data_with_given_ref_store(&ref_store_list[idx], range, &opt) .unwrap(), ); } step!( "verify restore data done, kv count: {:?}, case: {}:{}:{}", verify_res, case_idx, loop_idx, idx ); } } // Write more data to verify the sequences. for id in 0..KEYSPACE_COUNT as u32 { let count = if id == keyspace_id { data_count } else { BASIC_DATA_COUNT }; client.put_kv(0..count, gen_keyspace_key(id), i_to_val(FINAL_DATA_LEN)); } let verify_res = client.verify_data_with_ref_store(); step!("verify more writes done, verify result: {:?}", verify_res); }