fix(scanner): preserve unknown catch-up state on ledger failure (#8877)

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.

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::<Vec<_>>();
+        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::<HashSet<_>>();
+        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<Option<Arc<CoordinatorBuc
 /// its disk. The hook is absent from non-test builds and a no-op until a test
 /// installs a probe; the probe then holds the walker so overlapping
 /// admissions become observable as a peak.
-pub(super) async fn observe_coordinator_bucket_scan_admitted(_bucket: &str) {
+pub(super) async fn observe_coordinator_bucket_scan_admitted(worker_index: usize, bucket: &str) {
+    collector_lifecycle::observe_bucket_admission(worker_index, bucket).await;
     let probe = COORDINATOR_BUCKET_SCAN_PROBE
         .lock()
         .unwrap_or_else(std::sync::PoisonError::into_inner)
@@ -4981,6 +5020,10 @@ pub(super) async fn observe_coordinator_bucket_scan_admitted(_bucket: &str) {
     probe.active.fetch_sub(1, Ordering::SeqCst);
 }
 
+pub(super) async fn observe_coordinator_bucket_worker_started(worker_index: usize) {
+    collector_lifecycle::observe_worker_start(worker_index).await;
+}
+
 fn install_coordinator_bucket_scan_probe(hold: Duration) -> Arc<CoordinatorBucketScanProbe> {
     let probe = Arc::new(CoordinatorBucketScanProbe {
         active: AtomicUsize::new(0),

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<Option<Arc<Probe>>> = StdMutex::new(None);
+
+struct ProbeGuard(Arc<Probe>);
+
+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<ECStore>, 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<SetDisks>, names: &[&str], cycle: u64) -> ScannerBucketScanPlan {
+    let buckets = names.iter().map(|name| bucket_info(name)).collect::<Vec<_>>();
+    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;
+}

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]