diff --git a/crates/scanner/src/scanner/cycle_state.rs b/crates/scanner/src/scanner/cycle_state.rs index 87540cf98..cd3d200c3 100644 --- a/crates/scanner/src/scanner/cycle_state.rs +++ b/crates/scanner/src/scanner/cycle_state.rs @@ -1327,6 +1327,13 @@ pub(super) async fn persisted_usage_floor_for_startup( let mut floor = PersistedUsageFloor::default(); let mut found_any = false; let mut bootstrap_pending = false; + // A valid JSON object without a baseline identity is not a floor and must + // never be treated as an empty one. It can, however, be a partially + // written v2 primary left behind during an upgrade. Keep its epoch as a + // fence while looking for a durable companion snapshot; if no companion + // is new enough, the caller still fails closed below. + let mut invalid_baseline_path: Option = None; + let mut invalid_baseline_epoch: Option = None; let update_floor = |floor: &mut PersistedUsageFloor, usage: &DataUsageInfo, path: &str| -> Result<(), ScannerError> { floor.leader_epoch = floor.leader_epoch.max(usage.scanner_epoch.unwrap_or_default()); if let Some(completed_cycle) = usage.scanner_cycle { @@ -1353,9 +1360,9 @@ pub(super) async fn persisted_usage_floor_for_startup( update_floor(&mut floor, &usage, primary_path)?; None } else if !data_usage_info_has_persisted_baseline_identity(&usage) { - return Err(ScannerError::Other(format!( - "scanner usage floor from {primary_path} has no persisted baseline identity" - ))); + invalid_baseline_path.get_or_insert_with(|| primary_path.to_string()); + invalid_baseline_epoch = invalid_baseline_epoch.max(usage.scanner_epoch); + None } else { let epoch = usage.scanner_epoch.unwrap_or_default(); update_floor(&mut floor, &usage, primary_path)?; @@ -1377,21 +1384,27 @@ pub(super) async fn persisted_usage_floor_for_startup( "scanner usage bootstrap conflicts with a persisted backup".to_string(), )); } - any_found = true; let usage = serde_json::from_slice::(&data).map_err(|err| { ScannerError::Other(format!("failed to decode scanner usage floor from {backup_path}: {err}")) })?; if !data_usage_info_has_persisted_baseline_identity(&usage) { - return Err(ScannerError::Other(format!( - "scanner usage floor from {backup_path} has no persisted baseline identity" - ))); - } - let backup_epoch = usage.scanner_epoch.unwrap_or_default(); - // A backup write from an older leader may complete after the - // primary epoch has been fenced. It must not advance the startup - // floor unless its epoch is at least as new as the primary. - if primary_epoch.is_none_or(|epoch| backup_epoch >= epoch) { - update_floor(&mut floor, &usage, &backup_path)?; + invalid_baseline_path.get_or_insert_with(|| backup_path.clone()); + invalid_baseline_epoch = invalid_baseline_epoch.max(usage.scanner_epoch); + // This is still persisted state, so it must not enable a + // missing-state bootstrap. Continue to a legacy pair in + // case it contains a complete, fenced snapshot. + any_found = false; + } else { + let backup_epoch = usage.scanner_epoch.unwrap_or_default(); + // A backup write from an older leader may complete after the + // primary epoch has been fenced. It must not advance the startup + // floor unless its epoch is at least as new as the primary. + if primary_epoch.is_none_or(|epoch| backup_epoch >= epoch) + && invalid_baseline_epoch.is_none_or(|epoch| backup_epoch >= epoch) + { + update_floor(&mut floor, &usage, &backup_path)?; + any_found = true; + } } } Ok((None, _)) => {} @@ -1413,6 +1426,11 @@ pub(super) async fn persisted_usage_floor_for_startup( } if !found_any && !bootstrap_pending { + if let Some(path) = invalid_baseline_path { + return Err(ScannerError::Other(format!( + "persisted scanner usage floor from {path} has no authoritative baseline or newer valid backup" + ))); + } if !allow_missing_for_bootstrap { return Err(ScannerError::Other( "persisted scanner usage floor has no authoritative baseline".to_string(), diff --git a/crates/scanner/src/scanner/leadership.rs b/crates/scanner/src/scanner/leadership.rs index d2afa1d1d..a99687f33 100644 --- a/crates/scanner/src/scanner/leadership.rs +++ b/crates/scanner/src/scanner/leadership.rs @@ -82,9 +82,30 @@ pub(super) async fn usage_snapshot_for_epoch_fence( primary: Option<&[u8]>, allow_bootstrap_pending: bool, ) -> Result, ScannerError> { + // A partially written v2 primary is not itself a baseline, but a durable + // companion may still provide one after an interrupted upgrade. Keep the + // primary epoch as a fence while checking those companions; malformed + // bytes and bootstrap markers retain their fail-closed behavior. + let mut invalid_primary_epoch = None; if let Some(primary) = primary { - return decode_usage_snapshot_for_epoch_fence(primary, DATA_USAGE_OBJ_NAME_PATH.as_str(), allow_bootstrap_pending) - .map(Some); + let usage: DataUsageInfo = serde_json::from_slice(primary).map_err(|err| { + ScannerError::Other(format!( + "failed to decode scanner usage epoch fence from {}: {err}", + DATA_USAGE_OBJ_NAME_PATH.as_str() + )) + })?; + if data_usage_info_has_persisted_baseline_identity(&usage) + || (allow_bootstrap_pending && data_usage_info_is_bootstrap_pending(&usage)) + { + return Ok(Some(usage)); + } + if data_usage_info_is_bootstrap_pending(&usage) { + return Err(ScannerError::Other(format!( + "scanner usage epoch fence from {} has no persisted baseline identity", + DATA_USAGE_OBJ_NAME_PATH.as_str() + ))); + } + invalid_primary_epoch = usage.scanner_epoch; } let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()); @@ -92,7 +113,10 @@ pub(super) async fn usage_snapshot_for_epoch_fence( .await .map_err(|err| ScannerError::Other(format!("failed to read scanner usage epoch fence backup: {err}")))?; if let Some(backup) = backup.as_deref() { - return decode_usage_snapshot_for_epoch_fence(backup, &backup_path, false).map(Some); + let usage = decode_usage_snapshot_for_epoch_fence(backup, &backup_path, false)?; + if invalid_primary_epoch.is_none_or(|epoch| usage.scanner_epoch.unwrap_or_default() >= epoch) { + return Ok(Some(usage)); + } } for path in [ @@ -103,7 +127,10 @@ pub(super) async fn usage_snapshot_for_epoch_fence( .await .map_err(|err| ScannerError::Other(format!("failed to read legacy scanner usage epoch fence: {err}")))?; if let Some(legacy) = legacy.as_deref() { - return decode_usage_snapshot_for_epoch_fence(legacy, &path, false).map(Some); + let usage = decode_usage_snapshot_for_epoch_fence(legacy, &path, false)?; + if invalid_primary_epoch.is_none_or(|epoch| usage.scanner_epoch.unwrap_or_default() >= epoch) { + return Ok(Some(usage)); + } } } // A missing usage snapshot is an uninitialized state, not an empty diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index c427e6035..08c882841 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -1923,6 +1923,116 @@ async fn scanner_startup_uses_primary_and_backup_usage_floor() { assert_eq!(epoch, 11); } +#[tokio::test] +async fn scanner_usage_floor_recovers_from_incomplete_v2_primary_using_fenced_backup() { + let store = Arc::new(MemoryConfigStore::default()); + let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()); + + // This shape is valid JSON from an interrupted v2 publication, but it is + // not a durable baseline because the snapshot is incomplete. It must not + // be converted into an empty floor. + let primary = DataUsageInfo { + scanner_epoch: Some(7), + scanner_cycle: Some(100), + usage_snapshot_complete: false, + ..Default::default() + }; + let mut backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0); + backup.scanner_epoch = Some(7); + backup.scanner_cycle = Some(103); + + for (path, usage) in [(DATA_USAGE_OBJ_NAME_PATH.as_str(), primary), (backup_path.as_str(), backup)] { + store.objects.lock().await.insert( + memory_config_key(RUSTFS_META_BUCKET, path), + serde_json::to_vec(&usage).expect("usage snapshot should encode"), + ); + } + + assert_eq!( + persisted_usage_floor(store) + .await + .expect("valid backup should recover the usage floor"), + PersistedUsageFloor { + next_cycle: 104, + leader_epoch: 7, + } + ); +} + +#[tokio::test] +async fn scanner_usage_floor_does_not_bootstrap_over_incomplete_v2_primary() { + let store = Arc::new(MemoryConfigStore::default()); + let primary = DataUsageInfo { + scanner_epoch: Some(7), + scanner_cycle: Some(100), + usage_snapshot_complete: false, + ..Default::default() + }; + store.objects.lock().await.insert( + memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()), + serde_json::to_vec(&primary).expect("usage snapshot should encode"), + ); + + let err = persisted_usage_floor_for_startup(store, true) + .await + .expect_err("an existing incomplete primary must remain fail-closed"); + assert!(err.to_string().contains("no authoritative baseline")); +} + +#[tokio::test] +async fn scanner_usage_floor_rejects_backup_older_than_incomplete_v2_primary() { + let store = Arc::new(MemoryConfigStore::default()); + let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()); + let primary = DataUsageInfo { + scanner_epoch: Some(7), + scanner_cycle: Some(100), + usage_snapshot_complete: false, + ..Default::default() + }; + let mut backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0); + backup.scanner_epoch = Some(6); + backup.scanner_cycle = Some(10_000); + + for (path, usage) in [(DATA_USAGE_OBJ_NAME_PATH.as_str(), primary), (backup_path.as_str(), backup)] { + store.objects.lock().await.insert( + memory_config_key(RUSTFS_META_BUCKET, path), + serde_json::to_vec(&usage).expect("usage snapshot should encode"), + ); + } + + let err = persisted_usage_floor_for_startup(store, true) + .await + .expect_err("an older backup must not cross the incomplete primary epoch fence"); + assert!(err.to_string().contains("no authoritative baseline")); +} + +#[tokio::test] +async fn scanner_leadership_fencing_recovers_incomplete_v2_primary_from_backup() { + let store = Arc::new(MemoryConfigStore::default()); + let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()); + let primary = serde_json::to_vec(&DataUsageInfo { + scanner_epoch: Some(7), + scanner_cycle: Some(100), + usage_snapshot_complete: false, + ..Default::default() + }) + .expect("incomplete usage snapshot should encode"); + let mut backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0); + backup.scanner_epoch = Some(7); + backup.scanner_cycle = Some(103); + store.objects.lock().await.insert( + memory_config_key(RUSTFS_META_BUCKET, &backup_path), + serde_json::to_vec(&backup).expect("backup usage snapshot should encode"), + ); + + let recovered = usage_snapshot_for_epoch_fence(store, Some(&primary), false) + .await + .expect("a valid backup should provide the fencing baseline") + .expect("the fencing baseline should be present"); + assert_eq!(recovered.scanner_epoch, Some(7)); + assert_eq!(recovered.scanner_cycle, Some(103)); +} + #[tokio::test] async fn scanner_usage_floor_ignores_older_backup_after_primary_epoch_fence() { let store = Arc::new(MemoryConfigStore::default());