From 1590d9107bde322ee36b3a6bad98042f3c27a093 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Wed, 26 Aug 2026 12:30:53 +0800 Subject: [PATCH] fix(scanner): rebuild missing usage floor after upgrade (#6624) * fix(scanner): rebuild missing usage floor after upgrade * fix(scanner): preserve missing-floor reset across conflicts --------- Co-authored-by: Henry Guo Co-authored-by: overtrue --- crates/scanner/src/scanner.rs | 121 +++++----- crates/scanner/src/scanner/cycle_state.rs | 20 +- crates/scanner/src/scanner/leadership.rs | 67 +++--- crates/scanner/src/scanner/tests.rs | 256 ++++++++++++++++------ 4 files changed, 293 insertions(+), 171 deletions(-) diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 8fd155387..e31cad163 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -403,7 +403,7 @@ pub(super) fn data_usage_info_has_persisted_baseline_identity(info: &DataUsageIn && u64::try_from(info.buckets_usage.len()).ok() == Some(info.buckets_count) } -pub(super) fn data_usage_info_is_pristine_bootstrap_pending(info: &DataUsageInfo) -> bool { +pub(super) fn data_usage_info_is_bootstrap_pending(info: &DataUsageInfo) -> bool { if info.last_update.is_none() || info.scanner_cycle.is_some() { return false; } @@ -681,27 +681,30 @@ async fn initial_scanner_startup_usage_state(storeapi: &Arc) -> (bool, (persisted_usage_cache_is_cold_for_startup(storeapi).await, has_buckets) } -fn scanner_cycle_state_is_pristine( - cycle_info: &CurrentCycle, - leader_epoch: u64, - cycle_revision: &DataUsageCacheRevision, -) -> bool { - cycle_info.next == 0 && leader_epoch == 0 && matches!(cycle_revision, DataUsageCacheRevision::Missing) -} - -fn scanner_may_bootstrap_missing_usage_floor( - cycle_info: &CurrentCycle, - leader_epoch: u64, - cycle_revision: &DataUsageCacheRevision, -) -> bool { - // The server becomes ready before the scanner starts, so a first bucket may - // already exist. The bootstrap marker is non-authoritative; only prior - // durable scanner progress must block its creation. - scanner_cycle_state_is_pristine(cycle_info, leader_epoch, cycle_revision) -} - -fn scanner_may_resume_pristine_usage_bootstrap(cycle_info: &CurrentCycle) -> bool { - cycle_info.next == 0 +fn prepare_cycle_for_usage_floor_bootstrap( + cycle_info: &mut CurrentCycle, + usage_floor: PersistedUsageFloor, + startup: PersistedUsageFloorStartup, +) -> (bool, bool) { + match startup { + PersistedUsageFloorStartup::Authoritative => (false, false), + 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) + } + PersistedUsageFloorStartup::BootstrapPending => { + // An unfenced marker may have been written before an upgrade's old + // cycle state was replaced. A fenced marker belongs to the current + // scanner generation and may retain partial-cycle progress. + if usage_floor.leader_epoch == 0 { + *cycle_info = CurrentCycle::default(); + } + (true, usage_floor.leader_epoch == 0) + } + } } pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { @@ -1220,7 +1223,7 @@ where LockLost: Future, { let fence_ctx = ctx.child_token(); - let claim = claim_scanner_leadership(&fence_ctx, storeapi, cycle_info, cycle_revision, leader_epoch, false); + let claim = claim_scanner_leadership(&fence_ctx, storeapi, cycle_info, cycle_revision, leader_epoch, false, false); tokio::pin!(claim); tokio::pin!(lock_lost); tokio::select! { @@ -2186,47 +2189,30 @@ async fn run_data_scanner_with_maintenance_state( return Err(err); } }; - let may_bootstrap_missing_usage_floor = scanner_may_bootstrap_missing_usage_floor(&cycle_info, leader_epoch, &cycle_revision); - let (usage_floor, usage_floor_startup) = - match persisted_usage_floor_for_startup(storeapi.clone(), may_bootstrap_missing_usage_floor).await { - Ok(result) => result, - Err(err) => { - error!( - target: "rustfs::scanner", - event = EVENT_SCANNER_PERSIST_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_RUNTIME, - path = %DATA_USAGE_OBJ_NAME_PATH.as_str(), - state = "usage_floor_load_failed", - error = %err, - "Scanner stopped because the persisted usage floor could not be loaded" - ); - global_metrics().set_cycle(None).await; - return Ok(()); - } - }; - if usage_floor_startup == PersistedUsageFloorStartup::BootstrapPending - && !scanner_may_resume_pristine_usage_bootstrap(&cycle_info) - { - error!( - target: "rustfs::scanner", - event = EVENT_SCANNER_PERSIST_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_RUNTIME, - path = %DATA_USAGE_OBJ_NAME_PATH.as_str(), - state = "usage_floor_bootstrap_conflict", - next_cycle = cycle_info.next, - "Scanner stopped because a pristine usage bootstrap conflicts with persisted cycle progress" - ); - global_metrics().set_cycle(None).await; - return Ok(()); - } + let (usage_floor, usage_floor_startup) = match persisted_usage_floor_for_startup(storeapi.clone(), true).await { + Ok(result) => result, + Err(err) => { + error!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + path = %DATA_USAGE_OBJ_NAME_PATH.as_str(), + state = "usage_floor_load_failed", + error = %err, + "Scanner stopped because the persisted usage floor could not be loaded" + ); + global_metrics().set_cycle(None).await; + return Ok(()); + } + }; + let (allow_usage_floor_bootstrap_pending, reset_usage_floor_bootstrap_cycle_on_conflict) = + 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); - let allow_pristine_bootstrap_pending = match usage_floor_startup { - PersistedUsageFloorStartup::Authoritative => false, - PersistedUsageFloorStartup::BootstrapPending => true, + match usage_floor_startup { + PersistedUsageFloorStartup::Authoritative | PersistedUsageFloorStartup::BootstrapPending => {} PersistedUsageFloorStartup::Missing => { - if !may_bootstrap_missing_usage_floor || ctx.is_cancelled() || guard.is_lock_lost() { + if ctx.is_cancelled() || guard.is_lock_lost() { global_metrics().set_cycle(None).await; return Ok(()); } @@ -2234,12 +2220,12 @@ async fn run_data_scanner_with_maintenance_state( let bootstrap_ctx = ctx.child_token(); match await_scanner_cycle_with_lock_fence( &bootstrap_ctx, - initialize_pristine_usage_baseline(storeapi.clone()), + initialize_usage_baseline_bootstrap(storeapi.clone()), guard.lock_lost_notified(), ) .await { - Some(Ok(())) => true, + Some(Ok(())) => {} Some(Err(err)) => { error!( target: "rustfs::scanner", @@ -2249,7 +2235,7 @@ async fn run_data_scanner_with_maintenance_state( path = %DATA_USAGE_OBJ_NAME_PATH.as_str(), state = "usage_floor_bootstrap_failed", error = %err, - "Scanner stopped because the pristine usage bootstrap could not be initialized" + "Scanner stopped because the usage baseline bootstrap could not be initialized" ); global_metrics().set_cycle(None).await; return Ok(()); @@ -2260,7 +2246,7 @@ async fn run_data_scanner_with_maintenance_state( } } } - }; + } if ctx.is_cancelled() || guard.is_lock_lost() { global_metrics().set_cycle(None).await; @@ -2275,7 +2261,8 @@ async fn run_data_scanner_with_maintenance_state( &mut cycle_info, &mut cycle_revision, &mut leader_epoch, - allow_pristine_bootstrap_pending, + allow_usage_floor_bootstrap_pending, + reset_usage_floor_bootstrap_cycle_on_conflict, ), guard.lock_lost_notified(), ) diff --git a/crates/scanner/src/scanner/cycle_state.rs b/crates/scanner/src/scanner/cycle_state.rs index c09e8e360..87540cf98 100644 --- a/crates/scanner/src/scanner/cycle_state.rs +++ b/crates/scanner/src/scanner/cycle_state.rs @@ -1319,7 +1319,7 @@ pub(super) async fn persisted_usage_floor( pub(super) async fn persisted_usage_floor_for_startup( storeapi: Arc, - allow_missing_for_pristine_startup: bool, + allow_missing_for_bootstrap: bool, ) -> Result<(PersistedUsageFloor, PersistedUsageFloorStartup), ScannerError> { let Some(read_epoch) = scanner_publication_epoch(storeapi.clone()).await else { return Err(ScannerError::Other("scanner usage floor read is blocked by data movement".to_string())); @@ -1345,11 +1345,9 @@ pub(super) async fn persisted_usage_floor_for_startup( let usage = serde_json::from_slice::(&data).map_err(|err| { ScannerError::Other(format!("failed to decode scanner usage floor from {primary_path}: {err}")) })?; - if data_usage_info_is_pristine_bootstrap_pending(&usage) && primary_path == DATA_USAGE_OBJ_NAME_PATH.as_str() { + if data_usage_info_is_bootstrap_pending(&usage) && primary_path == DATA_USAGE_OBJ_NAME_PATH.as_str() { if bootstrap_pending { - return Err(ScannerError::Other( - "multiple pristine scanner usage bootstrap markers were found".to_string(), - )); + return Err(ScannerError::Other("multiple scanner usage bootstrap markers were found".to_string())); } bootstrap_pending = true; update_floor(&mut floor, &usage, primary_path)?; @@ -1376,7 +1374,7 @@ pub(super) async fn persisted_usage_floor_for_startup( Ok((Some(data), _)) => { if bootstrap_pending { return Err(ScannerError::Other( - "pristine scanner usage bootstrap conflicts with a persisted backup".to_string(), + "scanner usage bootstrap conflicts with a persisted backup".to_string(), )); } any_found = true; @@ -1406,7 +1404,7 @@ pub(super) async fn persisted_usage_floor_for_startup( if any_found { if bootstrap_pending { return Err(ScannerError::Other( - "pristine scanner usage bootstrap conflicts with an authoritative usage floor".to_string(), + "scanner usage bootstrap conflicts with an authoritative usage floor".to_string(), )); } found_any = true; @@ -1415,14 +1413,14 @@ pub(super) async fn persisted_usage_floor_for_startup( } if !found_any && !bootstrap_pending { - if !allow_missing_for_pristine_startup { + if !allow_missing_for_bootstrap { return Err(ScannerError::Other( "persisted scanner usage floor has no authoritative baseline".to_string(), )); } let Some(publication_admission) = scanner_publication_admission_for_epoch(storeapi.clone(), read_epoch).await else { return Err(ScannerError::Other( - "scanner usage floor changed before pristine state confirmation".to_string(), + "scanner usage floor changed before missing-state confirmation".to_string(), )); }; for path in [ @@ -1435,12 +1433,12 @@ pub(super) async fn persisted_usage_floor_for_startup( Ok((None, _)) => {} Ok((Some(_), _)) => { return Err(ScannerError::Other(format!( - "scanner usage floor changed while confirming pristine state: {path} appeared" + "scanner usage floor changed while confirming missing state: {path} appeared" ))); } Err(err) => { return Err(ScannerError::Other(format!( - "failed to confirm pristine scanner usage floor at {path}: {err}" + "failed to confirm missing scanner usage floor at {path}: {err}" ))); } } diff --git a/crates/scanner/src/scanner/leadership.rs b/crates/scanner/src/scanner/leadership.rs index 80df0ff18..d2afa1d1d 100644 --- a/crates/scanner/src/scanner/leadership.rs +++ b/crates/scanner/src/scanner/leadership.rs @@ -63,14 +63,12 @@ pub(super) async fn reconcile_scanner_leadership_claim( pub(super) fn decode_usage_snapshot_for_epoch_fence( data: &[u8], path: &str, - allow_pristine_bootstrap_pending: bool, + allow_bootstrap_pending: bool, ) -> Result { let usage: DataUsageInfo = serde_json::from_slice(data) .map_err(|err| ScannerError::Other(format!("failed to decode scanner usage epoch fence from {path}: {err}")))?; if !data_usage_info_has_persisted_baseline_identity(&usage) - && !(allow_pristine_bootstrap_pending - && path == DATA_USAGE_OBJ_NAME_PATH.as_str() - && data_usage_info_is_pristine_bootstrap_pending(&usage)) + && !(allow_bootstrap_pending && path == DATA_USAGE_OBJ_NAME_PATH.as_str() && data_usage_info_is_bootstrap_pending(&usage)) { return Err(ScannerError::Other(format!( "scanner usage epoch fence from {path} has no persisted baseline identity" @@ -82,15 +80,11 @@ pub(super) fn decode_usage_snapshot_for_epoch_fence( pub(super) async fn usage_snapshot_for_epoch_fence( storeapi: Arc, primary: Option<&[u8]>, - allow_pristine_bootstrap_pending: bool, + allow_bootstrap_pending: bool, ) -> Result, ScannerError> { if let Some(primary) = primary { - return decode_usage_snapshot_for_epoch_fence( - primary, - DATA_USAGE_OBJ_NAME_PATH.as_str(), - allow_pristine_bootstrap_pending, - ) - .map(Some); + return decode_usage_snapshot_for_epoch_fence(primary, DATA_USAGE_OBJ_NAME_PATH.as_str(), allow_bootstrap_pending) + .map(Some); } let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()); @@ -118,12 +112,12 @@ pub(super) async fn usage_snapshot_for_epoch_fence( Ok(None) } -pub(super) async fn initialize_pristine_usage_baseline( +pub(super) async fn initialize_usage_baseline_bootstrap( storeapi: Arc, ) -> Result<(), ScannerError> { let Some(expected_epoch) = scanner_publication_epoch(storeapi.clone()).await else { return Err(ScannerError::Other( - "pristine scanner usage baseline initialization is blocked by data movement".to_string(), + "scanner usage baseline bootstrap is blocked by data movement".to_string(), )); }; let baseline = DataUsageInfo { @@ -133,7 +127,7 @@ pub(super) async fn initialize_pristine_usage_baseline( ..Default::default() }; let data = serde_json::to_vec(&baseline) - .map_err(|err| ScannerError::Other(format!("failed to encode pristine scanner usage baseline: {err}")))?; + .map_err(|err| ScannerError::Other(format!("failed to encode scanner usage baseline bootstrap: {err}")))?; let save_result = save_config_with_publication_admission_for_epoch( storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), @@ -153,14 +147,14 @@ pub(super) async fn initialize_pristine_usage_baseline( let (persisted, revision) = read_config_with_revision(storeapi, DATA_USAGE_OBJ_NAME_PATH.as_str()) .await - .map_err(|err| ScannerError::Other(format!("failed to reconcile pristine scanner usage bootstrap: {err}")))?; + .map_err(|err| ScannerError::Other(format!("failed to reconcile scanner usage bootstrap: {err}")))?; if persisted.as_deref() == Some(data.as_slice()) && matches!(revision, DataUsageCacheRevision::Etag(_)) { return Ok(()); } Err(ScannerError::Other(match save_result { - Ok(_) => "pristine scanner usage bootstrap returned no ETag and could not be confirmed".to_string(), - Err(err) => format!("failed to persist pristine scanner usage bootstrap: {err}"), + Ok(_) => "scanner usage bootstrap returned no ETag and could not be confirmed".to_string(), + Err(err) => format!("failed to persist scanner usage bootstrap: {err}"), })) } @@ -169,7 +163,7 @@ pub(super) async fn fence_scanner_usage_epoch_with_expected_epoch( storeapi: Arc, claimed_epoch: u64, expected_publication_epoch: Option, - allow_pristine_bootstrap_pending: bool, + allow_bootstrap_pending: bool, ) -> Result<(), ScannerError> { for retry in 0..=SCANNER_PERSIST_CAS_RETRIES { if ctx.is_cancelled() { @@ -193,7 +187,7 @@ pub(super) async fn fence_scanner_usage_epoch_with_expected_epoch( .await .map_err(|err| ScannerError::Other(format!("failed to read scanner usage epoch fence: {err}")))?; let Some(mut usage) = - usage_snapshot_for_epoch_fence(storeapi.clone(), primary.as_deref(), allow_pristine_bootstrap_pending).await? + usage_snapshot_for_epoch_fence(storeapi.clone(), primary.as_deref(), allow_bootstrap_pending).await? else { let Some(_publication_admission) = scanner_publication_admission_for_epoch(storeapi.clone(), read_epoch).await else { if retry < SCANNER_PERSIST_CAS_RETRIES { @@ -243,11 +237,8 @@ pub(super) async fn fence_scanner_usage_epoch_with_expected_epoch( .await .map_err(|err| ScannerError::Other(format!("failed to reconcile scanner usage epoch fence: {err}")))?; if let Some(persisted) = persisted { - let persisted = decode_usage_snapshot_for_epoch_fence( - &persisted, - DATA_USAGE_OBJ_NAME_PATH.as_str(), - allow_pristine_bootstrap_pending, - )?; + let persisted = + decode_usage_snapshot_for_epoch_fence(&persisted, DATA_USAGE_OBJ_NAME_PATH.as_str(), allow_bootstrap_pending)?; match persisted.scanner_epoch { Some(epoch) if epoch == claimed_epoch => return Ok(()), Some(epoch) if epoch > claimed_epoch => { @@ -277,14 +268,14 @@ pub(super) async fn complete_scanner_leadership_claim( storeapi: Arc, claimed_epoch: u64, expected_publication_epoch: Option, - allow_pristine_bootstrap_pending: bool, + allow_bootstrap_pending: bool, ) -> bool { if let Err(err) = fence_scanner_usage_epoch_with_expected_epoch( ctx, storeapi, claimed_epoch, expected_publication_epoch, - allow_pristine_bootstrap_pending, + allow_bootstrap_pending, ) .await { @@ -310,7 +301,8 @@ pub(super) async fn claim_scanner_leadership( cycle_info: &mut CurrentCycle, revision: &mut DataUsageCacheRevision, persisted_epoch: &mut u64, - allow_pristine_bootstrap_pending: bool, + allow_bootstrap_pending: bool, + reset_bootstrap_cycle_on_conflict: bool, ) -> bool { for retry in 0..=SCANNER_PERSIST_CAS_RETRIES { if ctx.is_cancelled() { @@ -365,7 +357,7 @@ pub(super) async fn claim_scanner_leadership( return false; } }; - match usage_snapshot_for_epoch_fence(storeapi.clone(), usage_primary.as_deref(), allow_pristine_bootstrap_pending).await { + match usage_snapshot_for_epoch_fence(storeapi.clone(), usage_primary.as_deref(), allow_bootstrap_pending).await { Ok(Some(_)) => {} Ok(None) => { warn!( @@ -413,7 +405,7 @@ pub(super) async fn claim_scanner_leadership( storeapi, claimed_epoch, Some(read_epoch), - allow_pristine_bootstrap_pending, + allow_bootstrap_pending, ) .await; } @@ -435,11 +427,16 @@ pub(super) async fn claim_scanner_leadership( storeapi, claimed_epoch, Some(read_epoch), - allow_pristine_bootstrap_pending, + allow_bootstrap_pending, ) .await; } - Ok(ScannerLeadershipClaimReconcile::Changed) if retry < SCANNER_PERSIST_CAS_RETRIES => continue, + Ok(ScannerLeadershipClaimReconcile::Changed) if retry < SCANNER_PERSIST_CAS_RETRIES => { + if reset_bootstrap_cycle_on_conflict { + *cycle_info = CurrentCycle::default(); + } + continue; + } Ok(ScannerLeadershipClaimReconcile::Changed | ScannerLeadershipClaimReconcile::Unchanged) => { error!( target: "rustfs::scanner", @@ -486,18 +483,24 @@ pub(super) async fn claim_scanner_leadership( storeapi, claimed_epoch, Some(read_epoch), - allow_pristine_bootstrap_pending, + allow_bootstrap_pending, ) .await; } Ok(ScannerLeadershipClaimReconcile::Changed) if retry < SCANNER_PERSIST_CAS_RETRIES && !ctx.is_cancelled() => { + if reset_bootstrap_cycle_on_conflict { + *cycle_info = CurrentCycle::default(); + } 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(); + } continue; } Ok(ScannerLeadershipClaimReconcile::Changed | ScannerLeadershipClaimReconcile::Unchanged) => { diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 28c7e0659..e0da552b1 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -2111,11 +2111,11 @@ async fn scanner_usage_floor_fails_closed_on_corrupt_or_exhausted_usage_state() } #[tokio::test] -async fn scanner_usage_floor_allows_only_explicit_pristine_bootstrap() { +async fn scanner_usage_floor_allows_only_explicit_missing_state_bootstrap() { let store = Arc::new(MemoryConfigStore::default()); let (floor, state) = persisted_usage_floor_for_startup(store.clone(), true) .await - .expect("a verified pristine startup should use the empty floor"); + .expect("a verified missing state should use the empty floor"); assert_eq!(floor, PersistedUsageFloor::default()); assert_eq!(state, PersistedUsageFloorStartup::Missing); assert!(persisted_usage_floor_for_startup(store.clone(), false).await.is_err()); @@ -2126,7 +2126,7 @@ async fn scanner_usage_floor_allows_only_explicit_pristine_bootstrap() { ); assert!( persisted_usage_floor_for_startup(store, true).await.is_err(), - "pristine bootstrap must not hide corrupt persisted state" + "usage bootstrap must not hide corrupt persisted state" ); } @@ -2156,16 +2156,16 @@ async fn scanner_usage_floor_fails_closed_on_zero_byte_usage_objects() { let err = persisted_usage_floor_for_startup(appearing, true) .await - .expect_err("an empty usage object appearing during confirmation must prevent pristine bootstrap"); + .expect_err("an empty usage object appearing during confirmation must prevent usage bootstrap"); assert!( - err.to_string().contains("changed while confirming pristine state"), + err.to_string().contains("changed while confirming missing state"), "unexpected confirmation error for {path}: {err}" ); } } #[tokio::test] -async fn scanner_usage_floor_requires_publication_admission_for_pristine_bootstrap() { +async fn scanner_usage_floor_requires_publication_admission_for_bootstrap() { let store = Arc::new(MemoryConfigStore::default()); store.publication_admission_blocked.store(true, Ordering::Release); @@ -2173,18 +2173,18 @@ async fn scanner_usage_floor_requires_publication_admission_for_pristine_bootstr } #[tokio::test] -async fn scanner_usage_floor_fails_closed_when_usage_appears_during_pristine_confirmation() { +async fn scanner_usage_floor_fails_closed_when_usage_appears_during_missing_confirmation() { let store = Arc::new(MemoryConfigStore::default()); insert_usage_after_first_legacy_backup_read(store.as_ref()).await; let err = persisted_usage_floor_for_startup(store, true) .await - .expect_err("an appearing usage snapshot must prevent pristine bootstrap"); - assert!(err.to_string().contains("changed while confirming pristine state")); + .expect_err("an appearing usage snapshot must prevent usage bootstrap"); + assert!(err.to_string().contains("changed while confirming missing state")); } #[tokio::test] -async fn scanner_usage_floor_rejects_publication_change_during_pristine_confirmation() { +async fn scanner_usage_floor_rejects_publication_change_during_missing_confirmation() { let store = Arc::new(MemoryConfigStore::default()); store.block_publication_after_admissions.store(2, Ordering::Release); @@ -2192,42 +2192,139 @@ async fn scanner_usage_floor_rejects_publication_change_during_pristine_confirma } #[test] -fn scanner_pristine_cycle_state_requires_no_durable_progress() { - let cycle = CurrentCycle::default(); - assert!(scanner_cycle_state_is_pristine(&cycle, 0, &DataUsageCacheRevision::Missing)); - assert!(scanner_may_resume_pristine_usage_bootstrap(&cycle)); - assert!(!scanner_cycle_state_is_pristine( - &CurrentCycle { - next: 1, - ..Default::default() - }, - 0, - &DataUsageCacheRevision::Missing - )); - assert!(!scanner_may_resume_pristine_usage_bootstrap(&CurrentCycle { - next: 1, +fn missing_usage_floor_discards_unfenced_cycle_progress() { + let mut cycle = CurrentCycle { + current: 11, + next: 12, + cycle_completed: vec![Utc::now()], + started: Utc::now(), + }; + + assert_eq!( + prepare_cycle_for_usage_floor_bootstrap(&mut cycle, PersistedUsageFloor::default(), PersistedUsageFloorStartup::Missing,), + (true, true) + ); + assert_eq!(cycle.next, 0); + assert_eq!(cycle.current, 0); + assert!(cycle.cycle_completed.is_empty()); + + cycle.next = 12; + assert_eq!( + prepare_cycle_for_usage_floor_bootstrap( + &mut cycle, + PersistedUsageFloor::default(), + PersistedUsageFloorStartup::BootstrapPending, + ), + (true, true) + ); + assert_eq!(cycle.next, 0); +} + +#[test] +fn fenced_usage_bootstrap_retains_partial_cycle_progress() { + let mut cycle = CurrentCycle { + next: 12, ..Default::default() - })); - assert!(!scanner_cycle_state_is_pristine(&cycle, 1, &DataUsageCacheRevision::Missing)); - assert!(!scanner_cycle_state_is_pristine( - &cycle, - 0, - &DataUsageCacheRevision::Etag("etag".to_string()) - )); + }; + + assert_eq!( + prepare_cycle_for_usage_floor_bootstrap( + &mut cycle, + PersistedUsageFloor { + next_cycle: 0, + leader_epoch: 7, + }, + PersistedUsageFloorStartup::BootstrapPending, + ), + (true, false) + ); + assert_eq!(cycle.next, 12); + + assert_eq!( + prepare_cycle_for_usage_floor_bootstrap( + &mut cycle, + PersistedUsageFloor { + next_cycle: 13, + leader_epoch: 7, + }, + PersistedUsageFloorStartup::Authoritative, + ), + (false, false) + ); + assert_eq!(cycle.next, 12); } #[tokio::test] -async fn scanner_pristine_bootstrap_allows_first_bucket_to_win_startup() { +async fn missing_usage_floor_rebuilds_persisted_cycle_before_leadership_claim() { + let store = Arc::new(MemoryConfigStore::default()); + let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()); + let stale_cycle = CurrentCycle { + next: 12, + ..Default::default() + }; + store.objects.lock().await.insert( + state_key.clone(), + encode_scanner_cycle_state(&stale_cycle, 4).expect("stale cycle state should encode"), + ); + store.revisions.lock().await.insert(state_key, 7); + + let ScannerCycleStateStartup::Ready { + cycle: mut cycle_info, + leader_epoch: mut persisted_epoch, + revision: mut cycle_revision, + } = load_scanner_cycle_state_for_startup(store.clone()).await + else { + panic!("valid persisted cycle state should load"); + }; + let (usage_floor, startup) = persisted_usage_floor_for_startup(store.clone(), true) + .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) = + 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()) + .await + .expect("missing usage floor should publish a pending marker"); + + assert!( + claim_scanner_leadership( + &CancellationToken::new(), + store.clone(), + &mut cycle_info, + &mut cycle_revision, + &mut persisted_epoch, + allow_bootstrap_pending, + reset_bootstrap_cycle_on_conflict, + ) + .await + ); + let persisted_cycle = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH) + .await + .expect("rebuilt cycle state should be persisted"); + let (persisted_cycle, persisted_cycle_epoch) = + decode_scanner_cycle_state(&persisted_cycle).expect("rebuilt cycle state should decode"); + assert_eq!(persisted_cycle.next, 0); + assert_eq!(persisted_cycle_epoch, 5); + + let pending = read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str()) + .await + .expect("fenced bootstrap marker should remain persisted"); + let pending = serde_json::from_slice::(&pending).expect("bootstrap marker should decode"); + assert!(data_usage_info_is_bootstrap_pending(&pending)); + assert_eq!(pending.scanner_epoch, Some(5)); + assert!(!data_usage_info_has_persisted_baseline_identity(&pending)); +} + +#[tokio::test] +async fn scanner_usage_bootstrap_allows_first_bucket_to_win_startup() { let (_temp_dir, store) = setup_scanner_cycle_store_with_usage_baseline(false).await; - let cycle = CurrentCycle::default(); - let revision = DataUsageCacheRevision::Missing; store .make_bucket("first-user-bucket", &crate::storage_api::scan::MakeBucketOptions::default()) .await .expect("test bucket should be created"); - assert!(scanner_may_bootstrap_missing_usage_floor(&cycle, 0, &revision)); assert_eq!( persisted_usage_floor_for_startup(store.clone(), true) .await @@ -2235,14 +2332,14 @@ async fn scanner_pristine_bootstrap_allows_first_bucket_to_win_startup() { .1, PersistedUsageFloorStartup::Missing ); - initialize_pristine_usage_baseline(store.clone()) + initialize_usage_baseline_bootstrap(store.clone()) .await .expect("first startup should persist its pending marker"); let pending = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) .await .expect("pending marker should be stored"); let pending = serde_json::from_slice::(&pending).expect("pending marker should decode"); - assert!(data_usage_info_is_pristine_bootstrap_pending(&pending)); + assert!(data_usage_info_is_bootstrap_pending(&pending)); assert!(!data_usage_info_has_persisted_baseline_identity(&pending)); assert_eq!( persisted_usage_floor_for_startup(store.clone(), false) @@ -2553,7 +2650,7 @@ 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).await); + assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await); let state = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH) .await @@ -2565,6 +2662,42 @@ async fn test_leadership_claim_preserves_usage_epoch_floor_across_old_epoch_conf assert_eq!(store.put_counts.lock().await.get(&key), Some(&3)); } +#[tokio::test] +async fn unfenced_usage_bootstrap_discards_old_epoch_conflict_progress() { + 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); + initialize_usage_baseline_bootstrap(store.clone()) + .await + .expect("missing usage floor should publish a pending marker"); + + cycle = CurrentCycle::default(); + let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()); + let stale_cycle = CurrentCycle { + next: 14, + ..Default::default() + }; + store.interleaving_puts.lock().await.insert( + key, + (2, encode_scanner_cycle_state(&stale_cycle, 1).expect("stale cycle state should encode")), + ); + + let mut persisted_epoch = 1; + assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, true, true).await); + + let state = read_config(store, &DATA_USAGE_BLOOM_NAME_PATH) + .await + .expect("rebuilt leadership claim should remain persisted"); + let (claimed_cycle, claimed_epoch) = decode_scanner_cycle_state(&state).expect("claimed cycle state should decode"); + assert_eq!(claimed_cycle.next, 0); + assert_eq!(claimed_epoch, 2); +} + #[tokio::test] async fn test_leadership_claim_rejects_terminal_epoch() { let store = Arc::new(MemoryConfigStore::default()); @@ -2576,7 +2709,7 @@ 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).await); + assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await); assert_eq!(persisted_epoch, u64::MAX - 1); assert!(read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await.is_err()); } @@ -2592,64 +2725,64 @@ 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).await); + assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).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()); } #[tokio::test] -async fn pristine_usage_bootstrap_pending_unblocks_first_leadership_claim() { +async fn usage_bootstrap_pending_unblocks_first_leadership_claim() { let store = Arc::new(MemoryConfigStore::default()); - initialize_pristine_usage_baseline(store.clone()) + initialize_usage_baseline_bootstrap(store.clone()) .await - .expect("verified pristine startup should publish its pending marker"); + .expect("verified missing state should publish its pending marker"); let ctx = CancellationToken::new(); 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,).await); + assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false,).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).await); + assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, true, true).await); let usage = read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str()) .await - .expect("leadership claim should fence the pristine baseline"); - let usage = serde_json::from_slice::(&usage).expect("pristine bootstrap marker should remain valid"); - assert!(data_usage_info_is_pristine_bootstrap_pending(&usage)); + .expect("leadership claim should fence the bootstrap marker"); + let usage = serde_json::from_slice::(&usage).expect("bootstrap marker should remain valid"); + assert!(data_usage_info_is_bootstrap_pending(&usage)); assert!(!data_usage_info_has_persisted_baseline_identity(&usage)); assert_eq!(usage.scanner_epoch, Some(1)); } #[tokio::test] -async fn existing_pristine_usage_bootstrap_is_resumed_after_restart() { +async fn existing_usage_bootstrap_is_resumed_after_restart() { let store = Arc::new(MemoryConfigStore::default()); - initialize_pristine_usage_baseline(store.clone()) + initialize_usage_baseline_bootstrap(store.clone()) .await - .expect("verified pristine startup should publish its pending marker"); + .expect("verified missing state should publish its pending marker"); let (floor, state) = persisted_usage_floor_for_startup(store.clone(), false) .await - .expect("restart should recognize the pending pristine bootstrap"); + .expect("restart should recognize the pending usage bootstrap"); assert_eq!(floor, PersistedUsageFloor::default()); assert_eq!(state, PersistedUsageFloorStartup::BootstrapPending); assert!(persisted_usage_floor(store).await.is_err()); } #[tokio::test] -async fn pristine_usage_bootstrap_reconciles_post_commit_error() { +async fn usage_bootstrap_reconciles_post_commit_error() { let store = Arc::new(MemoryConfigStore::default()); let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()); store.error_after_commit_put_number.lock().await.insert(key, 1); - initialize_pristine_usage_baseline(store.clone()) + initialize_usage_baseline_bootstrap(store.clone()) .await .expect("a committed pending marker should reconcile after a lost response"); let usage = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) .await .expect("the reconciled pending marker should remain"); let usage = serde_json::from_slice::(&usage).expect("pending marker should decode"); - assert!(data_usage_info_is_pristine_bootstrap_pending(&usage)); + assert!(data_usage_info_is_bootstrap_pending(&usage)); assert!(!data_usage_info_has_persisted_baseline_identity(&usage)); assert_eq!( persisted_usage_floor_for_startup(store.clone(), false) @@ -2662,7 +2795,7 @@ async fn pristine_usage_bootstrap_reconciles_post_commit_error() { } #[tokio::test] -async fn pristine_usage_bootstrap_does_not_overwrite_concurrent_replacement() { +async fn usage_bootstrap_does_not_overwrite_concurrent_replacement() { let store = Arc::new(MemoryConfigStore::default()); let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()); let replacement = serde_json::to_vec(&complete_usage_with_bucket_count(None, 1)).expect("replacement should encode"); @@ -2672,7 +2805,7 @@ async fn pristine_usage_bootstrap_does_not_overwrite_concurrent_replacement() { .await .insert(key, (1, replacement.clone())); - initialize_pristine_usage_baseline(store.clone()) + initialize_usage_baseline_bootstrap(store.clone()) .await .expect("the bootstrap write completed before the replacement"); assert_eq!( @@ -2698,7 +2831,7 @@ 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).await); + assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await); assert!(read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await.is_err()); } @@ -2718,7 +2851,7 @@ 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).await); + assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await); assert!(read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await.is_err()); } @@ -2740,7 +2873,7 @@ 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).await); + assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await); let state = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH) .await @@ -2796,7 +2929,7 @@ 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).await); + assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await); let (fenced_data, fenced_revision) = read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) .await @@ -2862,6 +2995,7 @@ async fn cycle_budget_lease_takeover_rejects_old_generation() { &mut replacement_revision, &mut replacement_epoch, false, + false, ) .await );