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 );