diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index 64ceb292e..c566bd0a9 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -725,7 +725,7 @@ impl PeerRestClient { /// never take it offline no matter what its message says. The substring /// fallback only covers failures that exist purely as text, such as the /// dial errors `get_client` wraps. - fn is_network_like_error(err: &Error) -> bool { + pub(crate) fn is_network_like_error(err: &Error) -> bool { if let Error::Io(io_err) = err && let Some(status) = embedded_tonic_status(io_err) { diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index 0c78007c2..b69c60152 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -1483,7 +1483,7 @@ impl NotificationSys { futures.push(async move { let client = client.ok_or_else(|| Error::other(format!("scanner activity peer[{idx}] is unreachable")))?; let host = client.grid_host.clone(); - scanner_activity_with_timeout(SCANNER_ACTIVITY_PROBE_TIMEOUT, &host, client.scanner_activity()) + scanner_activity_with_retry(&client, &host) .await .map(|activity| (host, activity)) }); @@ -1962,6 +1962,44 @@ where .map_err(|_| Error::other(format!("scanner activity peer {host} timed out after {timeout_duration:?}")))? } +fn scanner_activity_should_retry(first_error: Option<&Error>, timed_out: bool) -> bool { + timed_out || first_error.is_some_and(PeerRestClient::is_network_like_error) +} + +/// Retry one activity probe after a bounded reconnect when the first attempt +/// failed at the transport boundary. A peer that answered with an invalid or +/// incompatible activity response is not retried here: it must remain a hard +/// fail-closed result for the all-peer publication proof. +async fn scanner_activity_with_retry(client: &PeerRestClient, host: &str) -> Result { + let first = timeout(SCANNER_ACTIVITY_PROBE_TIMEOUT, client.scanner_activity()).await; + let should_retry = match &first { + Ok(Ok(_)) => false, + Ok(Err(err)) => scanner_activity_should_retry(Some(err), false), + Err(_) => scanner_activity_should_retry(None, true), + }; + + match first { + Ok(Ok(activity)) => return Ok(activity), + Ok(Err(err)) if !should_retry => return Err(err), + Ok(Err(err)) => { + debug!(peer = host, error = %err, "scanner activity probe failed on first transport attempt; reconnecting"); + client.prepare_retry().await; + } + Err(_) => { + debug!(peer = host, timeout = ?SCANNER_ACTIVITY_PROBE_TIMEOUT, "scanner activity probe timed out on first attempt; reconnecting"); + client.prepare_retry().await; + } + } + + match timeout(SCANNER_ACTIVITY_PROBE_TIMEOUT, client.scanner_activity()).await { + Ok(result) => result, + Err(_) => { + client.evict_connection().await; + Err(Error::Timeout) + } + } +} + #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")] async fn call_peer_with_timeout( timeout_dur: Duration, @@ -2882,6 +2920,20 @@ mod tests { assert!(err.to_string().contains("peer-1")); } + #[test] + fn scanner_activity_retry_only_reconnects_transport_failures() { + assert!(scanner_activity_should_retry(None, true)); + assert!(scanner_activity_should_retry(Some(&Error::other("connection refused")), false)); + assert!(!scanner_activity_should_retry( + Some(&Error::other("peer returned an invalid scanner activity response proof")), + false + )); + assert!(!scanner_activity_should_retry( + Some(&Error::from(tonic::Status::internal("peer rejected activity"))), + false + )); + } + #[tokio::test] async fn scanner_dirty_usage_acknowledgement_rejects_missing_and_duplicate_targets() { let sys = NotificationSys { diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index d985931ad..788426b64 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1205,6 +1205,10 @@ fn data_usage_persist_timeout() -> Duration { DataUsageCache::persistence_timeout() } +fn scanner_publication_lease_budget_allows_persistence(timeout: Duration) -> bool { + timeout < Duration::from_millis(SCANNER_PUBLICATION_LEASE_TTL_MS) +} + #[cfg(not(test))] const SCANNER_CYCLE_EPOCH_FENCE_TIMEOUT: Duration = Duration::from_secs(30); #[cfg(test)] @@ -1416,13 +1420,10 @@ async fn run_data_scanner_cycle_with_budget( mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await; return ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::DataMovement); }; - let usage_persist_baseline_result = read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()).await; + let usage_persist_baseline_result = read_data_usage_persist_baseline(storeapi.clone()).await; drop(baseline_publication_guard); let usage_persist_baseline = match usage_persist_baseline_result { - Ok((data, revision)) => DataUsagePersistBaseline { - data: data.map(Bytes::from), - revision, - }, + Ok(baseline) => baseline, Err(err) => { error!( target: "rustfs::scanner", @@ -1462,10 +1463,23 @@ async fn run_data_scanner_cycle_with_budget( { Some(ScannerCycleDeferReason::DataMovement) } + // A complete walk can still be retained as an observational snapshot + // when only the final activity proof was unavailable. It must not + // block the observation receiver: the authoritative publication + // fence remains enforced by the usage store and the cycle is advanced + // as partial without acknowledging dirty usage. + Ok(result) + if result.has_observational_snapshot() + && matches!( + result.status, + ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable) + ) => + { + None + } Ok(result) => final_data_usage_publication_defer_reason(storeapi.as_ref(), result.status).await, Err(_) => Some(ScannerCycleDeferReason::ActivityBaselineUnavailable), }; - let publication_deferred = publication_defer_reason.is_some(); let publication_epoch = scan_result.as_ref().ok().and_then(ScannerCycleResult::publication_epoch); let remote_publication_lease_targets = if publication_defer_reason.is_none() { scan_result @@ -1479,11 +1493,11 @@ async fn run_data_scanner_cycle_with_budget( let mut remote_publication_leases = None; let remote_lease_defer_reason = if remote_publication_lease_targets.is_empty() { None - } else if usage_persist_timeout >= Duration::from_millis(SCANNER_PUBLICATION_LEASE_TTL_MS) { + } else if !scanner_publication_lease_budget_allows_persistence(usage_persist_timeout) { // The lease is intentionally fixed-duration and has no renewal path. // Refuse a persistence budget that could outlive it instead of // allowing the peer to admit movement while a local PUT is in flight. - Some(ScannerCycleDeferReason::ActivityBaselineUnavailable) + Some(ScannerCycleDeferReason::PublicationLeaseBudgetExceeded) } else if let Some(notification_system) = storeapi.notification_system() { match notification_system .acquire_scanner_publication_leases(remote_publication_lease_targets.clone()) @@ -1543,8 +1557,13 @@ async fn run_data_scanner_cycle_with_budget( .or(remote_lease_defer_reason) .or(remote_lease_fence_defer_reason); let publication_defer_reason = (!remote_lease_covers_persistence) - .then_some(ScannerCycleDeferReason::ActivityBaselineUnavailable) + .then_some(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded) .or(publication_defer_reason); + // Include reasons discovered while acquiring or validating remote leases. + // In particular, the static budget gate above is reached after the scan + // result is classified, so computing this flag earlier would suppress its + // deferred metric. + let publication_deferred = publication_defer_reason.is_some(); let budget_elapsed = cycle_budget.budget_elapsed() && !ctx.is_cancelled(); let remote_lease_probe = remote_publication_leases .as_ref() @@ -1640,14 +1659,17 @@ async fn run_data_scanner_cycle_with_budget( .is_some_and(|(_, grants)| grants.iter().any(|grant| !grant.lease.is_valid())); if let Some((notification_system, grants)) = remote_publication_leases.take() { let release_result = notification_system.release_scanner_publication_leases(grants).await; - if lease_expired || release_result.is_err() { + let lease_release_failed = release_result.is_err(); + if lease_expired || lease_release_failed { // A lease that expired or could not be released is never treated // as a successful authoritative publication. The peer may have // admitted movement immediately after the lease ended. usage_persist_outcome = if usage_persist_outcome == DataUsagePersistOutcome::Failed { DataUsagePersistOutcome::Failed + } else if lease_release_failed { + DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseReleaseFailed) } else { - DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable) + DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded) }; } } @@ -2803,12 +2825,7 @@ fn finalize_scanner_cycle_result( scan_cycle_result: crate::scanner_io::ScannerCycleResult, usage_persist_outcome: DataUsagePersistOutcome, ) -> (ScannerCycleOutcome, bool, Vec) { - let completion_outcome = scanner_cycle_completion_outcome( - scan_cycle_result.status, - usage_persist_outcome, - scan_cycle_result.has_dirty_usage_to_acknowledge(), - scan_cycle_result.has_failed_dirty_usage(), - ); + let completion_outcome = scanner_cycle_completion_outcome_for_result(&scan_cycle_result, usage_persist_outcome); let pending_maintenance_work = scan_cycle_result.has_pending_maintenance_work(); let durable_complete_snapshot = scan_cycle_result.status == ScannerCycleStatus::Complete && matches!( @@ -2823,6 +2840,34 @@ fn finalize_scanner_cycle_result( (completion_outcome, pending_maintenance_work, remote_dirty_usage_acknowledgements) } +fn scanner_cycle_completion_outcome_for_result( + scan_cycle_result: &crate::scanner_io::ScannerCycleResult, + usage_persist_outcome: DataUsagePersistOutcome, +) -> ScannerCycleOutcome { + let has_dirty_usage = scan_cycle_result.has_dirty_usage_to_acknowledge(); + let has_failed_dirty_usage = scan_cycle_result.has_failed_dirty_usage(); + if scan_cycle_result.has_observational_snapshot() + && matches!( + scan_cycle_result.status, + ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable) + ) + { + return match usage_persist_outcome { + DataUsagePersistOutcome::Saved + | DataUsagePersistOutcome::AlreadyDurable + | DataUsagePersistOutcome::PriorCycleDurable + | DataUsagePersistOutcome::Current + if !has_failed_dirty_usage => + { + ScannerCycleOutcome::Partial + } + DataUsagePersistOutcome::Deferred(reason) => ScannerCycleOutcome::Deferred(reason), + _ => ScannerCycleOutcome::Failed, + }; + } + scanner_cycle_completion_outcome(scan_cycle_result.status, usage_persist_outcome, has_dirty_usage, has_failed_dirty_usage) +} + /// Decide whether an incoming usage snapshot must be skipped as stale, given the local /// wall clock `now`. Mirrors `stale_data_usage_persist_reason` in /// `crates/ecstore/src/data_usage/mod.rs` — keep the two consistent. diff --git a/crates/scanner/src/scanner/cycle_state.rs b/crates/scanner/src/scanner/cycle_state.rs index 87540cf98..95a42e37f 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 { @@ -1340,6 +1347,7 @@ pub(super) async fn persisted_usage_floor_for_startup( }; for primary_path in [DATA_USAGE_OBJ_NAME_PATH.as_str(), LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()] { let backup_path = format!("{primary_path}.bkp"); + let is_v2_path = primary_path == DATA_USAGE_OBJ_NAME_PATH.as_str(); let primary_epoch = match read_config_with_revision(storeapi.clone(), primary_path).await { Ok((Some(data), _)) => { let usage = serde_json::from_slice::(&data).map_err(|err| { @@ -1353,13 +1361,20 @@ 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)?; - Some(epoch) + // A legacy snapshot may be structurally valid but older + // than an incomplete v2 snapshot left by a newer leader. + // Do not let that candidate regress the startup floor. + if !is_v2_path && invalid_baseline_epoch.is_some_and(|fenced_epoch| epoch < fenced_epoch) { + None + } else { + update_floor(&mut floor, &usage, primary_path)?; + Some(epoch) + } } } Ok((None, _)) => None, @@ -1377,21 +1392,26 @@ 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. + } 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 +1433,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..1aae284ee 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -1923,6 +1923,176 @@ async fn scanner_startup_uses_primary_and_backup_usage_floor() { assert_eq!(epoch, 11); } +#[tokio::test] +async fn scanner_usage_floor_keeps_valid_primary_when_backup_has_no_identity() { + let store = Arc::new(MemoryConfigStore::default()); + let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()); + let mut primary = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0); + primary.scanner_epoch = Some(8); + primary.scanner_cycle = Some(100); + let backup = DataUsageInfo { + scanner_epoch: Some(9), + scanner_cycle: Some(101), + usage_snapshot_complete: false, + ..Default::default() + }; + + 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 primary should remain authoritative"), + PersistedUsageFloor { + next_cycle: 101, + leader_epoch: 8, + } + ); +} + +#[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_usage_floor_rejects_older_legacy_primary_after_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() + }; + let mut legacy = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0); + legacy.scanner_epoch = Some(6); + legacy.scanner_cycle = Some(103); + for (path, usage) in [ + (DATA_USAGE_OBJ_NAME_PATH.as_str(), primary), + (LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str(), legacy), + ] { + 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 legacy baseline must not cross the incomplete v2 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()); @@ -3182,6 +3352,79 @@ async fn test_observational_usage_defers_when_authoritative_baseline_is_missing( ); } +#[tokio::test] +async fn test_observational_usage_uses_fenced_backup_when_v2_primary_has_no_identity() { + let store = Arc::new(MemoryConfigStore::default()); + 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); + store.objects.lock().await.insert( + memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()), + serde_json::to_vec(&primary).expect("incomplete primary should encode"), + ); + store.objects.lock().await.insert( + memory_config_key(RUSTFS_META_BUCKET, &format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str())), + serde_json::to_vec(&backup).expect("backup baseline should encode"), + ); + + let (sender, receiver) = mpsc::channel(1); + let mut observation = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)), 1); + observation.usage_snapshot_converged = Some(false); + sender.send(observation).await.expect("observation should enqueue"); + drop(sender); + + let outcome = store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe( + CancellationToken::new(), + store.clone(), + receiver, + None, + None, + || async { false }, + ) + .await; + + assert_eq!(outcome, DataUsagePersistOutcome::Saved); + let observed = read_config(store, DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str()) + .await + .expect("observational snapshot should be persisted"); + let observed = serde_json::from_slice::(&observed).expect("observational snapshot should decode"); + assert_eq!(observed.usage_snapshot_authoritative_baseline, Some(backup.snapshot_identity())); +} + +#[tokio::test] +async fn usage_baseline_does_not_fall_back_to_older_legacy_snapshot() { + let store = Arc::new(MemoryConfigStore::default()); + let primary = DataUsageInfo { + scanner_epoch: Some(7), + scanner_cycle: Some(100), + usage_snapshot_complete: false, + ..Default::default() + }; + let mut legacy = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0); + legacy.scanner_epoch = Some(6); + legacy.scanner_cycle = Some(103); + let primary_data = serde_json::to_vec(&primary).expect("incomplete primary should encode"); + store.objects.lock().await.insert( + memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()), + primary_data.clone(), + ); + store.objects.lock().await.insert( + memory_config_key(RUSTFS_META_BUCKET, LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()), + serde_json::to_vec(&legacy).expect("legacy baseline should encode"), + ); + + let baseline = read_data_usage_persist_baseline(store) + .await + .expect("baseline inspection should complete"); + assert_eq!(baseline.data.as_deref(), Some(primary_data.as_slice())); +} + #[tokio::test] #[serial] async fn test_usage_route_barrier_precedes_durable_reconciliation() { @@ -3251,6 +3494,44 @@ async fn coordinator_does_not_put_after_remote_generation_flip() { assert_eq!(store.put_counts.lock().await.get(&key), None); } +#[tokio::test] +async fn coordinator_classifies_an_expired_publication_lease() { + let store = Arc::new(MemoryConfigStore::default()); + let (sender, receiver) = mpsc::channel(1); + sender + .send(complete_usage_with_bucket_count( + Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)), + 1, + )) + .await + .expect("usage snapshot should enqueue"); + drop(sender); + + let expired = std::time::Instant::now() + .checked_sub(std::time::Duration::from_secs(1)) + .expect("test instant should support a one-second subtraction"); + let outcome = + store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence( + CancellationToken::new(), + store.clone(), + receiver, + None, + Some(DataUsagePersistBaseline { + data: None, + revision: DataUsageCacheRevision::Missing, + }), + ScannerPublicationFence::new(None, Some(expired), None), + || async { false }, + ) + .await; + + assert_eq!( + outcome, + DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded) + ); + assert!(store.put_counts.lock().await.is_empty(), "expired lease must prevent a PUT"); +} + #[tokio::test] #[serial] async fn test_deferred_usage_save_keeps_last_real_save_metric() { @@ -4302,6 +4583,9 @@ fn scanner_cycle_cache_floor_stays_pending_during_deferred_usage_publication() { for reason in [ ScannerCycleDeferReason::DataMovement, ScannerCycleDeferReason::ActivityBaselineUnavailable, + ScannerCycleDeferReason::PublicationLeaseBudgetExceeded, + ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded, + ScannerCycleDeferReason::PublicationLeaseReleaseFailed, ] { let deferred = DataUsagePersistOutcome::Deferred(reason); assert_eq!( @@ -4368,6 +4652,26 @@ fn finalizing_a_deferred_usage_save_keeps_dirty_work_pending() { crate::scanner_io::clear_dirty_usage_bucket("photos"); } +#[test] +#[serial] +fn finalizing_post_scan_observation_advances_partially_without_dirty_ack() { + crate::scanner_io::clear_dirty_usage_bucket("photos"); + crate::scanner_io::record_dirty_usage_bucket("photos"); + let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests(); + let observed = crate::scanner_io::ScannerCycleResult::new( + ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable), + Some(dirty_snapshot), + ) + .with_observational_snapshot_published(true); + + let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(observed, DataUsagePersistOutcome::Saved); + + assert_eq!(outcome, ScannerCycleOutcome::Partial); + assert!(acknowledgements.is_empty()); + assert!(crate::scanner_io::dirty_usage_buckets_pending()); + crate::scanner_io::clear_dirty_usage_bucket("photos"); +} + #[tokio::test] async fn scanner_cycle_keeps_remote_pending_acknowledgement() { let pending = remote_dirty_usage_acknowledgement_pending(7, 1, std::future::ready(Ok::(true))).await; @@ -4451,6 +4755,29 @@ fn data_usage_persist_wait_covers_cache_retries_and_backup() { crate::runtime_config::refresh_scanner_runtime_config_for_tests(); } +#[test] +fn scanner_publication_lease_budget_has_a_strict_ttl_boundary() { + let ttl = Duration::from_millis(SCANNER_PUBLICATION_LEASE_TTL_MS); + + assert!(scanner_publication_lease_budget_allows_persistence( + ttl.saturating_sub(Duration::from_millis(1)) + )); + assert!(!scanner_publication_lease_budget_allows_persistence(ttl)); + assert!(!scanner_publication_lease_budget_allows_persistence(ttl + Duration::from_millis(1))); + assert_eq!( + ScannerCycleDeferReason::PublicationLeaseBudgetExceeded.as_str(), + "publication_lease_budget_exceeded" + ); + assert_eq!( + ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded.as_str(), + "publication_lease_deadline_exceeded" + ); + assert_eq!( + ScannerCycleDeferReason::PublicationLeaseReleaseFailed.as_str(), + "publication_lease_release_failed" + ); +} + #[tokio::test] async fn data_usage_persist_wait_aborts_when_scanner_is_cancelled() { let ctx = CancellationToken::new(); diff --git a/crates/scanner/src/scanner/usage_store.rs b/crates/scanner/src/scanner/usage_store.rs index a32e0cecf..23bfe60d7 100644 --- a/crates/scanner/src/scanner/usage_store.rs +++ b/crates/scanner/src/scanner/usage_store.rs @@ -40,6 +40,84 @@ pub(super) struct DataUsagePersistBaseline { pub(super) revision: DataUsageCacheRevision, } +/// Read the bytes used as the baseline for a usage publication while keeping +/// the v2 primary revision as the CAS fence. During an interrupted upgrade the +/// primary can be valid JSON without a baseline identity; in that case a +/// same-or-newer durable companion may still be used, but an older legacy +/// snapshot must not cross the primary's epoch fence. +pub(super) async fn read_data_usage_persist_baseline( + storeapi: Arc, +) -> Result { + let (primary, revision) = read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()).await?; + let Some(primary) = primary else { + for path in [ + format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()), + LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(), + format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()), + ] { + let (candidate, _) = read_config_with_revision(storeapi.clone(), &path).await?; + let Some(candidate) = candidate else { + continue; + }; + let Ok(usage) = serde_json::from_slice::(&candidate) else { + continue; + }; + if data_usage_info_has_persisted_baseline_identity(&usage) { + return Ok(DataUsagePersistBaseline { + data: Some(Bytes::from(candidate)), + revision, + }); + } + } + return Ok(DataUsagePersistBaseline { data: None, revision }); + }; + + let Ok(primary_info) = serde_json::from_slice::(&primary) else { + // Preserve the original bytes and revision. A completed scan may + // replace the invalid primary under this CAS fence; an observation + // will still reject it below because it has no verifiable identity. + return Ok(DataUsagePersistBaseline { + data: Some(Bytes::from(primary)), + revision, + }); + }; + if data_usage_info_has_persisted_baseline_identity(&primary_info) || data_usage_info_is_bootstrap_pending(&primary_info) { + return Ok(DataUsagePersistBaseline { + data: Some(Bytes::from(primary)), + revision, + }); + } + + let invalid_primary_epoch = primary_info.scanner_epoch; + for path in [ + format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()), + LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(), + format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()), + ] { + let (candidate, _) = read_config_with_revision(storeapi.clone(), &path).await?; + let Some(candidate) = candidate else { + continue; + }; + let Ok(usage) = serde_json::from_slice::(&candidate) else { + continue; + }; + let candidate_epoch = usage.scanner_epoch.unwrap_or_default(); + if data_usage_info_has_persisted_baseline_identity(&usage) + && invalid_primary_epoch.is_none_or(|epoch| candidate_epoch >= epoch) + { + return Ok(DataUsagePersistBaseline { + data: Some(Bytes::from(candidate)), + revision, + }); + } + } + + Ok(DataUsagePersistBaseline { + data: Some(Bytes::from(primary)), + revision, + }) +} + /// Short-lived publication inputs captured for one usage persistence attempt. /// Keeping the movement epoch, lease deadline, and target fence together makes /// it explicit that they are one proof rather than independent options. @@ -225,7 +303,7 @@ where data_usage_info.scanner_epoch = Some(leader_epoch); } if remote_lease_expired(remote_lease_deadline) { - outcome = DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable); + outcome = DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded); break 'updates; } if let Some(expected_epoch) = expected_publication_epoch @@ -281,8 +359,8 @@ where publication_epoch = Some(read_epoch); let authoritative_data = match next_baseline.as_ref() { Some(baseline) => baseline.data.clone(), - None => match read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()).await { - Ok((data, _)) => data.map(Bytes::from), + None => match read_data_usage_persist_baseline(storeapi.clone()).await { + Ok(baseline) => baseline.data, Err(err) => { error!( target: "rustfs::scanner", @@ -497,7 +575,7 @@ where break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement); } if remote_lease_expired(remote_lease_deadline) { - break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable); + break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded); } let done_save = Metrics::time(Metric::SaveUsage); @@ -510,7 +588,7 @@ where }; if remote_lease_expired(remote_lease_deadline) { done_save(); - break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable); + break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded); } save_config_shared_with_preconditions_and_lease_fence( storeapi.clone(), diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 8753b2139..e1abbf474 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -269,6 +269,10 @@ fn should_publish_usage_snapshot(status: ScannerCycleStatus) -> bool { matches!(status, ScannerCycleStatus::Complete | ScannerCycleStatus::Superseded) } +fn should_publish_observational_snapshot(status: ScannerCycleStatus) -> bool { + matches!(status, ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable)) +} + fn prepare_usage_snapshot_for_publication( status: ScannerCycleStatus, mut data_usage_info: DataUsageInfo, @@ -570,6 +574,17 @@ pub(crate) async fn scanner_set_disk_inventory(set: &SetDisks) -> Vec> pub(crate) enum ScannerCycleDeferReason { ActivityBaselineUnavailable, DataMovement, + /// The configured persistence budget cannot fit within the fixed remote + /// publication-lease TTL. This is a deterministic configuration/contract + /// mismatch, not evidence that a peer activity probe failed. + PublicationLeaseBudgetExceeded, + /// A granted lease's absolute deadline cannot cover the persistence + /// operation. This can occur even when the configured budget fits the + /// nominal TTL because lease acquisition consumed part of the window. + PublicationLeaseDeadlineExceeded, + /// A remote lease could not be released after the persistence attempt. + /// Keep the cycle deferred because the peer may still admit movement. + PublicationLeaseReleaseFailed, } impl ScannerCycleDeferReason { @@ -577,6 +592,9 @@ impl ScannerCycleDeferReason { match self { Self::ActivityBaselineUnavailable => "activity_baseline_unavailable", Self::DataMovement => "data_movement", + Self::PublicationLeaseBudgetExceeded => "publication_lease_budget_exceeded", + Self::PublicationLeaseDeadlineExceeded => "publication_lease_deadline_exceeded", + Self::PublicationLeaseReleaseFailed => "publication_lease_release_failed", } } } @@ -611,6 +629,7 @@ fn scanner_activity_preflight( pub(crate) struct ScannerCycleResult { pub(crate) status: ScannerCycleStatus, publication_epoch: Option, + observational_snapshot_published: bool, dirty_usage_clear: Option, remote_dirty_usage_acknowledgements: Vec, remote_publication_lease_targets: Vec<(String, String, u64)>, @@ -624,6 +643,7 @@ impl ScannerCycleResult { Self { status, publication_epoch: None, + observational_snapshot_published: false, dirty_usage_clear, remote_dirty_usage_acknowledgements: Vec::new(), remote_publication_lease_targets: Vec::new(), @@ -642,6 +662,15 @@ impl ScannerCycleResult { self.publication_epoch } + pub(crate) fn with_observational_snapshot_published(mut self, published: bool) -> Self { + self.observational_snapshot_published = published; + self + } + + pub(crate) fn has_observational_snapshot(&self) -> bool { + self.observational_snapshot_published + } + fn with_failed_dirty_usage(mut self, failed_dirty_usage: bool) -> Self { self.failed_dirty_usage = failed_dirty_usage; self diff --git a/crates/scanner/src/scanner_io/guards.rs b/crates/scanner/src/scanner_io/guards.rs index 52a3db927..3a1ad0f60 100644 --- a/crates/scanner/src/scanner_io/guards.rs +++ b/crates/scanner/src/scanner_io/guards.rs @@ -286,6 +286,16 @@ pub(super) fn scanner_task_join_error(stage: &str, err: tokio::task::JoinError) mod tests { use super::*; use rustfs_scanner_contracts::metrics::{ScannerWorkSource, global_metrics}; + use tokio::sync::oneshot; + + fn active_bucket_drive_count(source: ScannerWorkSource, bucket: &str, drive: &str) -> u64 { + global_metrics() + .scanner_runtime_details_report() + .active_bucket_drive_scans + .into_iter() + .find(|active| active.source == source.as_str() && active.bucket == bucket && active.drive == drive) + .map_or(0, |active| active.count) + } #[test] fn bucket_drive_failure_guard_retires_active_scan_on_drop() { @@ -305,4 +315,85 @@ mod tests { .any(|active| active.source == source.as_str() && active.bucket == bucket && active.drive == drive) ); } + + #[tokio::test] + async fn bucket_drive_failure_guard_retires_active_scan_after_cancellation() { + let source = ScannerWorkSource::Usage; + let bucket = "__guard_cancel_lifecycle_test__"; + let drive = "/__guard_cancel_lifecycle_test__"; + global_metrics().record_scan_bucket_drive_start(source, bucket, drive); + + let cancellation = CancellationToken::new(); + let worker_cancellation = cancellation.clone(); + let worker = tokio::spawn(async move { + let mut guard = BucketDriveFailureGuard::new(source, bucket, drive); + worker_cancellation.cancelled().await; + guard.mark_not_failed(); + }); + + cancellation.cancel(); + worker.await.expect("cancelled scanner worker should finish"); + + assert_eq!(active_bucket_drive_count(source, bucket, drive), 0); + } + + #[tokio::test] + async fn bucket_drive_failure_guard_retires_active_scan_when_worker_is_aborted() { + let source = ScannerWorkSource::Bitrot; + let bucket = "__guard_abort_lifecycle_test__"; + let drive = "/__guard_abort_lifecycle_test__"; + global_metrics().record_scan_bucket_drive_start(source, bucket, drive); + + let (started_sender, started_receiver) = oneshot::channel(); + let worker = tokio::spawn(async move { + let _guard = BucketDriveFailureGuard::new(source, bucket, drive); + started_sender.send(()).expect("test should observe worker start"); + std::future::pending::<()>().await; + }); + started_receiver.await.expect("scanner worker should start"); + assert_eq!(active_bucket_drive_count(source, bucket, drive), 1); + + worker.abort(); + worker.await.expect_err("aborted scanner worker should report cancellation"); + + assert_eq!(active_bucket_drive_count(source, bucket, drive), 0); + } + + #[tokio::test] + async fn bucket_drive_failure_guards_track_overlapping_scans_independently() { + let source = ScannerWorkSource::Usage; + let bucket = "__guard_overlap_lifecycle_test__"; + let drive = "/__guard_overlap_lifecycle_test__"; + global_metrics().record_scan_bucket_drive_start(source, bucket, drive); + global_metrics().record_scan_bucket_drive_start(source, bucket, drive); + + let (first_release_sender, first_release_receiver) = oneshot::channel(); + let (second_release_sender, second_release_receiver) = oneshot::channel(); + let (first_started_sender, first_started_receiver) = oneshot::channel(); + let (second_started_sender, second_started_receiver) = oneshot::channel(); + let first = tokio::spawn(async move { + let _guard = BucketDriveFailureGuard::new(source, bucket, drive); + first_started_sender.send(()).expect("test should observe first worker start"); + first_release_receiver.await.expect("first worker should be released"); + }); + let second = tokio::spawn(async move { + let _guard = BucketDriveFailureGuard::new(source, bucket, drive); + second_started_sender + .send(()) + .expect("test should observe second worker start"); + second_release_receiver.await.expect("second worker should be released"); + }); + + first_started_receiver.await.expect("first scanner worker should start"); + second_started_receiver.await.expect("second scanner worker should start"); + assert_eq!(active_bucket_drive_count(source, bucket, drive), 2); + + first_release_sender.send(()).expect("first worker should be released"); + first.await.expect("first scanner worker should finish"); + assert_eq!(active_bucket_drive_count(source, bucket, drive), 1); + + second_release_sender.send(()).expect("second worker should be released"); + second.await.expect("second scanner worker should finish"); + assert_eq!(active_bucket_drive_count(source, bucket, drive), 0); + } } diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index a5c447be3..ce9236277 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -162,18 +162,18 @@ impl ScannerIOCycle for ECStore { dirty_usage_status, activity_status, ); - if !publish_usage_snapshot( - &updates, - status, - DataUsageInfo { - last_update: Some(SystemTime::now()), - scanner_cycle: Some(want_cycle), - usage_snapshot_complete: true, - ..Default::default() - }, - ) - .await? - { + let empty_usage = DataUsageInfo { + last_update: Some(SystemTime::now()), + scanner_cycle: Some(want_cycle), + usage_snapshot_complete: true, + ..Default::default() + }; + let observational_snapshot_published = if should_publish_observational_snapshot(status) { + publish_observational_snapshot(&updates, empty_usage).await? + } else { + publish_usage_snapshot(&updates, status, empty_usage).await? + }; + if !observational_snapshot_published { return Ok(ScannerCycleResult::new(status, None).with_publication_epoch(publication_epoch)); } if status == ScannerCycleStatus::Complete { @@ -188,6 +188,7 @@ impl ScannerIOCycle for ECStore { }; return Ok(ScannerCycleResult::new(status, dirty_usage_clear) .with_publication_epoch(publication_epoch) + .with_observational_snapshot_published(observational_snapshot_published) .with_remote_publication_lease_targets(remote_publication_lease_targets) .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements)); } @@ -437,13 +438,19 @@ impl ScannerIOCycle for ECStore { dirty_usage_status, activity_status, ); - if let Some((data_usage_info, _)) = completed_usage { - publish_usage_snapshot(&updates, cycle_status, data_usage_info).await?; + let observational_snapshot_published = if let Some((data_usage_info, _)) = completed_usage { + if should_publish_observational_snapshot(cycle_status) { + publish_observational_snapshot(&updates, data_usage_info).await? + } else { + publish_usage_snapshot(&updates, cycle_status, data_usage_info).await? + } } else if !ctx.is_cancelled() && let Some((data_usage_info, _)) = observational_usage { - publish_observational_snapshot(&updates, data_usage_info).await?; - } + publish_observational_snapshot(&updates, data_usage_info).await? + } else { + false + }; let dirty_usage_clear = should_clear_dirty_usage_snapshot( result.is_ok(), structurally_complete_snapshot, @@ -463,6 +470,7 @@ impl ScannerIOCycle for ECStore { }; Ok(ScannerCycleResult::new(cycle_status, dirty_usage_clear) .with_publication_epoch(publication_epoch) + .with_observational_snapshot_published(observational_snapshot_published) .with_remote_publication_lease_targets(remote_publication_lease_targets) .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements) .with_failed_dirty_usage(!failed_buckets.is_empty()) diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index ef0e98063..df86fdc1d 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -909,6 +909,47 @@ async fn structurally_complete_superseded_cycles_publish_without_claiming_conver ); } +#[tokio::test] +async fn post_scan_activity_failure_retains_complete_usage_as_observation() { + let (updates, mut receiver) = mpsc::channel(1); + let status = ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable); + + assert!(should_publish_observational_snapshot(status)); + assert!( + publish_observational_snapshot( + &updates, + DataUsageInfo { + last_update: Some(SystemTime::now()), + scanner_cycle: Some(7), + objects_total_count: 3, + objects_total_size: 12, + usage_snapshot_complete: true, + ..Default::default() + }, + ) + .await + .expect("post-scan activity failure should retain an observation") + ); + + let observed = receiver.recv().await.expect("observational update should be queued"); + assert!(!observed.usage_snapshot_complete); + assert!(observed.usage_snapshot_partial); + assert_eq!(observed.usage_snapshot_converged, Some(false)); + assert_eq!(observed.objects_total_count, 3); + assert_eq!(observed.objects_total_size, 12); +} + +#[test] +fn only_unverified_activity_allows_post_scan_observation() { + assert!(should_publish_observational_snapshot(ScannerCycleStatus::Deferred( + ScannerCycleDeferReason::ActivityBaselineUnavailable + ))); + assert!(!should_publish_observational_snapshot(ScannerCycleStatus::Deferred( + ScannerCycleDeferReason::DataMovement + ))); + assert!(!should_publish_observational_snapshot(ScannerCycleStatus::Incomplete)); +} + #[test] fn scanner_cycle_fails_closed_for_namespace_disappearance() { for activity_status in [ScannerCycleActivityStatus::Changed, ScannerCycleActivityStatus::Unchanged] {