diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index f85907713..670f93887 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -787,7 +787,11 @@ impl ScannerIOCache for SetDisks { NamespaceScannerWorkerMode::RemoteV4(server_epoch) => Some(server_epoch), NamespaceScannerWorkerMode::Coordinator => None, }; + #[cfg(test)] + let worker_index = futs.len(); futs.push(AbortOnDropHandle::new(tokio::spawn(async move { + #[cfg(test)] + super::tests::observe_coordinator_bucket_worker_started(worker_index).await; let remote_session_id = uuid::Uuid::new_v4(); let mut remote_session_sequence = 0_u64; loop { @@ -1056,7 +1060,7 @@ impl ScannerIOCache for SetDisks { None }; #[cfg(test)] - super::tests::observe_coordinator_bucket_scan_admitted(&bucket.name).await; + super::tests::observe_coordinator_bucket_scan_admitted(worker_index, &bucket.name).await; // Lock order: scanner leader fence -> set-scoped per-bucket cache lock -> // cache object read/write. diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index a1fd568d4..c733a1371 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -45,6 +45,7 @@ use time::OffsetDateTime; use uuid::Uuid; mod checkpoint_coverage_identity; +mod collector_lifecycle; mod listing_window_scope; mod partial_bucket_carry; mod per_bucket_certification; @@ -4646,9 +4647,13 @@ fn cache_root_entry_info_rejects_missing_or_dangling_roots() { #[tokio::test] async fn completed_bucket_results_count_distinct_current_roots_even_after_cancellation() { - for cancelled in [false, true] { + const BUCKETS: usize = 53; + for (cancel_after, fill_missing) in [None, Some(0), Some(26)] + .into_iter() + .flat_map(|cancel_after| [false, true].map(|fill_missing| (cancel_after, fill_missing))) + { let ctx = CancellationToken::new(); - if cancelled { + if cancel_after == Some(0) { ctx.cancel(); } let mut cache = DataUsageCache { @@ -4667,27 +4672,60 @@ async fn completed_bucket_results_count_distinct_current_roots_even_after_cancel ..Default::default() }, ); - let (tx, rx) = mpsc::channel(3); - for (bucket, generation, size) in [("complete", 7, 1), ("complete", 7, 2), ("missing", 8, 11)] { - tx.send(DataUsageEntryInfo { - name: bucket.to_string(), - parent: DATA_USAGE_ROOT.to_string(), - entry: DataUsageEntry { - size, - ..Default::default() - }, - bucket_incarnation: None, - tier_registry_generation: Some(generation), - }) + let (tx, rx) = mpsc::channel(1); + let collector = + AbortOnDropHandle::new(tokio::spawn(super::io_cache::collect_completed_bucket_results(ctx.clone(), cache, rx))); + let mut entries = (0..BUCKETS - 1) + .map(|index| (format!("bucket-{index:02}"), 7, 1)) + .collect::>(); + entries.extend([("bucket-00".to_string(), 7, 2), ("missing".to_string(), 8, 11)]); + if fill_missing { + entries.push(("missing".to_string(), 7, 5)); + } + for (index, (bucket, generation, size)) in entries.into_iter().enumerate() { + if cancel_after == Some(index) { + ctx.cancel(); + } + tokio::time::timeout( + Duration::from_secs(5), + tx.send(DataUsageEntryInfo { + name: bucket, + parent: DATA_USAGE_ROOT.to_string(), + entry: DataUsageEntry { + size, + ..Default::default() + }, + bucket_incarnation: None, + tier_registry_generation: Some(generation), + }), + ) .await + .expect("collector must continue receiving after cancellation") .expect("fixture result channel"); } drop(tx); - let (cache, complete) = super::io_cache::collect_completed_bucket_results(ctx, cache, rx).await; - assert_eq!(complete, HashSet::from(["complete".to_string()])); - assert!(!should_publish_completed_snapshot(complete.len(), 2, false, false)); - assert_eq!(cache.find("complete").expect("applied root").size, 2); - assert_eq!(cache.find("missing").expect("retained old root").size, 3); + let (cache, complete) = tokio::time::timeout(Duration::from_secs(5), collector) + .await + .expect("collector must drain the closed channel after cancellation") + .expect("collector should join"); + let mut expected = (0..BUCKETS - 1) + .map(|index| format!("bucket-{index:02}")) + .collect::>(); + if fill_missing { + expected.insert("missing".to_string()); + } + assert_eq!(complete, expected, "duplicate and wrong-tier roots cannot fill the missing bucket"); + assert_eq!(complete.len(), if fill_missing { 53 } else { 52 }); + assert_eq!(should_publish_completed_snapshot(complete.len(), BUCKETS, false, false), fill_missing); + assert!( + !should_publish_completed_snapshot(complete.len(), BUCKETS, false, true), + "drained coverage must not override cancellation" + ); + assert_eq!(cache.find("bucket-00").expect("applied duplicate root").size, 2); + assert_eq!( + cache.find("missing").expect("retained or completed root").size, + if fill_missing { 5 } else { 3 } + ); } } @@ -4969,7 +5007,8 @@ static COORDINATOR_BUCKET_SCAN_PROBE: std::sync::Mutex Arc { let probe = Arc::new(CoordinatorBucketScanProbe { active: AtomicUsize::new(0), diff --git a/crates/scanner/src/scanner_io/tests/collector_lifecycle.rs b/crates/scanner/src/scanner_io/tests/collector_lifecycle.rs new file mode 100644 index 000000000..d0e85505f --- /dev/null +++ b/crates/scanner/src/scanner_io/tests/collector_lifecycle.rs @@ -0,0 +1,356 @@ +// Copyright 2026 RustFS Team +// SPDX-License-Identifier: Apache-2.0 + +use super::*; +use crate::data_usage_define::{DATA_USAGE_OBJ_NAME_PATH, read_config_with_revision}; + +const HEALTHY_BUCKET: &str = "collector-b-healthy"; +const FAULT_BUCKET: &str = "collector-a-fault"; +const PARTIAL_BUCKET: &str = "collector-z-partial"; + +#[derive(Clone, Copy)] +enum Stop { + WorkerPanic, + ParentCancellation, +} + +struct Probe { + stop: Stop, + fault_entered: CancellationToken, + partial_entered: CancellationToken, + release_fault: CancellationToken, + fault_exited: CancellationToken, + release_partial: CancellationToken, + partial_exited: CancellationToken, +} + +static PROBE: StdMutex>> = StdMutex::new(None); + +struct ProbeGuard(Arc); + +impl ProbeGuard { + fn install(stop: Stop) -> Self { + let probe = Arc::new(Probe { + stop, + fault_entered: CancellationToken::new(), + partial_entered: CancellationToken::new(), + release_fault: CancellationToken::new(), + fault_exited: CancellationToken::new(), + release_partial: CancellationToken::new(), + partial_exited: CancellationToken::new(), + }); + let mut slot = PROBE.lock().expect("collector lifecycle probe lock"); + assert!(slot.is_none(), "collector lifecycle fixtures must run serially"); + *slot = Some(probe.clone()); + Self(probe) + } +} + +impl Drop for ProbeGuard { + fn drop(&mut self) { + self.0.release_fault.cancel(); + self.0.release_partial.cancel(); + *PROBE.lock().expect("collector lifecycle probe cleanup") = None; + clear_dirty_usage_buckets_for_tests(); + } +} + +struct AdmissionExit(CancellationToken); + +impl Drop for AdmissionExit { + fn drop(&mut self) { + self.0.cancel(); + } +} + +pub(super) async fn observe_worker_start(worker_index: usize) { + let probe = PROBE.lock().expect("collector lifecycle probe lock").clone(); + let Some(probe) = probe else { return }; + if worker_index != 0 { + probe.fault_entered.cancelled().await; + } +} + +pub(super) async fn observe_bucket_admission(worker_index: usize, bucket: &str) { + let probe = PROBE.lock().expect("collector lifecycle probe lock").clone(); + let Some(probe) = probe else { return }; + match bucket { + FAULT_BUCKET => { + assert_eq!(worker_index, 0, "the first waited handle must own the injected failure"); + let _exit = AdmissionExit(probe.fault_exited.clone()); + probe.fault_entered.cancel(); + probe.release_fault.cancelled().await; + if matches!(probe.stop, Stop::WorkerPanic) { + panic!("injected collector lifecycle worker panic"); + } + } + PARTIAL_BUCKET => { + assert_eq!(worker_index, 1, "the second worker must retain the admitted resume"); + let _exit = AdmissionExit(probe.partial_exited.clone()); + probe.partial_entered.cancel(); + probe.release_partial.cancelled().await; + } + HEALTHY_BUCKET => assert_eq!(worker_index, 1, "the second worker must send the healthy root first"), + _ => {} + } +} + +async fn create_bucket(store: &Arc, name: &str, objects: usize) { + store + .make_bucket(name, &MakeBucketOptions::default()) + .await + .expect("create collector fixture bucket"); + for index in 0..objects { + store + .put_object( + name, + &format!("prefix-{index}/object"), + &mut ScannerPutObjReader::from_vec(b"collector".to_vec()), + &ScannerObjectOptions::default(), + ) + .await + .expect("persist collector fixture object"); + } + wait_for_namespace_commit_tails(store.as_ref()).await; +} + +async fn scan_plan(set: &Arc, names: &[&str], cycle: u64) -> ScannerBucketScanPlan { + let buckets = names.iter().map(|name| bucket_info(name)).collect::>(); + let source = DataUsageCacheSource::new(set.pool_index, set.set_index); + let mut cohort = ScannerServiceCohort::default(); + cohort.refresh(&HashMap::from([(source, buckets.clone())])); + let digest = scanner_bucket_plan_digest(&buckets, [7; 32]); + let tier_registry_generation = runtime_tier_registry_for_cycle(cycle, 11).await.generation; + ScannerBucketScanPlan { + all_buckets: Arc::new(buckets.clone()), + buckets, + scope: ScannerBucketScanScope::default(), + digest, + bucket_coverage_digest: digest, + checkpoint_coverage_digest: digest, + requires_full_scan: true, + service_cohort: Some(Arc::new(StdMutex::new(cohort))), + execution_digest: digest, + leader_epoch: 11, + tier_registry_generation, + publication_epoch: Some( + scanner_publication_epoch(set.clone()) + .await + .expect("fixture publication admission"), + ), + dirty_usage_buckets: Arc::new(dirty_usage_buckets_for_tests()), + bucket_failures: ScannerBucketFailureState::default(), + pending_maintenance_work: Arc::new(AtomicBool::new(false)), + cache_cycle_floor: Arc::new(AtomicU64::new(cycle)), + cold_zero_walk_reuse_observed: Arc::new(AtomicBool::new(false)), + segment_invalidation_proof: None, + } +} + +async fn assert_collector_lifecycle(stop: Stop) { + let (_dir, store) = setup_local_scanner_store(1, 1).await; + clear_dirty_usage_buckets_for_tests(); + create_bucket(&store, PARTIAL_BUCKET, 4).await; + let set = store.pools[0].disk_set[0].clone(); + let seed_parent = CancellationToken::new(); + let seed_budget = ScannerCycleBudget::new_with_progress_tracking( + &seed_parent, + ScannerCycleBudgetConfig { + max_objects: Some(1), + ..Default::default() + }, + ); + let (updates, mut seed_results) = mpsc::channel(1); + set.clone() + .nsscanner_cache( + seed_budget.token(), + seed_budget.clone(), + scan_plan(&set, &[PARTIAL_BUCKET], 1).await, + updates, + 1, + HealScanMode::Normal, + ) + .await + .expect("budgeted production walk must save partial progress"); + assert_eq!(seed_budget.progress().0, 1, "the checkpoint must come from actual object work"); + assert!(!seed_results.recv().await.expect("partial set result").info.snapshot_complete); + assert!(seed_results.recv().await.is_none()); + let partial_name = path_join_buf(&[PARTIAL_BUCKET, DATA_USAGE_CACHE_NAME]); + let mut partial = DataUsageCache::default(); + let partial_revision = partial + .load_with_revisions(set.clone(), &partial_name) + .await + .expect("read the checkpoint saved by the production worker"); + assert!(!partial.info.snapshot_complete); + assert!(partial.info.scan_identity.is_some()); + assert!(partial.has_scan_position(), "the fixture must retain a real interrupted position"); + let saved_partial = serde_json::to_value(&partial).expect("checkpoint comparison value"); + + create_bucket(&store, HEALTHY_BUCKET, 1).await; + create_bucket(&store, FAULT_BUCKET, 1).await; + for bucket in [HEALTHY_BUCKET, FAULT_BUCKET, PARTIAL_BUCKET] { + record_dirty_usage_bucket(bucket); + } + let mut set_caches_before = Vec::new(); + for name in [DATA_USAGE_CACHE_NAME.to_string(), format!("{DATA_USAGE_CACHE_NAME}.bkp")] { + let path = path_join_buf(&[crate::BUCKET_META_PREFIX, &name]); + let before = read_config_with_revision(set.clone(), &path) + .await + .expect("set cache before interrupted cycle"); + set_caches_before.push((path, before)); + } + let dirty_before = dirty_usage_buckets_for_tests(); + let root_before = read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) + .await + .expect("usage root before interrupted cycle"); + let parent = CancellationToken::new(); + let budget = ScannerCycleBudget::new_with_progress_tracking(&parent, ScannerCycleBudgetConfig::default()); + let (updates, mut results) = mpsc::channel(1); + let guard = ProbeGuard::install(stop); + let probe = &guard.0; + let scan = set.clone().nsscanner_cache( + budget.token(), + budget.clone(), + scan_plan(&set, &[HEALTHY_BUCKET, FAULT_BUCKET, PARTIAL_BUCKET], 2).await, + updates, + 2, + HealScanMode::Normal, + ); + tokio::pin!(scan); + tokio::select! { + result = &mut scan => panic!("set finished before both workers reached their gates: {result:?}"), + _ = async { + probe.fault_entered.cancelled().await; + probe.partial_entered.cancelled().await; + } => {} + } + // Worker zero owns the first waited handle and is held at FAULT_BUCKET. + // Worker one starts only after that admission, then reaches PARTIAL_BUCKET + // after HEALTHY_BUCKET's root send completed in the production loop. + let mut healthy = DataUsageCache::default(); + healthy + .load(set.clone(), &path_join_buf(&[HEALTHY_BUCKET, DATA_USAGE_CACHE_NAME])) + .await + .expect("healthy root must be durable before the injected stop"); + assert!(healthy.info.snapshot_complete); + assert_eq!(healthy.checked_flatten(HEALTHY_BUCKET).expect("healthy root").objects, 1); + assert_eq!(budget.progress().0, 1); + assert!(!parent.is_cancelled(), "cancellation must occur during an active scan"); + assert!(matches!(results.try_recv(), Err(mpsc::error::TryRecvError::Empty))); + + if matches!(stop, Stop::ParentCancellation) { + parent.cancel(); + } + probe.release_fault.cancel(); + tokio::select! { + result = &mut scan => panic!("set returned before the other admitted worker exited: {result:?}"), + _ = probe.fault_exited.cancelled() => {} + } + // These current-thread Tokio tests observe the unwind after the panic + // task has yielded back to the executor. The set must still join the + // other worker, whose gate has not been released. + assert!(futures::poll!(&mut scan).is_pending(), "set must await its still-admitted worker"); + assert!(!probe.partial_exited.is_cancelled()); + parent.cancel(); + probe.release_partial.cancel(); + let result = scan.await; + assert!( + probe.partial_exited.is_cancelled(), + "the remaining worker must leave before the set returns" + ); + assert!(budget.reason().is_none(), "this stop must come from the parent, not a budget deadline"); + match stop { + Stop::WorkerPanic => { + let error = result.expect_err("worker panic must remain a set join failure"); + assert!(error.to_string().contains("scanner disk bucket task join failed")); + assert!(error.to_string().contains("injected collector lifecycle worker panic")); + assert!(results.recv().await.is_none(), "failed set must not emit complete or partial authority"); + } + Stop::ParentCancellation => { + result.expect("cancelled workers must still drain the accepted roots"); + let drained = results + .recv() + .await + .expect("cancelled set must retain its incomplete observation"); + assert!(!drained.info.snapshot_complete); + assert!(drained.info.last_update.is_none()); + assert_eq!( + drained + .find(HEALTHY_BUCKET) + .expect("collector retained the accepted root") + .objects, + 1 + ); + assert!(results.recv().await.is_none(), "set output must close after its collector is joined"); + } + } + for (path, before) in set_caches_before { + assert_eq!( + read_config_with_revision(set.clone(), &path) + .await + .expect("set cache after worker shutdown"), + before, + "interrupted workers must not publish a complete set snapshot or advance its revision: {path}" + ); + } + assert_eq!( + partial + .load_with_revisions(set, &partial_name) + .await + .expect("reload partial after worker shutdown"), + partial_revision, + "stopping an admitted resume must not replace the durable checkpoint" + ); + assert_eq!(serde_json::to_value(&partial).expect("checkpoint comparison value"), saved_partial); + // These global checks guard side effects of the set entry point only. + // cycle_budget_publication covers the outer publication and ACK proof. + assert_eq!( + dirty_usage_buckets_for_tests(), + dirty_before, + "set scan must not mutate global dirty buckets" + ); + assert_eq!( + read_config_with_revision(store, DATA_USAGE_OBJ_NAME_PATH.as_str()) + .await + .expect("usage root after interrupted cycle"), + root_before, + "set scan must not replace global usage as a side effect" + ); +} + +struct RuntimeConfigReset; + +impl Drop for RuntimeConfigReset { + fn drop(&mut self) { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); + } +} + +async fn run_collector_lifecycle(stop: Stop) { + let _reset = RuntimeConfigReset; + temp_env::async_with_vars( + [ + (ENV_SCANNER_WALKERS_PER_DISK, Some("2")), + (ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, Some("2")), + ], + async { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); + tokio::time::timeout(Duration::from_secs(60), assert_collector_lifecycle(stop)) + .await + .expect("collector lifecycle fixture must finish within its existing bounded deadline"); + }, + ) + .await; +} + +#[tokio::test] +#[serial] +async fn collector_worker_panic_waits_for_admitted_resume_and_preserves_checkpoint() { + run_collector_lifecycle(Stop::WorkerPanic).await; +} + +#[tokio::test] +#[serial] +async fn collector_parent_cancel_drains_accepted_root_and_preserves_checkpoint() { + run_collector_lifecycle(Stop::ParentCancellation).await; +} diff --git a/rustfs/src/admin/handlers/scanner.rs b/rustfs/src/admin/handlers/scanner.rs index a90e4f3c1..15946332b 100644 --- a/rustfs/src/admin/handlers/scanner.rs +++ b/rustfs/src/admin/handlers/scanner.rs @@ -128,8 +128,10 @@ fn scanner_catch_up_estimate( backlog: &rustfs_scanner::ScannerPauseBacklogStatus, metrics: &ScannerMetricsReport, ) -> ScannerCatchUpEstimate { + // A confirmed missing ledger is empty; a failed read does not establish that. + let backlog_observed = backlog.error.is_none() && (backlog.durable || backlog.persistence_state == "missing"); ScannerCatchUpEstimate { - estimated: pause.paused || backlog.phase != rustfs_scanner::ScannerPauseBacklogPhase::Idle, + estimated: !backlog_observed || pause.paused || backlog.phase != rustfs_scanner::ScannerPauseBacklogPhase::Idle, movement_work_items: pause.movement_backlog_work_items.max(backlog.movement_work_items), dirty_usage_buckets: metrics.usage_freshness.dirty_pending_buckets.max(backlog.dirty_usage_buckets), discovered_expiry_items: metrics @@ -144,7 +146,7 @@ fn scanner_catch_up_estimate( .saturating_add(metrics.lifecycle_transition.compensation_pending) .saturating_add(metrics.lifecycle_transition.compensation_running) .max(backlog.discovered_transition_items), - undiscovered_ilm_items_known: !pause.paused && !backlog.pending_full_scan, + undiscovered_ilm_items_known: backlog_observed && !pause.paused && !backlog.pending_full_scan, usage_baseline_unix_secs: metrics.usage_freshness.last_durable_success_unix_secs, } } @@ -824,7 +826,10 @@ mod tests { rustfs_scanner::scanner_runtime_config_status(), rustfs_scanner::ScannerCycleScheduleStatus::default(), ScannerDataMovementPauseStatus::default(), - rustfs_scanner::ScannerPauseBacklogStatus::default(), + rustfs_scanner::ScannerPauseBacklogStatus { + persistence_state: "missing".to_string(), + ..Default::default() + }, ); let encoded = serde_json::to_value(response).expect("scanner status should serialize"); @@ -865,6 +870,70 @@ mod tests { vec![rustfs_scanner::ScannerPauseBacklogAlertReason::PersistenceUnavailable] ); assert_eq!(backlog.error.as_deref(), Some("storage layer not initialized")); + + let scanner = scanner_status_response( + true, + ScannerMetricsReport::default(), + rustfs_scanner::scanner_runtime_config_status(), + rustfs_scanner::ScannerCycleScheduleStatus::default(), + ScannerDataMovementPauseStatus::default(), + backlog.clone(), + ); + let ilm = ilm_expiry_status_response( + true, + ScannerMetricsReport::default(), + rustfs_scanner::scanner_runtime_config_status(), + rustfs_scanner::ScannerCycleScheduleStatus::default(), + ScannerDataMovementPauseStatus::default(), + backlog, + ); + for encoded in [ + serde_json::to_value(scanner).expect("scanner status should serialize"), + serde_json::to_value(ilm).expect("ILM status should serialize"), + ] { + assert_eq!(encoded["catch_up_estimate"]["estimated"], true); + assert_eq!(encoded["catch_up_estimate"]["undiscovered_ilm_items_known"], false); + } + } + + #[test] + fn scanner_catch_up_estimate_requires_an_observed_pause_ledger() { + for (state, durable, error, observed) in [ + ("missing", false, None, true), + ("healthy", true, None, true), + ("membership_repair_pending", true, None, true), + ("committed_pending_stabilization", true, None, true), + ("rolled_back_partial_commit", true, None, true), + ("rolled_forward_partial_commit", true, None, true), + ("", false, None, false), + ("unavailable", false, Some("no readable ledger"), false), + ("degraded", false, Some("ledger claim failed"), false), + ("degraded", true, Some("ledger update failed"), false), + ] { + let backlog = rustfs_scanner::ScannerPauseBacklogStatus { + persistence_state: state.to_string(), + durable, + error: error.map(str::to_string), + discovered_expiry_items: 5, + ..Default::default() + }; + let metrics = ScannerMetricsReport { + lifecycle_expiry: ScannerLifecycleExpirySnapshot { + current_queued: 7, + current_active: 2, + ..Default::default() + }, + ..Default::default() + }; + let estimate = scanner_catch_up_estimate(&ScannerDataMovementPauseStatus::default(), &backlog, &metrics); + + assert_eq!(estimate.estimated, !observed, "ledger state {state:?}, durable={durable}"); + assert_eq!( + estimate.undiscovered_ilm_items_known, observed, + "ledger state {state:?}, durable={durable}" + ); + assert_eq!(estimate.discovered_expiry_items, 9, "keep the independently observed queue count"); + } } #[test]