From 87333f7b24151718cb739823898e8718d160ebdc Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 31 Aug 2026 07:37:34 +0800 Subject: [PATCH 1/4] test(e2e): exercise cluster volume fault proxy (#6919) Co-authored-by: heihutu --- .../src/cluster_multidrive_pool_test.rs | 30 +++++++++++++++++-- docs/testing/e2e-suite-inventory.md | 4 +-- 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/crates/e2e_test/src/cluster_multidrive_pool_test.rs b/crates/e2e_test/src/cluster_multidrive_pool_test.rs index 22b456ffa..411336754 100644 --- a/crates/e2e_test/src/cluster_multidrive_pool_test.rs +++ b/crates/e2e_test/src/cluster_multidrive_pool_test.rs @@ -27,8 +27,10 @@ //! Readiness is established by the harness's `start()` handshake (TCP reachability //! plus an S3 `ListBuckets` poll) — there are no fixed sleeps. //! -//! Out of scope for this block (tracked separately): network fault injection -//! (toxiproxy / socket proxy) and 5GiB large-object budgets. +//! The volume-proxy smoke below also proves that the socket-level fault proxy +//! can be installed before startup without changing the client-facing node URL. +//! A full lock-plane partition matrix and 5GiB large-object budget remain +//! tracked separately. use crate::common::{ClusterTopology, RustFSTestClusterEnvironment}; @@ -125,3 +127,27 @@ async fn cluster_two_pool_smoke() -> TestResult { put_get_roundtrip(&cluster, "twopool/object", &payload).await?; Ok(()) } + +/// A real cluster smoke for the volume FaultProxy wiring. The proxy target is +/// not listening yet when it is created; cluster startup must still converge +/// once the target node starts, and peer disk/RPC traffic must traverse it. +#[tokio::test] +async fn cluster_volume_fault_proxy_pass_smoke() -> TestResult { + crate::common::init_logging(); + + let mut cluster = RustFSTestClusterEnvironment::with_topology(ClusterTopology::single_pool_multidrive(2, 2)).await?; + let proxy = cluster.start_volume_proxy_for_node(0).await?; + let proxied = proxy.local_addr().to_string(); + assert!(cluster.rustfs_volumes_arg().contains(&proxied)); + + let result: TestResult = async { + cluster.start().await?; + cluster.create_test_bucket(BUCKET).await?; + let payload = vec![0x6Du8; 256 * 1024]; + put_get_roundtrip(&cluster, "volume-proxy/object", &payload).await + } + .await; + + proxy.shutdown().await; + result +} diff --git a/docs/testing/e2e-suite-inventory.md b/docs/testing/e2e-suite-inventory.md index eef3ea1c7..c7e4938ef 100644 --- a/docs/testing/e2e-suite-inventory.md +++ b/docs/testing/e2e-suite-inventory.md @@ -30,7 +30,7 @@ | chaos | 2 | | | checksum_upload_test | 7 | | | cluster_concurrency_test | 3 | 🌙 | -| cluster_multidrive_pool_test | 3 | 🌙 | +| cluster_multidrive_pool_test | 4 | 🌙 | | common | 17 | | | compression_test | 6 | ✅ | | connection_cap_test | 2 | | @@ -103,4 +103,4 @@ | tls_hot_reload_test | 1 | ✅ | | version_id_regression_test | 10 | ✅ | -**Total listed: 620 tests across 86 modules · PR smoke: 165 tests / 36 modules · merge/main full: 495 tests / 77 modules · nightly replication: 56 tests · nightly cluster faults: 30 tests / 7 modules · nightly protocols: 16 tests** · updated 2026-08-31. +**Total listed: 621 tests across 86 modules · PR smoke: 165 tests / 36 modules · merge/main full: 495 tests / 77 modules · nightly replication: 56 tests · nightly cluster faults: 31 tests / 7 modules · nightly protocols: 16 tests** · updated 2026-08-31. From 042a0c3014e3febbeabd0399fbca1f2c93cea271 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Mon, 31 Aug 2026 07:53:02 +0800 Subject: [PATCH 2/4] docs: register persisted XML compatibility cleanup (#6918) docs: register persisted XML compatibility --- crates/ecstore/src/bucket/utils.rs | 1 + docs/architecture/compat-cleanup-register.md | 1 + 2 files changed, 2 insertions(+) diff --git a/crates/ecstore/src/bucket/utils.rs b/crates/ecstore/src/bucket/utils.rs index d24ef95b5..f659468bb 100644 --- a/crates/ecstore/src/bucket/utils.rs +++ b/crates/ecstore/src/bucket/utils.rs @@ -73,6 +73,7 @@ pub fn check_valid_bucket_name_strict(bucket_name: &str) -> Result<()> { check_bucket_name_common(bucket_name, true) } +// RUSTFS_COMPAT_TODO(s3gate-metadata-xml): the s3s codec reads persisted XML during migration. Remove after every supported writer uses the gateway codec and every retained metadata object and backup archive is verified or rewritten. pub fn deserialize(input: &[u8]) -> xml::DeResult where T: for<'xml> xml::Deserialize<'xml>, diff --git a/docs/architecture/compat-cleanup-register.md b/docs/architecture/compat-cleanup-register.md index e54dac90c..8b56a5939 100644 --- a/docs/architecture/compat-cleanup-register.md +++ b/docs/architecture/compat-cleanup-register.md @@ -13,6 +13,7 @@ for later deletion. ## Open Items - `backlog-2102` rc.2/rc.3 empty scanner usage floor recovery: old DeleteBucket cleanup could synthesize an empty incomplete v2 usage primary/backup before leadership added an epoch, while newer scanners require a durable authoritative baseline identity. New scanners recognize only that exact serialized empty-fence shape, preserve its epoch through a CAS-protected recovery marker, and rebuild namespace coverage without treating zero usage as authoritative. Remove this recovery path and marker after rc.2 and rc.3 are no longer supported direct-upgrade sources. +- `s3gate-metadata-xml` persisted bucket XML migration: mixed-version site-replication peers, retained `.metadata.bin` objects, and backup archives can all carry XML written by the s3s codec, so the gateway migration must keep the legacy codec available until every stored form has crossed a verified rewrite boundary. Remove the legacy s3s parser and serializer only after the minimum supported direct-upgrade release reads and writes every persisted XML configuration family through the gateway codec, the four-way D1-D5 gate has remained clean for one full support window, every supported mixed-version site-replication topology has completed its writer upgrade, and migration tooling has verified or rewritten every retained bucket metadata object and restorable backup archive. - `rustfs-6339` legacy bucket policy ID casing: earlier RustFS releases persisted the top-level policy identifier as "ID", while current writes use the S3-compatible "Id" spelling. Readers accept both spellings so retained bucket metadata remains usable after upgrade. Remove the legacy alias after migration tooling has rewritten every retained bucket policy using "ID". - `table-publication-fence-v1` table publication fencing: nodes that predate table and table-bucket publication fences can mutate live files while a new node is publishing a catalog pointer. New nodes retain exact object guards until the operator confirms that every serving node uses the new fences. Fleet confirmation also requires non-overlapping active warehouse prefixes and lifecycle workers that exclude table buckets. Remove the exact live-file fallback and the fleet-confirmation gate after the minimum supported RustFS release acquires table fences for registered-table mutations and table-bucket fences for unresolved-prefix mutations. - `table-catalog-strong-snapshot-v1` durable strong catalog snapshot compatibility: version 1 writes continue during mixed-version rollout until operators confirm that every serving node reads version 2, and version 1 table/view identifier collisions remain available only for cleanup. Remove version 1 writes and collision cleanup after the minimum supported RustFS release reads version 2 and every retained durable strong snapshot is collision-free and has been upgraded to version 2. From e6234d3714ef101509dbfd21f482a54804355730 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Mon, 31 Aug 2026 08:03:07 +0800 Subject: [PATCH 3/4] test(ecstore): pin default bucket config bytes (#6920) --- crates/ecstore/src/store/mod.rs | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index c07a25fab..53c5346e4 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -1205,6 +1205,23 @@ mod tests { use serial_test::serial; use tempfile::TempDir; + #[test] + fn g_d2_008_default_versioning_config_keeps_persisted_bytes() { + let bytes = crate::bucket::utils::serialize::(&ENABLED_VERSIONING_CONFIG) + .expect("the default Versioning configuration must serialize"); + assert_eq!(bytes, b"Enabled"); + } + + #[test] + fn g_d2_009_default_object_lock_config_keeps_persisted_bytes() { + let bytes = crate::bucket::utils::serialize::(&ENABLED_OBJECT_LOCK_CONFIG) + .expect("the default Object Lock configuration must serialize"); + assert_eq!( + bytes, + b"Enabled" + ); + } + #[tokio::test] async fn test_get_disk_infos() { let disks = vec![None, None]; // Empty disks for testing From 8ecd8f2520fadedf63123651681963b26565665a Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 31 Aug 2026 08:19:54 +0800 Subject: [PATCH 4/4] fix(scanner): preserve cache cycle during usage recovery (#6921) Co-authored-by: heihutu --- crates/scanner/src/scanner.rs | 73 ++++- crates/scanner/src/scanner/cycle_state.rs | 55 +++- crates/scanner/src/scanner/leadership.rs | 38 ++- crates/scanner/src/scanner/tests.rs | 356 ++++++++++++++++++++-- 4 files changed, 468 insertions(+), 54 deletions(-) diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 66d563362..001ae6f8a 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -751,15 +751,15 @@ fn prepare_cycle_for_usage_floor_bootstrap( cycle_info: &mut CurrentCycle, usage_floor: PersistedUsageFloor, startup: PersistedUsageFloorStartup, -) -> (bool, bool) { +) -> (bool, ScannerCycleResetPolicy) { match startup { - PersistedUsageFloorStartup::Authoritative => (false, false), + PersistedUsageFloorStartup::Authoritative => (false, ScannerCycleResetPolicy::None), PersistedUsageFloorStartup::Missing => { // Cycle progress without its corresponding usage floor cannot // prove namespace coverage. Restart from cycle zero while keeping // the separately fenced leader epoch monotonic. *cycle_info = CurrentCycle::default(); - (true, true) + (true, ScannerCycleResetPolicy::ResetAll) } PersistedUsageFloorStartup::BootstrapPending => { // An unfenced marker may have been written before an upgrade's old @@ -768,14 +768,25 @@ fn prepare_cycle_for_usage_floor_bootstrap( if usage_floor.leader_epoch == 0 { *cycle_info = CurrentCycle::default(); } - (true, usage_floor.leader_epoch == 0) + ( + true, + if usage_floor.leader_epoch == 0 { + ScannerCycleResetPolicy::ResetAll + } else { + ScannerCycleResetPolicy::None + }, + ) } PersistedUsageFloorStartup::RecoveredLegacyEmptyFence => { // The legacy empty fence proves only its leader epoch, not - // namespace coverage. Restart coverage from zero while retaining - // that epoch as the lower bound for the next leadership claim. - *cycle_info = CurrentCycle::default(); - (true, true) + // namespace coverage. Clear coverage while retaining the durable + // cycle number so surviving caches cannot force a regression. + let next = cycle_info.next; + *cycle_info = CurrentCycle { + next, + ..Default::default() + }; + (true, ScannerCycleResetPolicy::ResetCoveragePreservingNext) } } } @@ -1296,7 +1307,15 @@ where LockLost: Future, { let fence_ctx = ctx.child_token(); - let claim = claim_scanner_leadership(&fence_ctx, storeapi, cycle_info, cycle_revision, leader_epoch, false, false); + let claim = claim_scanner_leadership( + &fence_ctx, + storeapi, + cycle_info, + cycle_revision, + leader_epoch, + false, + ScannerCycleResetPolicy::None, + ); tokio::pin!(claim); tokio::pin!(lock_lost); tokio::select! { @@ -1792,8 +1811,18 @@ async fn run_data_scanner_cycle_with_budget( mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await; return ScannerCycleOutcome::Failed; } - match scanner_cycle_pre_commit_outcome(scan_cycle_result.required_cycle_floor(), &usage_persist_outcome) { + let required_cycle_floor = scan_cycle_result.required_cycle_floor(); + let pre_commit_outcome = scanner_cycle_pre_commit_outcome(required_cycle_floor, &usage_persist_outcome); + update_scanner_cache_cycle_recovery_status( + cycle_info.current, + leader_epoch, + required_cycle_floor, + pre_commit_outcome, + scan_cycle_result.status == ScannerCycleStatus::Complete, + ); + match pre_commit_outcome { Some(ScannerCyclePreCommitOutcome::RecoverCacheCycle(required_cycle)) => { + record_scanner_cache_cycle_recovery_attempt(); warn!( target: "rustfs::scanner", event = EVENT_SCANNER_CYCLE_STATE, @@ -2312,7 +2341,7 @@ async fn run_data_scanner_with_maintenance_state( return Ok(()); } }; - let (allow_usage_floor_bootstrap_pending, reset_usage_floor_bootstrap_cycle_on_conflict) = + let (allow_usage_floor_bootstrap_pending, usage_floor_cycle_reset_policy) = prepare_cycle_for_usage_floor_bootstrap(&mut cycle_info, usage_floor, usage_floor_startup); apply_persisted_usage_floor(&mut cycle_info, &mut leader_epoch, usage_floor); match usage_floor_startup { @@ -2374,7 +2403,7 @@ async fn run_data_scanner_with_maintenance_state( &mut cycle_revision, &mut leader_epoch, allow_usage_floor_bootstrap_pending, - reset_usage_floor_bootstrap_cycle_on_conflict, + usage_floor_cycle_reset_policy, ), guard.lock_lost_notified(), ) @@ -2902,6 +2931,26 @@ fn scanner_cycle_pre_commit_outcome( } } +fn update_scanner_cache_cycle_recovery_status( + requested_cycle: u64, + leader_epoch: u64, + required_cycle_floor: Option, + pre_commit_outcome: Option, + cache_scope_complete: bool, +) { + match (required_cycle_floor, pre_commit_outcome) { + (Some(required_cycle), _) => { + record_scanner_cache_cycle_ahead(requested_cycle, required_cycle, leader_epoch); + } + (None, Some(ScannerCyclePreCommitOutcome::Deferred(_))) => { + // A deferred scan may not have covered the cache that established + // the existing floor, so it cannot prove recovery is complete. + } + (None, _) if cache_scope_complete => clear_scanner_cache_cycle_ahead(), + (None, _) => {} + } +} + fn scanner_cycle_completion_outcome( scan_status: ScannerCycleStatus, usage_persist_outcome: DataUsagePersistOutcome, diff --git a/crates/scanner/src/scanner/cycle_state.rs b/crates/scanner/src/scanner/cycle_state.rs index f34de9317..f8a11c6f3 100644 --- a/crates/scanner/src/scanner/cycle_state.rs +++ b/crates/scanner/src/scanner/cycle_state.rs @@ -25,6 +25,7 @@ const METRIC_SCANNER_CYCLE_RECOVERY_REQUIRED: &str = "rustfs_scanner_cycle_recov const METRIC_SCANNER_CYCLE_RECOVERY_RETRY_COUNT: &str = "rustfs_scanner_cycle_recovery_retry_count"; const USAGE_FLOOR_LOAD_FAILED: &str = "usage_floor_load_failed"; const LEGACY_EMPTY_USAGE_FLOOR_RECOVERY: &str = "legacy_empty_usage_floor"; +const CACHE_CYCLE_AHEAD: &str = "cache_cycle_ahead"; #[derive(Clone, Debug, Default, Serialize)] pub struct ScannerCycleRecoveryStatus { @@ -40,6 +41,7 @@ pub struct ScannerCycleRecoveryStatus { pub first_detected_at_unix_secs: Option, pub last_attempt_at_unix_secs: Option, pub retry_count: u64, + /// Maximum automatic retries, or zero when the recovery is unbounded. pub max_retries: u32, /// Whether the scanner may retry this state automatically. pub retryable: bool, @@ -72,6 +74,7 @@ fn set_scanner_cycle_recovery_status(status: ScannerCycleRecoveryStatus) { | "cleanup-pending" | "usage_floor_load_failed" | "usage_floor_recovery_pending" + | "cache_cycle_ahead" ) { 1.0 } else { @@ -113,6 +116,51 @@ pub(super) fn clear_scanner_usage_floor_failure() { } } +pub(super) fn record_scanner_cache_cycle_ahead(requested_cycle: u64, required_cycle: u64, leader_epoch: u64) { + let previous = scanner_cycle_recovery_status(); + let same_floor = previous.classification.as_deref() == Some(CACHE_CYCLE_AHEAD) + && previous.generation == Some(required_cycle) + && previous.leader_epoch == Some(leader_epoch); + let now = unix_now_secs(); + let (first_detected_at_unix_secs, retry_count) = if same_floor { + (previous.first_detected_at_unix_secs.or(Some(now)), previous.retry_count) + } else { + (Some(now), 0) + }; + set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus { + path: DATA_USAGE_BLOOM_NAME_PATH.clone(), + state: CACHE_CYCLE_AHEAD.to_string(), + classification: Some(CACHE_CYCLE_AHEAD.to_string()), + generation: Some(required_cycle), + leader_epoch: Some(leader_epoch), + first_detected_at_unix_secs, + last_attempt_at_unix_secs: Some(now), + retry_count, + max_retries: 0, + retryable: true, + reason: Some(format!( + "persisted scanner cache cycle {required_cycle} is ahead of requested cycle {requested_cycle}" + )), + ..Default::default() + }); +} + +pub(super) fn record_scanner_cache_cycle_recovery_attempt() { + let mut status = scanner_cycle_recovery_status(); + if status.classification.as_deref() != Some(CACHE_CYCLE_AHEAD) { + return; + } + status.retry_count = status.retry_count.saturating_add(1); + status.last_attempt_at_unix_secs = Some(unix_now_secs()); + set_scanner_cycle_recovery_status(status); +} + +pub(super) fn clear_scanner_cache_cycle_ahead() { + if scanner_cycle_recovery_status().classification.as_deref() == Some(CACHE_CYCLE_AHEAD) { + set_scanner_cycle_recovery_status(recovery_status("healthy", None, false)); + } +} + pub(super) fn record_legacy_empty_usage_floor_recovery_pending(leader_epoch: u64) { let previous = scanner_cycle_recovery_status(); let same_recovery = previous.classification.as_deref() == Some(LEGACY_EMPTY_USAGE_FLOOR_RECOVERY) @@ -147,9 +195,12 @@ pub(super) fn clear_legacy_empty_usage_floor_recovery_status() { pub(super) fn record_scanner_cycle_recovery_retry(attempt: u32) -> bool { let mut status = scanner_cycle_recovery_status(); - status.retry_count = u64::from(attempt); + if status.classification.as_deref() == Some(CACHE_CYCLE_AHEAD) { + return true; + } + status.retry_count = status.retry_count.max(u64::from(attempt)); status.last_attempt_at_unix_secs = Some(unix_now_secs()); - if attempt >= MAX_SCANNER_CYCLE_RECOVERY_RETRIES { + if status.max_retries != 0 && status.retry_count >= u64::from(status.max_retries) { status.state = "paused".to_string(); status.retryable = false; status.reason = Some("scanner cycle recovery retry budget reached; sparse backend probes continue".to_string()); diff --git a/crates/scanner/src/scanner/leadership.rs b/crates/scanner/src/scanner/leadership.rs index a99687f33..e728e7420 100644 --- a/crates/scanner/src/scanner/leadership.rs +++ b/crates/scanner/src/scanner/leadership.rs @@ -21,6 +21,29 @@ pub(super) enum ScannerLeadershipClaimReconcile { Unchanged, } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum ScannerCycleResetPolicy { + None, + ResetAll, + ResetCoveragePreservingNext, +} + +impl ScannerCycleResetPolicy { + fn apply(self, cycle_info: &mut CurrentCycle, attempted_next: u64) { + match self { + Self::None => {} + Self::ResetAll => *cycle_info = CurrentCycle::default(), + Self::ResetCoveragePreservingNext => { + let next = cycle_info.next.max(attempted_next); + *cycle_info = CurrentCycle { + next, + ..Default::default() + }; + } + } + } +} + pub(super) async fn reconcile_scanner_leadership_claim( storeapi: Arc, attempted: &[u8], @@ -329,7 +352,7 @@ pub(super) async fn claim_scanner_leadership( revision: &mut DataUsageCacheRevision, persisted_epoch: &mut u64, allow_bootstrap_pending: bool, - reset_bootstrap_cycle_on_conflict: bool, + cycle_reset_policy: ScannerCycleResetPolicy, ) -> bool { for retry in 0..=SCANNER_PERSIST_CAS_RETRIES { if ctx.is_cancelled() { @@ -347,6 +370,7 @@ pub(super) async fn claim_scanner_leadership( ); return false; }; + let attempted_next = cycle_info.next; let data = match encode_scanner_cycle_state(cycle_info, claimed_epoch) { Ok(data) => data, Err(err) => { @@ -459,9 +483,7 @@ pub(super) async fn claim_scanner_leadership( .await; } Ok(ScannerLeadershipClaimReconcile::Changed) if retry < SCANNER_PERSIST_CAS_RETRIES => { - if reset_bootstrap_cycle_on_conflict { - *cycle_info = CurrentCycle::default(); - } + cycle_reset_policy.apply(cycle_info, attempted_next); continue; } Ok(ScannerLeadershipClaimReconcile::Changed | ScannerLeadershipClaimReconcile::Unchanged) => { @@ -517,17 +539,13 @@ pub(super) async fn claim_scanner_leadership( Ok(ScannerLeadershipClaimReconcile::Changed) if retry < SCANNER_PERSIST_CAS_RETRIES && !ctx.is_cancelled() => { - if reset_bootstrap_cycle_on_conflict { - *cycle_info = CurrentCycle::default(); - } + cycle_reset_policy.apply(cycle_info, attempted_next); continue; } Ok(ScannerLeadershipClaimReconcile::Unchanged) if precondition_failed && retry < SCANNER_PERSIST_CAS_RETRIES && !ctx.is_cancelled() => { - if reset_bootstrap_cycle_on_conflict { - *cycle_info = CurrentCycle::default(); - } + cycle_reset_policy.apply(cycle_info, attempted_next); continue; } Ok(ScannerLeadershipClaimReconcile::Changed | ScannerLeadershipClaimReconcile::Unchanged) => { diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index b0ea9c202..60dacc00f 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -16,10 +16,11 @@ use super::heal_info::{classify_background_heal_read_error, decode_background_he use super::*; use crate::EcstoreResult; use crate::{ - DATA_USAGE_BLOOM_RECOVERY_PATH, Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, - ScannerGetObjectReader as GetObjectReader, ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, - ScannerPutObjReader as PutObjReader, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests, - init_local_disks_with_instance_ctx, + DATA_USAGE_BLOOM_RECOVERY_PATH, DATA_USAGE_CACHE_KEY_FORMAT, DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT, + DataUsageCachePrepareOutcome, DataUsageCacheSource, DataUsageEntry, DataUsageScanPlanDigest, Endpoint, EndpointServerPools, + Endpoints, InstanceContext, PoolEndpoints, ScannerGetObjectReader as GetObjectReader, ScannerObjectInfo as ObjectInfo, + ScannerObjectOptions as ObjectOptions, ScannerPutObjReader as PutObjReader, init_bucket_metadata_sys_for_scanner_tests, + init_ecstore_config_for_scanner_tests, init_local_disks_with_instance_ctx, }; use serial_test::serial; use std::collections::{HashMap, HashSet}; @@ -1991,7 +1992,7 @@ fn rc3_legacy_empty_usage_fence(epoch: Option) -> Vec { } #[tokio::test] -async fn scanner_usage_floor_recovers_rc3_empty_fences_and_restarts_from_zero() { +async fn scanner_usage_floor_recovers_rc3_empty_fences_and_preserves_cycle_number() { let store = Arc::new(MemoryConfigStore::default()); let primary_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()); let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()); @@ -2034,16 +2035,17 @@ async fn scanner_usage_floor_recovers_rc3_empty_fences_and_restarts_from_zero() assert_eq!(restart_floor.leader_epoch, 7); assert_eq!(restart_state, PersistedUsageFloorStartup::RecoveredLegacyEmptyFence); let mut cycle = CurrentCycle { - current: 41, - next: 42, - ..Default::default() + current: 17_117, + next: 17_118, + cycle_completed: vec![Utc::now()], + started: Utc::now(), }; assert_eq!( prepare_cycle_for_usage_floor_bootstrap(&mut cycle, restart_floor, restart_state), - (true, true) + (true, ScannerCycleResetPolicy::ResetCoveragePreservingNext) ); assert_eq!(cycle.current, 0); - assert_eq!(cycle.next, 0); + assert_eq!(cycle.next, 17_118); assert!(cycle.cycle_completed.is_empty()); let mut revision = DataUsageCacheRevision::Missing; @@ -2056,11 +2058,70 @@ async fn scanner_usage_floor_recovers_rc3_empty_fences_and_restarts_from_zero() &mut revision, &mut leader_epoch, true, - true, + ScannerCycleResetPolicy::ResetCoveragePreservingNext, ) .await ); assert_eq!(leader_epoch, 8); + assert_eq!(cycle.next, 17_118); + assert_eq!(cycle.current, 0); + assert!(cycle.cycle_completed.is_empty()); + + let source = DataUsageCacheSource::new(0, 0); + let scan_plan_digest = DataUsageScanPlanDigest([7; 32]); + for (cache_path, name) in [ + (DATA_USAGE_CACHE_NAME.to_string(), DATA_USAGE_ROOT), + (format!("photos/{DATA_USAGE_CACHE_NAME}"), "photos"), + ] { + let mut historical = DataUsageCache::default(); + historical.info.name = name.to_string(); + historical.info.next_cycle = 17_118; + historical.info.leader_epoch = 7; + historical.info.source = Some(source); + historical.info.scan_plan_digest = Some(scan_plan_digest); + historical.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT; + historical.info.snapshot_complete = true; + historical.replace(name, "", DataUsageEntry::default()); + historical + .save(store.clone(), &cache_path) + .await + .expect("historical scanner cache should persist through the storage path"); + + let mut recovered = DataUsageCache::default(); + let revisions = recovered + .load_with_revisions(store.clone(), &cache_path) + .await + .expect("historical scanner cache should reload with CAS revisions"); + assert_eq!(recovered.info.name, name); + assert_eq!(recovered.info.next_cycle, 17_118); + assert_eq!(recovered.info.leader_epoch, 7); + assert!(!recovered.cache.is_empty()); + + assert_eq!( + recovered.prepare_for_scan(name, cycle.next, leader_epoch, source, scan_plan_digest, true), + DataUsageCachePrepareOutcome::Reset, + "recovered cache should reset without a cycle regression: {cache_path}" + ); + assert_eq!(recovered.info.next_cycle, 17_118); + assert_eq!(recovered.info.leader_epoch, 8); + assert!(!recovered.info.snapshot_complete); + assert!(recovered.cache.is_empty()); + + recovered + .save_with_revisions(store.clone(), &cache_path, &revisions) + .await + .expect("reset scanner cache should persist with its loaded revisions"); + let mut persisted_reset = DataUsageCache::default(); + persisted_reset + .load(store.clone(), &cache_path) + .await + .expect("persisted reset scanner cache should reload"); + assert_eq!(persisted_reset.info.name, name); + assert_eq!(persisted_reset.info.next_cycle, 17_118); + assert_eq!(persisted_reset.info.leader_epoch, 8); + assert!(!persisted_reset.info.snapshot_complete); + assert!(persisted_reset.cache.is_empty()); + } complete_legacy_empty_usage_floor_recovery(store.clone(), leader_epoch) .await .expect("leadership claim should retire the recovery marker"); @@ -2253,7 +2314,7 @@ async fn scanner_usage_floor_recovery_reconciles_marker_delete_post_commit_error let mut cycle = CurrentCycle::default(); let mut revision = DataUsageCacheRevision::Missing; let mut leader_epoch = floor.leader_epoch; - let (allow_pending, reset_on_conflict) = prepare_cycle_for_usage_floor_bootstrap(&mut cycle, floor, state); + let (allow_pending, cycle_reset_policy) = prepare_cycle_for_usage_floor_bootstrap(&mut cycle, floor, state); assert!( claim_scanner_leadership( &CancellationToken::new(), @@ -2262,7 +2323,7 @@ async fn scanner_usage_floor_recovery_reconciles_marker_delete_post_commit_error &mut revision, &mut leader_epoch, allow_pending, - reset_on_conflict, + cycle_reset_policy, ) .await ); @@ -2494,6 +2555,53 @@ fn scanner_usage_floor_recovery_stays_retryable_until_claim_cleanup() { assert_eq!(scanner_cycle_recovery_status().state, "healthy"); } +#[test] +#[serial] +fn scanner_cache_cycle_ahead_is_visible_until_a_later_scan_clears_it() { + record_scanner_cache_cycle_ahead(0, 17_118, 8); + let pending = scanner_cycle_recovery_status(); + assert_eq!(pending.state, "cache_cycle_ahead"); + assert_eq!(pending.classification.as_deref(), Some("cache_cycle_ahead")); + assert_eq!(pending.generation, Some(17_118)); + assert_eq!(pending.leader_epoch, Some(8)); + assert!(pending.retryable); + assert_eq!(pending.max_retries, 0); + assert_eq!( + pending.reason.as_deref(), + Some("persisted scanner cache cycle 17118 is ahead of requested cycle 0") + ); + let first_detected = pending.first_detected_at_unix_secs; + + record_scanner_cache_cycle_ahead(0, 17_118, 8); + let observed_again = scanner_cycle_recovery_status(); + assert_eq!(observed_again.retry_count, 0); + assert_eq!(observed_again.first_detected_at_unix_secs, first_detected); + + assert!(record_scanner_cycle_recovery_retry(4)); + assert_eq!(scanner_cycle_recovery_status().retry_count, 0); + + record_scanner_cache_cycle_recovery_attempt(); + record_scanner_cache_cycle_recovery_attempt(); + let retried = scanner_cycle_recovery_status(); + assert_eq!(retried.retry_count, 2); + assert!(retried.retryable); + + update_scanner_cache_cycle_recovery_status( + 0, + 8, + None, + Some(ScannerCyclePreCommitOutcome::Deferred(ScannerCycleDeferReason::DataMovement)), + false, + ); + assert_eq!(scanner_cycle_recovery_status().classification.as_deref(), Some("cache_cycle_ahead")); + + update_scanner_cache_cycle_recovery_status(17_118, 8, None, None, false); + assert_eq!(scanner_cycle_recovery_status().classification.as_deref(), Some("cache_cycle_ahead")); + + update_scanner_cache_cycle_recovery_status(17_118, 8, None, None, true); + assert_eq!(scanner_cycle_recovery_status().state, "healthy"); +} + #[tokio::test] #[serial] async fn scanner_usage_floor_failure_clears_stale_leader_liveness() { @@ -2946,7 +3054,7 @@ fn missing_usage_floor_discards_unfenced_cycle_progress() { assert_eq!( prepare_cycle_for_usage_floor_bootstrap(&mut cycle, PersistedUsageFloor::default(), PersistedUsageFloorStartup::Missing,), - (true, true) + (true, ScannerCycleResetPolicy::ResetAll) ); assert_eq!(cycle.next, 0); assert_eq!(cycle.current, 0); @@ -2959,7 +3067,7 @@ fn missing_usage_floor_discards_unfenced_cycle_progress() { PersistedUsageFloor::default(), PersistedUsageFloorStartup::BootstrapPending, ), - (true, true) + (true, ScannerCycleResetPolicy::ResetAll) ); assert_eq!(cycle.next, 0); } @@ -2980,7 +3088,7 @@ fn fenced_usage_bootstrap_retains_partial_cycle_progress() { }, PersistedUsageFloorStartup::BootstrapPending, ), - (true, false) + (true, ScannerCycleResetPolicy::None) ); assert_eq!(cycle.next, 12); @@ -2993,7 +3101,7 @@ fn fenced_usage_bootstrap_retains_partial_cycle_progress() { }, PersistedUsageFloorStartup::Authoritative, ), - (false, false) + (false, ScannerCycleResetPolicy::None) ); assert_eq!(cycle.next, 12); } @@ -3024,7 +3132,7 @@ async fn missing_usage_floor_rebuilds_persisted_cycle_before_leadership_claim() .await .expect("stably missing usage floor should admit a bootstrap marker"); assert_eq!(startup, PersistedUsageFloorStartup::Missing); - let (allow_bootstrap_pending, reset_bootstrap_cycle_on_conflict) = + let (allow_bootstrap_pending, cycle_reset_policy) = prepare_cycle_for_usage_floor_bootstrap(&mut cycle_info, usage_floor, startup); apply_persisted_usage_floor(&mut cycle_info, &mut persisted_epoch, usage_floor); initialize_usage_baseline_bootstrap(store.clone()) @@ -3039,7 +3147,7 @@ async fn missing_usage_floor_rebuilds_persisted_cycle_before_leadership_claim() &mut cycle_revision, &mut persisted_epoch, allow_bootstrap_pending, - reset_bootstrap_cycle_on_conflict, + cycle_reset_policy, ) .await ); @@ -3402,7 +3510,18 @@ async fn test_leadership_claim_preserves_usage_epoch_floor_across_old_epoch_conf ); let mut persisted_epoch = 8; - assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await); + assert!( + claim_scanner_leadership( + &ctx, + store.clone(), + &mut cycle, + &mut revision, + &mut persisted_epoch, + false, + ScannerCycleResetPolicy::None, + ) + .await + ); let state = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH) .await @@ -3440,7 +3559,18 @@ async fn unfenced_usage_bootstrap_discards_old_epoch_conflict_progress() { ); let mut persisted_epoch = 1; - assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, true, true).await); + assert!( + claim_scanner_leadership( + &ctx, + store.clone(), + &mut cycle, + &mut revision, + &mut persisted_epoch, + true, + ScannerCycleResetPolicy::ResetAll, + ) + .await + ); let state = read_config(store, &DATA_USAGE_BLOOM_NAME_PATH) .await @@ -3450,6 +3580,84 @@ async fn unfenced_usage_bootstrap_discards_old_epoch_conflict_progress() { assert_eq!(claimed_epoch, 2); } +#[tokio::test] +async fn recovered_usage_bootstrap_claim_conflicts_preserve_the_highest_cycle_number() { + for winner_next in [42_u64, 20_000] { + let store = Arc::new(MemoryConfigStore::default()); + let ctx = CancellationToken::new(); + let mut revision = DataUsageCacheRevision::Missing; + let mut cycle = CurrentCycle { + next: 12, + ..Default::default() + }; + assert!(persist_scanner_cycle_state(&ctx, store.clone(), &mut cycle, &mut revision, 1).await); + seed_usage_snapshot_for_leadership_claim(&store).await; + + cycle = CurrentCycle { + current: 17_117, + next: 17_118, + cycle_completed: vec![Utc::now()], + started: Utc::now(), + }; + let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()); + let winner = CurrentCycle { + current: winner_next.saturating_sub(1), + next: winner_next, + cycle_completed: vec![Utc::now()], + started: Utc::now(), + }; + store + .interleaving_puts + .lock() + .await + .insert(key, (2, encode_scanner_cycle_state(&winner, 7).expect("conflict winner should encode"))); + + let mut persisted_epoch = 7; + assert!( + claim_scanner_leadership( + &ctx, + store.clone(), + &mut cycle, + &mut revision, + &mut persisted_epoch, + true, + ScannerCycleResetPolicy::ResetCoveragePreservingNext, + ) + .await + ); + + let persisted = read_config(store, DATA_USAGE_BLOOM_NAME_PATH.as_str()) + .await + .expect("recovered leadership claim should remain durable"); + let (persisted_cycle, claimed_epoch) = + decode_scanner_cycle_state(&persisted).expect("recovered leadership claim should decode"); + assert_eq!(persisted_cycle.next, 17_118_u64.max(winner_next)); + assert_eq!(persisted_cycle.current, 0); + assert!(persisted_cycle.cycle_completed.is_empty()); + assert_eq!(claimed_epoch, 8); + } +} + +#[test] +fn recovered_usage_cache_reset_keeps_cycle_and_leader_regression_guards() { + let source = DataUsageCacheSource::new(0, 0); + let digest = DataUsageScanPlanDigest([9; 32]); + let mut newer_cycle = DataUsageCache::default(); + newer_cycle.info.next_cycle = 17_119; + assert_eq!( + newer_cycle.prepare_for_scan(DATA_USAGE_ROOT, 17_118, 8, source, digest, true), + DataUsageCachePrepareOutcome::RejectedNewerCycle + ); + + let mut newer_leader = DataUsageCache::default(); + newer_leader.info.next_cycle = 17_118; + newer_leader.info.leader_epoch = 9; + assert_eq!( + newer_leader.prepare_for_scan(DATA_USAGE_ROOT, 17_118, 8, source, digest, true), + DataUsageCachePrepareOutcome::RejectedNewerLeader + ); +} + #[tokio::test] async fn test_leadership_claim_rejects_terminal_epoch() { let store = Arc::new(MemoryConfigStore::default()); @@ -3461,7 +3669,18 @@ async fn test_leadership_claim_rejects_terminal_epoch() { }; let mut persisted_epoch = u64::MAX - 1; - assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await); + assert!( + !claim_scanner_leadership( + &ctx, + store.clone(), + &mut cycle, + &mut revision, + &mut persisted_epoch, + false, + ScannerCycleResetPolicy::None, + ) + .await + ); assert_eq!(persisted_epoch, u64::MAX - 1); assert!(read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await.is_err()); } @@ -3477,7 +3696,18 @@ async fn scanner_defers_leadership_when_usage_snapshots_are_stably_absent() { }; let mut persisted_epoch = 0; - assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await); + assert!( + !claim_scanner_leadership( + &ctx, + store.clone(), + &mut cycle, + &mut revision, + &mut persisted_epoch, + false, + ScannerCycleResetPolicy::None, + ) + .await + ); assert!(read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH).await.is_err()); assert!(read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str()).await.is_err()); } @@ -3493,9 +3723,31 @@ async fn usage_bootstrap_pending_unblocks_first_leadership_claim() { let mut revision = DataUsageCacheRevision::Missing; let mut cycle = CurrentCycle::default(); let mut persisted_epoch = 0; - assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false,).await); + assert!( + !claim_scanner_leadership( + &ctx, + store.clone(), + &mut cycle, + &mut revision, + &mut persisted_epoch, + false, + ScannerCycleResetPolicy::None, + ) + .await + ); assert!(read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH).await.is_err()); - assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, true, true).await); + assert!( + claim_scanner_leadership( + &ctx, + store.clone(), + &mut cycle, + &mut revision, + &mut persisted_epoch, + true, + ScannerCycleResetPolicy::ResetAll, + ) + .await + ); let usage = read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str()) .await @@ -3583,7 +3835,18 @@ async fn leadership_claim_defers_on_corrupt_usage_baseline_without_bloom_write() }; let mut persisted_epoch = 0; - assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await); + assert!( + !claim_scanner_leadership( + &ctx, + store.clone(), + &mut cycle, + &mut revision, + &mut persisted_epoch, + false, + ScannerCycleResetPolicy::None, + ) + .await + ); assert!(read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await.is_err()); } @@ -3603,7 +3866,18 @@ async fn leadership_claim_defers_on_unidentified_usage_baseline_without_bloom_wr }; let mut persisted_epoch = 0; - assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await); + assert!( + !claim_scanner_leadership( + &ctx, + store.clone(), + &mut cycle, + &mut revision, + &mut persisted_epoch, + false, + ScannerCycleResetPolicy::None, + ) + .await + ); assert!(read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await.is_err()); } @@ -3625,7 +3899,18 @@ async fn test_leadership_claim_confirms_commit_after_returned_error() { let mut persisted_epoch = 0; seed_usage_snapshot_for_leadership_claim(&store).await; - assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await); + assert!( + claim_scanner_leadership( + &ctx, + store.clone(), + &mut cycle, + &mut revision, + &mut persisted_epoch, + false, + ScannerCycleResetPolicy::None, + ) + .await + ); let state = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH) .await @@ -3681,7 +3966,18 @@ async fn test_leadership_claim_usage_fence_rejects_old_inflight_writer() { ..Default::default() }; let mut persisted_epoch = 4; - assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await); + assert!( + claim_scanner_leadership( + &ctx, + store.clone(), + &mut cycle, + &mut revision, + &mut persisted_epoch, + false, + ScannerCycleResetPolicy::None, + ) + .await + ); let (fenced_data, fenced_revision) = read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) .await @@ -3747,7 +4043,7 @@ async fn cycle_budget_lease_takeover_rejects_old_generation() { &mut replacement_revision, &mut replacement_epoch, false, - false, + ScannerCycleResetPolicy::None, ) .await );