From adb90fc6e1c30f5df8c9046183c2de060e86cb8f Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 21 Aug 2026 17:32:59 +0800 Subject: [PATCH] fix(scanner): defer usage publication during pool recovery (#6333) * fix(scanner): defer usage publication during pool recovery * fix(scanner): preserve metrics when publication is deferred * fix(scanner): route test types through storage boundary * fix(scanner): keep cache floor deferred during movement --- crates/ecstore/src/store/mod.rs | 84 ++++++++ crates/scanner/src/scanner.rs | 244 +++++++++++++++------- crates/scanner/src/scanner/tests.rs | 221 ++++++++++++++++++++ crates/scanner/src/scanner/usage_store.rs | 88 +++++++- crates/scanner/src/scanner_io/io_cycle.rs | 19 ++ crates/scanner/src/scanner_io/tests.rs | 41 +++- crates/scanner/src/storage_api.rs | 8 +- 7 files changed, 622 insertions(+), 83 deletions(-) diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index eb3b6ef1c..4b8a9cb71 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -343,6 +343,23 @@ impl ECStore { let (decommission, rebalance) = tokio::join!(self.is_decommission_running(), self.is_rebalance_started()); decommission || rebalance } + + /// Returns whether scanner metadata may still be hidden by a local + /// data-movement state. Terminal failed/canceled decommission entries + /// remain suspended until an operator clears or retries them, so they are + /// a publication barrier even after the worker has stopped. + pub async fn scanner_data_usage_publication_blocked(&self) -> bool { + if self.scanner_data_movement_active().await { + return true; + } + + let pool_meta = self.pool_meta.read().await; + pool_meta.pools.iter().any(|pool| { + pool.decommission + .as_ref() + .is_some_and(|info| !info.queued && (info.failed || info.canceled)) + }) + } } // impl Clone for ECStore { @@ -875,6 +892,7 @@ impl crate::storage_api_contracts::admin::StorageAdminApi for ECStore { #[cfg(test)] mod tests { use super::*; + use crate::core::pools::{PoolDecommissionInfo, PoolStatus}; use crate::layout::endpoints::{Endpoints, PoolEndpoints, SetupType}; use crate::runtime::global::reset_local_disk_test_state; use crate::runtime::sources::{clear_local_disk_id_map_for_test, local_disk_path_by_id}; @@ -911,6 +929,72 @@ mod tests { }) } + #[tokio::test] + async fn scanner_data_usage_publication_blocks_active_and_unqueued_terminal_decommission() { + let store = build_store_with_ctx(Arc::new(InstanceContext::new())); + let cases = [ + ( + "active", + PoolDecommissionInfo { + start_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }, + true, + ), + ( + "failed", + PoolDecommissionInfo { + failed: true, + ..Default::default() + }, + true, + ), + ( + "canceled", + PoolDecommissionInfo { + canceled: true, + ..Default::default() + }, + true, + ), + ( + "queued_failed", + PoolDecommissionInfo { + failed: true, + queued: true, + ..Default::default() + }, + false, + ), + ( + "complete", + PoolDecommissionInfo { + complete: true, + ..Default::default() + }, + false, + ), + ("idle", PoolDecommissionInfo::default(), false), + ]; + + for (name, decommission, expected) in cases { + *store.pool_meta.write().await = PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: format!("scanner-publication-{name}"), + last_update: OffsetDateTime::now_utc(), + decommission: Some(decommission), + }], + ..Default::default() + }; + assert_eq!( + store.scanner_data_usage_publication_blocked().await, + expected, + "unexpected scanner publication barrier state for {name}" + ); + } + } + // The object graph is the isolation carrier: two ECStore instances holding // distinct contexts report independent erasure state through their real // `&self` accessors — no cross-contamination. diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 8ae5c7b85..4db558a54 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1081,18 +1081,6 @@ async fn run_data_scanner_cycle( } }; let (sender, receiver) = mpsc::channel::(1); - let storeapi_clone = storeapi.clone(); - let ctx_clone = ctx.clone(); - let mut usage_persist_task = AbortOnDropHandle::new(tokio::spawn(async move { - store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( - ctx_clone, - storeapi_clone, - receiver, - Some(leader_epoch), - Some(usage_persist_baseline), - ) - .await - })); let done_cycle = Metrics::time(Metric::ScanCycle); let cycle_budget = ScannerCycleBudget::new(ctx, cycle_budget_config); @@ -1107,47 +1095,78 @@ async fn run_data_scanner_cycle( scan_mode, ) .await; + let publication_defer_reason = match &scan_result { + Ok(result) => final_data_usage_publication_defer_reason(storeapi.as_ref(), result.status).await, + Err(_) => Some(ScannerCycleDeferReason::ActivityBaselineUnavailable), + }; let budget_elapsed = cycle_budget.budget_elapsed() && !ctx.is_cancelled(); - let usage_persist_outcome = match wait_for_data_usage_persist_task(ctx, &mut usage_persist_task, usage_persist_timeout).await - { - DataUsagePersistTaskResult::Completed(outcome) => outcome, - DataUsagePersistTaskResult::JoinFailed(err) => { - error!( - target: "rustfs::scanner", - event = EVENT_SCANNER_PERSIST_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_RUNTIME, - cycle = cycle_info.current, - state = "usage_persist_task_failed", - error = %err, - "Scanner data usage persistence task failed" - ); - DataUsagePersistOutcome::Failed + let usage_persist_outcome = match publication_defer_reason { + Some(reason) => { + drop(receiver); + DataUsagePersistOutcome::Deferred(reason) } - DataUsagePersistTaskResult::Cancelled => { - debug!( - target: "rustfs::scanner", - event = EVENT_SCANNER_PERSIST_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_RUNTIME, - cycle = cycle_info.current, - state = "usage_persist_task_cancelled", - "Scanner data usage persistence task cancelled" - ); - DataUsagePersistOutcome::Failed - } - DataUsagePersistTaskResult::TimedOut => { - error!( - target: "rustfs::scanner", - event = EVENT_SCANNER_PERSIST_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_RUNTIME, - cycle = cycle_info.current, - timeout = ?usage_persist_timeout, - state = "usage_persist_task_timed_out", - "Scanner data usage persistence task timed out" - ); - DataUsagePersistOutcome::Failed + None => { + // ScannerIO emits its complete or observational update only after + // all set workers finish. Persist after the final activity fence; + // this also avoids blocking the scanner on a denied publication. + let storeapi_clone = storeapi.clone(); + let ctx_clone = ctx.clone(); + let route_probe_store = storeapi.clone(); + let mut usage_persist_task = AbortOnDropHandle::new(tokio::spawn(async move { + store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe( + ctx_clone, + storeapi_clone, + receiver, + Some(leader_epoch), + Some(usage_persist_baseline), + move || { + let storeapi = route_probe_store.clone(); + async move { storeapi.scanner_data_usage_publication_blocked().await } + }, + ) + .await + })); + match wait_for_data_usage_persist_task(ctx, &mut usage_persist_task, usage_persist_timeout).await { + DataUsagePersistTaskResult::Completed(outcome) => outcome, + DataUsagePersistTaskResult::JoinFailed(err) => { + error!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + cycle = cycle_info.current, + state = "usage_persist_task_failed", + error = %err, + "Scanner data usage persistence task failed" + ); + DataUsagePersistOutcome::Failed + } + DataUsagePersistTaskResult::Cancelled => { + debug!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + cycle = cycle_info.current, + state = "usage_persist_task_cancelled", + "Scanner data usage persistence task cancelled" + ); + DataUsagePersistOutcome::Failed + } + DataUsagePersistTaskResult::TimedOut => { + error!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + cycle = cycle_info.current, + timeout = ?usage_persist_timeout, + state = "usage_persist_task_timed_out", + "Scanner data usage persistence task timed out" + ); + DataUsagePersistOutcome::Failed + } + } } }; let unresolved_heal_work = global_metrics().current_scan_cycle_has_unresolved_heal_work(); @@ -1191,33 +1210,51 @@ async fn run_data_scanner_cycle( mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await; return ScannerCycleOutcome::Failed; } - if let Some(required_cycle) = scan_cycle_result.required_cycle_floor() { - warn!( - target: "rustfs::scanner", - event = EVENT_SCANNER_CYCLE_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_RUNTIME, - cycle = cycle_info.current, - required_cycle, - state = "cache_cycle_ahead", - "Scanner cycle is recovering to a newer durable cache generation" - ); - emit_scan_cycle_partial_with_source(cycle_start.elapsed(), ScanCyclePartialReason::Unknown, None); - return if persist_required_scanner_cycle_floor( - ctx, - storeapi.clone(), - cycle_info, - cycle_revision, - leader_epoch, - required_cycle, - &mut cycle_metrics_guard, - ) - .await - { - ScannerCycleOutcome::Partial - } else { - ScannerCycleOutcome::Failed - }; + match scanner_cycle_pre_commit_outcome(scan_cycle_result.required_cycle_floor(), &usage_persist_outcome) { + Some(ScannerCyclePreCommitOutcome::RecoverCacheCycle(required_cycle)) => { + warn!( + target: "rustfs::scanner", + event = EVENT_SCANNER_CYCLE_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + cycle = cycle_info.current, + required_cycle, + state = "cache_cycle_ahead", + "Scanner cycle is recovering to a newer durable cache generation" + ); + emit_scan_cycle_partial_with_source(cycle_start.elapsed(), ScanCyclePartialReason::Unknown, None); + return if persist_required_scanner_cycle_floor( + ctx, + storeapi.clone(), + cycle_info, + cycle_revision, + leader_epoch, + required_cycle, + &mut cycle_metrics_guard, + ) + .await + { + ScannerCycleOutcome::Partial + } else { + ScannerCycleOutcome::Failed + }; + } + Some(ScannerCyclePreCommitOutcome::Deferred(reason)) => { + info!( + target: "rustfs::scanner", + event = EVENT_SCANNER_CYCLE_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + cycle = cycle_info.current, + reason = reason.as_str(), + state = "deferred", + "Scanner cycle deferred before data usage publication" + ); + emit_scan_cycle_deferred(cycle_start.elapsed()); + mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await; + return ScannerCycleOutcome::Deferred(reason); + } + None => {} } if usage_persist_outcome == DataUsagePersistOutcome::Failed { error!( @@ -2000,6 +2037,56 @@ impl Drop for ScannerScanModeGuard { } } +async fn final_data_usage_publication_defer_reason( + storeapi: &ECStore, + status: ScannerCycleStatus, +) -> Option { + match status { + ScannerCycleStatus::Complete | ScannerCycleStatus::Superseded => { + if storeapi.scanner_data_usage_publication_blocked().await { + return Some(ScannerCycleDeferReason::DataMovement); + } + if status == ScannerCycleStatus::Complete { + let distributed = storeapi.setup_is_dist_erasure().await; + match probe_scanner_activity(storeapi, distributed).await { + Ok(snapshot) if scanner_activity_allows_usage_publication(&snapshot) => None, + Ok(_) => Some(ScannerCycleDeferReason::DataMovement), + Err(_) => Some(ScannerCycleDeferReason::ActivityBaselineUnavailable), + } + } else { + // A superseded cycle is explicitly observational and cannot + // replace the authoritative snapshot. It may still be + // persisted as a convergence baseline for the next cycle. + None + } + } + ScannerCycleStatus::Deferred(reason) => Some(reason), + // Incomplete cycles do not publish a usage snapshot. Keep the + // decision permissive so existing partial-cycle handling remains + // unchanged if a future scanner path emits a bookkeeping update. + ScannerCycleStatus::Incomplete => None, + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum ScannerCyclePreCommitOutcome { + RecoverCacheCycle(u64), + Deferred(ScannerCycleDeferReason), +} + +fn scanner_cycle_pre_commit_outcome( + required_cycle_floor: Option, + usage_persist_outcome: &DataUsagePersistOutcome, +) -> Option { + // Keep the publication barrier fail-closed: `.bloomcycle.bin` uses the + // same routed writer and its floor must remain pending while data movement + // hides the source pool. + match usage_persist_outcome { + DataUsagePersistOutcome::Deferred(reason) => Some(ScannerCyclePreCommitOutcome::Deferred(*reason)), + _ => required_cycle_floor.map(ScannerCyclePreCommitOutcome::RecoverCacheCycle), + } +} + fn scanner_cycle_completion_outcome( scan_status: ScannerCycleStatus, usage_persist_outcome: DataUsagePersistOutcome, @@ -2007,6 +2094,7 @@ fn scanner_cycle_completion_outcome( has_failed_dirty_usage: bool, ) -> ScannerCycleOutcome { match (scan_status, usage_persist_outcome) { + (_, DataUsagePersistOutcome::Deferred(reason)) => ScannerCycleOutcome::Deferred(reason), (_, DataUsagePersistOutcome::Failed) => ScannerCycleOutcome::Failed, (ScannerCycleStatus::Deferred(reason), DataUsagePersistOutcome::NoUpdate) if !has_dirty_usage && !has_failed_dirty_usage => diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 334e66f97..fd321bfc3 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -153,6 +153,7 @@ struct MemoryConfigStore { objects: Mutex>>, revisions: Mutex>, fail_put_number: Mutex>, + object_not_found_put_number: Mutex>, error_after_commit_put_number: Mutex>, interleaving_puts: Mutex)>>, cancel_after_interleaving_puts: Mutex>, @@ -224,6 +225,9 @@ impl crate::storage_api::scanner_io::ObjectIO for MemoryConfigStore { if self.fail_put_number.lock().await.get(&key) == Some(&put_count) { return Err(EcstoreError::other("injected put failure")); } + if self.object_not_found_put_number.lock().await.get(&key) == Some(&put_count) { + return Err(EcstoreError::ObjectNotFound(bucket.to_string(), object.to_string())); + } let interleaving_data = { let mut interleaving_puts = self.interleaving_puts.lock().await; @@ -1431,6 +1435,170 @@ async fn test_store_data_usage_in_backend_preserves_newer_snapshot() { assert_eq!(outcome, DataUsagePersistOutcome::Current); } +#[tokio::test] +async fn test_usage_save_object_not_found_defers_only_with_a_fresh_route_barrier() { + for (route_blocked, expected) in [ + (true, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement)), + (false, DataUsagePersistOutcome::Failed), + ] { + let store = Arc::new(MemoryConfigStore::default()); + let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()); + let baseline = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(10)), 1); + let baseline_data = serde_json::to_vec(&baseline).expect("baseline usage snapshot should encode"); + store.objects.lock().await.insert(key.clone(), baseline_data.clone()); + store.revisions.lock().await.insert(key.clone(), 1); + store.object_not_found_put_number.lock().await.insert(key.clone(), 1); + + let (sender, receiver) = mpsc::channel(1); + sender + .send(complete_usage_with_bucket_count( + Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)), + 2, + )) + .await + .expect("new usage snapshot should enqueue"); + drop(sender); + let probe_calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let route_probe_calls = probe_calls.clone(); + + let outcome = store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe( + CancellationToken::new(), + store.clone(), + receiver, + None, + Some(DataUsagePersistBaseline { + data: Some(Bytes::from(baseline_data.clone())), + revision: DataUsageCacheRevision::Etag("memory-1".to_string()), + }), + move || { + let probe_calls = route_probe_calls.clone(); + async move { + let call = probe_calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + route_blocked && call > 1 + } + }, + ) + .await; + + assert_eq!(outcome, expected); + assert_eq!( + probe_calls.load(std::sync::atomic::Ordering::SeqCst), + 3, + "ObjectNotFound must be followed by a fresh route-barrier probe" + ); + assert_eq!( + store.objects.lock().await.get(&key), + Some(&baseline_data), + "a route failure must not replace the authoritative baseline" + ); + } +} + +#[tokio::test] +async fn test_usage_save_route_barrier_prevents_missing_snapshot_creation() { + for observational in [false, true] { + let store = Arc::new(MemoryConfigStore::default()); + let target_path = if observational { + DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str() + } else { + DATA_USAGE_OBJ_NAME_PATH.as_str() + }; + let target_key = memory_config_key(RUSTFS_META_BUCKET, target_path); + let mut incoming = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)), 1); + incoming.usage_snapshot_converged = Some(!observational); + let (sender, receiver) = mpsc::channel(1); + sender.send(incoming).await.expect("usage snapshot 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, + Some(DataUsagePersistBaseline { + data: None, + revision: DataUsageCacheRevision::Missing, + }), + || async { true }, + ) + .await; + + assert_eq!(outcome, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement)); + assert!(!store.objects.lock().await.contains_key(&target_key)); + assert_eq!( + store.put_counts.lock().await.get(&target_key), + None, + "the final pool-state fence must run before the first PUT" + ); + } +} + +#[tokio::test] +#[serial] +async fn test_usage_route_barrier_precedes_durable_reconciliation() { + let store = Arc::new(MemoryConfigStore::default()); + let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()); + let snapshot = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)), 1); + let snapshot_data = serde_json::to_vec(&snapshot).expect("usage snapshot should encode"); + let (sender, receiver) = mpsc::channel(1); + sender.send(snapshot).await.expect("usage snapshot 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, + Some(DataUsagePersistBaseline { + data: Some(Bytes::from(snapshot_data)), + revision: DataUsageCacheRevision::Etag("memory-1".to_string()), + }), + || async { true }, + ) + .await; + + assert_eq!(outcome, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement)); + assert_eq!(store.put_counts.lock().await.get(&key), None); +} + +#[tokio::test] +#[serial] +async fn test_deferred_usage_save_keeps_last_real_save_metric() { + let metrics = global_metrics(); + metrics.record_scanner_usage_save_result(ScannerUsageSaveResult::Success); + let before = metrics.report().await.usage_freshness; + + 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 outcome = store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe( + CancellationToken::new(), + store, + receiver, + None, + Some(DataUsagePersistBaseline { + data: None, + revision: DataUsageCacheRevision::Missing, + }), + || async { true }, + ) + .await; + + assert_eq!(outcome, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement)); + let after = metrics.report().await.usage_freshness; + assert_eq!(after.last_usage_save_result, before.last_usage_save_result); + assert_eq!(after.last_usage_save_result_code, before.last_usage_save_result_code); + assert_eq!(after.last_usage_save_unix_secs, before.last_usage_save_unix_secs); +} + #[tokio::test] async fn test_store_data_usage_in_backend_fences_interleaving_newer_writer() { let store = Arc::new(MemoryConfigStore::default()); @@ -2325,6 +2493,15 @@ async fn test_store_data_usage_in_backend_reports_missing_snapshot() { #[test] fn test_scanner_cycle_completion_prioritizes_persist_failure() { + assert_eq!( + scanner_cycle_completion_outcome( + ScannerCycleStatus::Complete, + DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement), + true, + false, + ), + ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::DataMovement) + ); assert_eq!( scanner_cycle_completion_outcome( ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable), @@ -2421,6 +2598,33 @@ fn test_scanner_cycle_completion_prioritizes_persist_failure() { ); } +#[test] +fn scanner_cycle_cache_floor_stays_pending_during_deferred_usage_publication() { + for reason in [ + ScannerCycleDeferReason::DataMovement, + ScannerCycleDeferReason::ActivityBaselineUnavailable, + ] { + let deferred = DataUsagePersistOutcome::Deferred(reason); + assert_eq!( + scanner_cycle_pre_commit_outcome(Some(19), &deferred), + Some(ScannerCyclePreCommitOutcome::Deferred(reason)), + "a blocked publication must not persist the routed scanner cycle floor" + ); + assert_eq!( + scanner_cycle_pre_commit_outcome(None, &deferred), + Some(ScannerCyclePreCommitOutcome::Deferred(reason)) + ); + } + assert_eq!( + scanner_cycle_pre_commit_outcome(Some(19), &DataUsagePersistOutcome::Saved), + Some(ScannerCyclePreCommitOutcome::RecoverCacheCycle(19)) + ); + assert_eq!( + scanner_cycle_pre_commit_outcome(Some(19), &DataUsagePersistOutcome::Failed), + Some(ScannerCyclePreCommitOutcome::RecoverCacheCycle(19)) + ); +} + #[test] #[serial] fn finalizing_a_saved_cycle_acknowledges_its_exact_dirty_snapshot() { @@ -2448,6 +2652,23 @@ fn finalizing_a_saved_cycle_acknowledges_its_exact_dirty_snapshot() { assert!(!crate::scanner_io::dirty_usage_buckets_pending()); } +#[test] +#[serial] +fn finalizing_a_deferred_usage_save_keeps_dirty_work_pending() { + 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 deferred = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot)); + + let (outcome, _, acknowledgements) = + finalize_scanner_cycle_result(deferred, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement)); + + assert_eq!(outcome, ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::DataMovement)); + 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; diff --git a/crates/scanner/src/scanner/usage_store.rs b/crates/scanner/src/scanner/usage_store.rs index 44ff2e8ec..17a550295 100644 --- a/crates/scanner/src/scanner/usage_store.rs +++ b/crates/scanner/src/scanner/usage_store.rs @@ -22,6 +22,10 @@ pub(super) enum DataUsagePersistOutcome { AlreadyDurable, PriorCycleDurable, Saved, + /// The metadata route is temporarily unavailable (for example while a + /// terminal decommission state keeps the source pool suspended). The + /// caller must retry without acknowledging dirty usage. + Deferred(ScannerCycleDeferReason), Failed, } @@ -92,10 +96,33 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch( pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( ctx: CancellationToken, storeapi: Arc, - mut receiver: mpsc::Receiver, + receiver: mpsc::Receiver, leader_epoch: Option, initial_baseline: Option, ) -> DataUsagePersistOutcome { + store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe( + ctx, + storeapi, + receiver, + leader_epoch, + initial_baseline, + || async { false }, + ) + .await +} + +pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe( + ctx: CancellationToken, + storeapi: Arc, + mut receiver: mpsc::Receiver, + leader_epoch: Option, + initial_baseline: Option, + route_probe: F, +) -> DataUsagePersistOutcome +where + F: Fn() -> Fut + Send + Sync, + Fut: Future + Send, +{ let mut outcome = DataUsagePersistOutcome::NoUpdate; let mut next_baseline = initial_baseline; @@ -113,6 +140,19 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel } else { DATA_USAGE_OBJ_NAME_PATH.as_str() }; + if route_probe().await { + debug!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + path = %target_path, + state = "publication_blocked_before_reconcile", + "Scanner data usage publication deferred by the pool-state fence" + ); + outcome = DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement); + break; + } if observational && data_usage_info.usage_snapshot_authoritative_baseline.is_none() { let authoritative_data = match next_baseline.as_ref() { @@ -275,6 +315,18 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel if ctx.is_cancelled() { break 'updates; } + if route_probe().await { + debug!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + path = %target_path, + state = "publication_blocked_before_save", + "Scanner data usage publication deferred by the final pool-state fence" + ); + break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement); + } let done_save = Metrics::time(Metric::SaveUsage); let save_result = save_config_shared_with_preconditions( @@ -313,6 +365,33 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel "Scanner data usage CAS conflict will be reconciled" ); } + Err(e @ EcstoreError::ObjectNotFound(_, _)) => { + let route_blocked = route_probe().await; + if route_blocked { + warn!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + path = %target_path, + state = "publication_deferred", + error = %e, + "Scanner data usage route is blocked by data movement; retrying later" + ); + break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement); + } + error!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + path = %target_path, + state = "save_failed", + error = %e, + "Scanner data usage save failed" + ); + break DataUsagePersistOutcome::Failed; + } Err(e) => { error!( target: "rustfs::scanner", @@ -370,6 +449,13 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel outcome = DataUsagePersistOutcome::Failed; continue; } + DataUsagePersistOutcome::Deferred(reason) => { + // A deferred publication is an intentional retryable state, not a + // failed save. Keep the last real save result so admin freshness + // reporting does not turn a pool-recovery fence into a false error. + outcome = DataUsagePersistOutcome::Deferred(reason); + break 'updates; + } DataUsagePersistOutcome::Saved => { if observational { invalidate_admin_data_usage_snapshot_cache().await; diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index ac803d280..c2a8eb253 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -49,6 +49,25 @@ impl ScannerIOCycle for ECStore { ) -> Result { let child_token = ctx.child_token(); + // Check the local pool metadata before listing buckets. A failed or + // canceled decommission remains suspended after its worker exits, so + // starting a scan in that state could build a snapshot that cannot be + // routed to the authoritative metadata object. + if self.scanner_data_usage_publication_blocked().await { + debug!( + target: "rustfs::scanner::io", + event = EVENT_SCANNER_SET_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_IO, + state = "cycle_data_usage_route_blocked", + "Scanner cycle deferred while data usage metadata remains hidden by data movement" + ); + return Ok(ScannerCycleResult::new( + ScannerCycleStatus::Deferred(ScannerCycleDeferReason::DataMovement), + None, + )); + } + let distributed = self.setup_is_dist_erasure().await; let activity_before = match scanner_activity_preflight(crate::scanner::probe_scanner_activity(self, distributed).await) { ScannerActivityPreflight::Ready(snapshot) => snapshot, diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index f7a7c9e73..e6ef3bafb 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -17,7 +17,9 @@ use super::io_disk::tier_stats_template; use super::*; use crate::scanner_budget::ScannerCycleBudgetConfig; use crate::scanner_folder::ScannerItem; -use crate::storage_api::owner::{EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats}; +use crate::storage_api::owner::{ + EcstorePoolDecommissionInfo, EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats, +}; use crate::storage_api::scan::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions, ObjectIO as _}; use crate::{ DiskOption, ECStore, Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, ScannerObjectOptions, @@ -182,6 +184,39 @@ async fn scanner_cycle_is_deferred_while_rebalance_is_active() { assert!(receiver.recv().await.is_none(), "rebalance-deferred cycle must not publish usage"); } +#[tokio::test] +#[serial] +async fn scanner_cycle_is_deferred_while_terminal_decommission_is_blocked() { + let (_temp_dir, store) = setup_two_pool_scanner_store().await; + for decommission in [ + EcstorePoolDecommissionInfo { + failed: true, + ..Default::default() + }, + EcstorePoolDecommissionInfo { + canceled: true, + ..Default::default() + }, + ] { + store.pool_meta.write().await.pools[0].decommission = Some(decommission); + assert!(store.scanner_data_usage_publication_blocked().await); + + let ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default()); + let (updates, mut receiver) = mpsc::channel(1); + let result = tokio::time::timeout( + Duration::from_secs(30), + ScannerIOCycle::nsscanner_with_status(store.as_ref(), ctx, budget, updates, 1, 1, HealScanMode::Normal), + ) + .await + .expect("terminal-decommission-deferred scanner cycle should finish") + .expect("terminal-decommission-deferred scanner cycle should succeed"); + + assert_eq!(result.status, ScannerCycleStatus::Deferred(ScannerCycleDeferReason::DataMovement)); + assert!(receiver.recv().await.is_none(), "blocked cycle must not publish usage"); + } +} + #[tokio::test] async fn data_usage_publish_fails_when_receiver_is_closed() { let (updates, receiver) = mpsc::channel(1); @@ -236,6 +271,10 @@ async fn multi_pool_scanner_cycle_publishes_combined_usage() { assert_eq!(bucket_usage.size, 11); assert_eq!(usage.objects_total_count, 2); assert_eq!(usage.objects_total_size, 11); + assert!( + receiver.recv().await.is_none(), + "a scanner cycle must publish at most one terminal usage snapshot" + ); } #[tokio::test] diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index d8779a0aa..b5bde363e 100644 --- a/crates/scanner/src/storage_api.rs +++ b/crates/scanner/src/storage_api.rs @@ -47,6 +47,8 @@ pub(crate) use rustfs_ecstore::api::bucket::versioning_sys::BucketVersioningSys pub(crate) use rustfs_ecstore::api::cache::{ ListPathRawOptions as EcstoreListPathRawOptions, list_path_raw as ecstore_list_path_raw, }; +#[cfg(test)] +pub(crate) use rustfs_ecstore::api::capacity::PoolDecommissionInfo as EcstorePoolDecommissionInfo; pub(crate) use rustfs_ecstore::api::capacity::{ is_reserved_or_invalid_bucket as ecstore_is_reserved_or_invalid_bucket, path2_bucket_object as ecstore_path2_bucket_object, path2_bucket_object_with_base_path as ecstore_path2_bucket_object_with_base_path, @@ -127,9 +129,9 @@ pub(crate) mod owner { #[cfg(test)] pub(crate) use super::{ EcstoreDiskOption, EcstoreDiskStore, EcstoreEndpoint, EcstoreEndpointServerPools, EcstoreEndpoints, - EcstoreInstanceContext, EcstorePoolEndpoints, EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, - EcstoreRebalanceStats, ecstore_config_init, ecstore_init_bucket_metadata_sys, ecstore_init_local_disks_with_instance_ctx, - ecstore_new_disk, + EcstoreInstanceContext, EcstorePoolDecommissionInfo, EcstorePoolEndpoints, EcstoreRebalStatus, EcstoreRebalanceInfo, + EcstoreRebalanceMeta, EcstoreRebalanceStats, ecstore_config_init, ecstore_init_bucket_metadata_sys, + ecstore_init_local_disks_with_instance_ctx, ecstore_new_disk, }; }