test(e2e): retain startup and shutdown failure diagnostics

* fix(usage): reconcile stale counters after lifecycle expiration

* fix(usage): account lifecycle expiry during continuous writes

* test(usage): run lifecycle usage scenarios on one scanner store

* test(usage): use a Windows-representable pre-mutation offset

* fix(usage): harden expiry accounting recovery and quota checks

Borrow expiry receipt bucket names and avoid allocating a map key on cache hits. Cover cancelled receipts, durable snapshot recovery, and legacy quota admission after scanner confirmation. Use representable timestamp offsets in the quota regression.

Refs rustfs/backlog#2689

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(usage): recover stale persisted counts through the scanner

Seed incorrect complete usage for empty and retained-object buckets, then run the real scanner and publication consumer without further object mutations. Verify durable and admin usage over two cycles instead of writing a corrected snapshot in the test.

Refs rustfs/backlog#2689
Refs rustfs/backlog#2691

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(e2e): bound delimiter pagination fixture concurrency

The 120-second smoke timeout expired after 1018 of 1200 serial fixture PUTs, before LIST ran. Prepare the same objects with at most eight concurrent requests and await every PUT. Retain the timeout and strengthen exact prefix, KeyCount, empty Contents, and continuation-token assertions with phase diagnostics.

Refs rustfs/backlog#2689

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(upgrade): establish a persisted previous-release baseline

Seed the pinned previous-release cluster and restart it once with its data intact before replacing any node. Require every old writer to pass the strict readiness probe and preserve the seed through both mixed phases and the final current cluster. Keep InternalError fail-fast behavior and all existing compatibility deadlines and assertions.

Refs rustfs/backlog#2689
Refs rustfs/backlog#2384

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(ecstore): bound cancellation metadata persistence waits

Use the system-bucket incarnation boundary now supplied by main PR #8268. Bound the three cancellation waits that previously hung during pool.bin persistence, retaining their remote-generation, target-cohort, and durable-state assertions.

Refs rustfs/backlog#2697
Refs rustfs/backlog#2689

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: Chris <anzhengchao@gmail.com>
Co-authored-by: Hauser <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

.config/e2e-full-selection.txt

@@ -1,2 +1,2 @@
-sha256-darwin=3b6ff112ff16acbaa4148c4736095a1b6b71cc11af0dd3513ff28c6bdc334ca8
-sha256-linux=4e8bfc01870565a0a608a1634e6919fc593ce120a970aa04d974f7edf2675c21
+sha256-darwin=14c8cd76b8cbf643329c371daa70a5edd4357d42259a2fd9c653e548675d4a6d
+sha256-linux=80d0fae627007c31f250fb1fb21c8a952861195d53d7f60b6e71636d5ca65b46

Cargo.lock

@@ -14075,9 +14075,9 @@ dependencies = [
 
 [[package]]
 name = "yoke-derive"
-version = "0.8.3"
+version = "0.8.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "33811428bee40dbceb6d545e95754741d17a6aef9a4849f0fd62e2ba4f412a78"
+checksum = "ec8ebde2db3681e8c9980cc27822030e68752690ddfa9473e739aeb4dbde6d71"
 dependencies = [
  "proc-macro2",
  "quote",

crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -1450,8 +1450,11 @@ mod tests {
             cluster.set_env("RUSTFS_HEAL_PAGE_PARALLEL_ENABLE", "false");
         }
         // Keep every node's Heal runtime enabled for normal disk registration.
-        let server_rust_log = std::env::var("RUSTFS_HEAL_CHAOS_SERVER_RUST_LOG")
-            .unwrap_or_else(|_| "rustfs::heal::task=info,rustfs=error".to_string());
+        // Capture durable handoff and marker cleanup failures during shutdown.
+        let server_rust_log = std::env::var("RUSTFS_HEAL_CHAOS_SERVER_RUST_LOG").unwrap_or_else(|_| {
+            "rustfs::heal::task=info,rustfs::heal::manager=info,rustfs::main::handle_shutdown=info,rustfs_heal::heal=warn,rustfs=error"
+                .to_string()
+        });
         cluster.set_env("RUST_LOG", server_rust_log);
         let log_dir = std::env::var("RUSTFS_HEAL_CHAOS_LOG_DIR").unwrap_or_else(|_| {
             // Keep explicit directories compatible and isolate parallel suite captures.

crates/e2e_test/src/inline_fast_path_cluster_test.rs

@@ -886,6 +886,8 @@ fn configure_reader_metric_cluster(cluster: &mut RustFSTestClusterEnvironment, c
     cluster.set_env("RUSTFS_OBS_METRICS_EXPORT_ENABLED", "true");
     cluster.set_env("RUSTFS_OBS_TRACES_EXPORT_ENABLED", "false");
     cluster.set_env("RUSTFS_OBS_LOGS_EXPORT_ENABLED", "false");
+    // Keep startup diagnostics available when OTLP log export is disabled.
+    cluster.set_env("RUSTFS_OBS_LOG_STDOUT_ENABLED", "true");
     cluster.set_env("RUSTFS_OBS_METER_INTERVAL", "1");
     cluster.set_env("RUSTFS_OBS_USE_STDOUT", "false");
     cluster.set_env("RUSTFS_GET_CODEC_STREAMING_ENABLE", "false");

crates/e2e_test/src/list_objects_v2_pagination_test.rs

@@ -1029,6 +1029,8 @@ mod tests {
     /// tests the non-delimiter case.
     #[tokio::test]
     async fn test_list_objects_v2_maxkeys_above_limit_with_delimiter() {
+        use futures::{StreamExt, stream};
+
         init_logging();
         info!("Starting test: ListObjectsV2 MaxKeys above limit with delimiter");
 
@@ -1043,27 +1045,39 @@ mod tests {
         // 12 dirs × 100 files = 1200 raw keys
         let dir_count = 12;
         let files_per_dir = 100;
-        for d in 0..dir_count {
-            for f in 0..files_per_dir {
-                let key = format!("dir-{:02}/file{:03}.txt", d, f);
-                client
-                    .put_object()
-                    .bucket(bucket)
-                    .key(&key)
-                    .body(ByteStream::from_static(b"x"))
-                    .send()
-                    .await
-                    .expect("Failed to put object");
-            }
-        }
-
+        const FIXTURE_CONCURRENCY: usize = 8;
+        let fixture_started = std::time::Instant::now();
         eprintln!(
-            "Seeded {} objects in {bucket}; starting ListObjectsV2 pagination",
+            "Preparing {} objects with at most {FIXTURE_CONCURRENCY} concurrent PUTs",
             dir_count * files_per_dir
         );
+        // Object creation order is irrelevant to this listing contract. Bound
+        // fixture I/O and await every PUT before issuing the LIST request.
+        stream::iter((0..dir_count).flat_map(|d| (0..files_per_dir).map(move |f| (d, f))))
+            .for_each_concurrent(FIXTURE_CONCURRENCY, |(d, f)| {
+                let client = &client;
+                async move {
+                    let key = format!("dir-{d:02}/file{f:03}.txt");
+                    client
+                        .put_object()
+                        .bucket(bucket)
+                        .key(&key)
+                        .body(ByteStream::from_static(b"x"))
+                        .send()
+                        .await
+                        .unwrap_or_else(|err| panic!("Failed to prepare fixture object {key}: {err:?}"));
+                }
+            })
+            .await;
+        eprintln!(
+            "Seeded {} objects in {bucket} in {:?}; starting ListObjectsV2 pagination",
+            dir_count * files_per_dir,
+            fixture_started.elapsed()
+        );
         let deadline = Instant::now() + PAGINATION_TIMEOUT;
 
         // With delimiter: 12 CommonPrefixes visible, all fit within capped 1000
+        let list_started = std::time::Instant::now();
         let output =
             listing_request_before(deadline, client.list_objects_v2().bucket(bucket).delimiter("/").max_keys(2000).send())
                 .await
@@ -1077,6 +1091,15 @@ mod tests {
             dir_count,
             output.common_prefixes().len()
         );
+        let prefixes = output
+            .common_prefixes()
+            .iter()
+            .map(|entry| entry.prefix().expect("each CommonPrefix must carry its prefix").to_owned())
+            .collect::<Vec<_>>();
+        let expected_prefixes = (0..dir_count).map(|d| format!("dir-{d:02}/")).collect::<Vec<_>>();
+        assert_eq!(prefixes, expected_prefixes);
+        assert!(output.contents().is_empty(), "all fixture objects must collapse into prefixes");
+        assert_eq!(output.key_count(), Some(i32::try_from(dir_count).expect("fixture count fits i32")));
         assert_eq!(output.max_keys(), Some(1000));
         // 12 visible < 1000 capped MaxKeys → not truncated
         assert!(
@@ -1085,6 +1108,11 @@ mod tests {
             output.common_prefixes().len()
         );
 
+        assert!(
+            output.next_continuation_token().is_none(),
+            "a complete page must not carry a continuation token"
+        );
+        eprintln!("ListObjectsV2 assertions completed in {:?}", list_started.elapsed());
         info!("MaxKeys above limit with delimiter test passed");
 
         env.stop_server();

crates/e2e_test/src/upgrade_compatibility_test.rs

@@ -52,6 +52,9 @@ const VERSIONED_BUCKET: &str = "upgrade-versioned-data";
 const MIXED_BUCKET: &str = "upgrade-mixed-version-data";
 const MIXED_NODE_COUNT: usize = 4;
 const UPGRADE_READINESS_BODY: &[u8] = b"upgrade write readiness";
+// Written by the "previous-seed" readiness probe through node 0 before any
+// current binary starts; every later phase must still read it.
+const PREVIOUS_RELEASE_SEED_KEY: &str = ".upgrade-readiness/previous-seed/node-0";
 const MULTIPART_WORKERS: usize = 16;
 const MULTIPART_UPLOADS_PER_WORKER: usize = 16;
 // Peers keep a restarted node's drive in Suspect/Returning for roughly
@@ -68,6 +71,7 @@ const CONFIG_REPLICA_BUCKET: &str = "upgrade-config-replica";
 const ROLLBACK_BUCKET: &str = "rollback-config-data";
 const ROLLBACK_REPLICA_BUCKET: &str = "rollback-config-replica";
 const BUCKET_QUOTA_BYTES: u64 = 64 * 1024 * 1024;
+const QUOTA_USAGE_READINESS_TIMEOUT: Duration = Duration::from_secs(90);
 const LIFECYCLE_RULE_ID: &str = "upgrade-expire-logs";
 const LIFECYCLE_PREFIX: &str = "logs/";
 const LIFECYCLE_DAYS: i32 = 30;
@@ -425,6 +429,11 @@ async fn exercise_mixed_cluster(
     let multipart_keys = write_multipart_load(&clients, phase).await?;
     let expected_count = multipart_keys.len() + 2;
     for (label, client) in [("current", current_client), ("previous", previous_client)] {
+        assert_eq!(
+            read_object(client, MIXED_BUCKET, PREVIOUS_RELEASE_SEED_KEY, None).await?.1,
+            UPGRADE_READINESS_BODY,
+            "{phase}: the {label} node must retain the previous-release seed"
+        );
         wait_for_phase_listing(
             client,
             phase,
@@ -523,9 +532,7 @@ async fn prepare_previous_release_baseline(cluster: &mut RustFSTestClusterEnviro
     wait_for_upgrade_write_readiness(&clients, "previous-baseline", LISTING_CONVERGENCE_TIMEOUT).await?;
     for (reader, client) in clients.iter().enumerate() {
         assert_eq!(
-            read_object(client, MIXED_BUCKET, ".upgrade-readiness/previous-seed/node-0", None)
-                .await?
-                .1,
+            read_object(client, MIXED_BUCKET, PREVIOUS_RELEASE_SEED_KEY, None).await?.1,
             UPGRADE_READINESS_BODY,
             "previous-release node {reader} must retain the seed across its rolling restart"
         );
@@ -935,6 +942,11 @@ async fn rolling_upgrade_from_rc2_preserves_mixed_version_contracts() -> TestRes
     cluster.start_node_from_binary(3, &current_binary).await?;
 
     for (node_idx, client) in cluster.create_all_clients()?.iter().enumerate() {
+        assert_eq!(
+            read_object(client, MIXED_BUCKET, PREVIOUS_RELEASE_SEED_KEY, None).await?.1,
+            UPGRADE_READINESS_BODY,
+            "current node {node_idx} must retain the previous-release seed"
+        );
         for phase in ["one-current-node", "one-previous-node"] {
             wait_for_phase_listing(
                 client,
@@ -962,7 +974,8 @@ async fn rolling_upgrade_from_rc2_preserves_mixed_version_contracts() -> TestRes
                 "assertions": [
                     "current node reads objects written through previous-release client",
                     "previous-release node reads objects written through current client",
-                    "all nodes list every mixed-version object after homogeneous-current convergence"
+                    "all nodes list every mixed-version object after homogeneous-current convergence",
+                    "the previous-release seed survives the baseline restart and every upgrade phase"
                 ],
             }),
         )?;
@@ -1067,12 +1080,23 @@ async fn put_object_through_quota_warmup(client: &Client, bucket: &str, key: &st
 
 async fn get_bucket_quota(env: &RustFSTestEnvironment, bucket: &str) -> Result<Option<u64>, BoxError> {
     let path = format!("/rustfs/admin/v3/quota/{bucket}");
-    let (status, response) = admin_request(&env.url, Method::GET, &path, None, &env.access_key, &env.secret_key).await?;
-    if status != StatusCode::OK {
-        return Err(format!("reading the quota of {bucket} failed: {status} {response}").into());
+    let deadline = Instant::now() + QUOTA_USAGE_READINESS_TIMEOUT;
+    loop {
+        let (status, response) = tokio::time::timeout_at(
+            deadline,
+            admin_request(&env.url, Method::GET, &path, None, &env.access_key, &env.secret_key),
+        )
+        .await
+        .map_err(|_| format!("reading the quota of {bucket} exceeded the usage-readiness timeout"))??;
+        if status == StatusCode::OK {
+            let quota: serde_json::Value = serde_json::from_str(&response)?;
+            return Ok(quota.get("quota").and_then(serde_json::Value::as_u64));
+        }
+        if status != StatusCode::SERVICE_UNAVAILABLE || Instant::now() >= deadline {
+            return Err(format!("reading the quota of {bucket} failed: {status} {response}").into());
+        }
+        sleep(Duration::from_millis(500)).await;
     }
-    let quota: serde_json::Value = serde_json::from_str(&response)?;
-    Ok(quota.get("quota").and_then(serde_json::Value::as_u64))
 }
 
 /// `GET /rustfs/admin/v3/list-remote-targets?bucket=...`.

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -5606,6 +5606,11 @@ async fn apply_expiry_on_non_transitioned_objects_with_lock_lost_signal(
 
     //debug!("lc_event.action: {:?}", lc_event.action);
     debug!("expiry_on_non_transitioned_objects opts: {:?}", opts);
+    let usage_accounting = if lc_event.action == IlmAction::DeleteAction && !versioned && !version_suspended && !oi.is_dir {
+        Some(crate::data_usage::begin_expiry_usage_accounting(&oi.bucket).await)
+    } else {
+        None
+    };
     let mut dobj = match api
         .delete_object_with_tier_delete_journal(&oi.bucket, &encode_dir_object(&oi.name), opts)
         .await
@@ -5625,6 +5630,14 @@ async fn apply_expiry_on_non_transitioned_objects_with_lock_lost_signal(
             return false;
         }
     };
+    if let Some(accounting) = usage_accounting
+        && !dobj.name.is_empty()
+        && !dobj.delete_marker
+        && !dobj.is_dir
+        && let Ok(size) = crate::data_usage::quota_object_size(&dobj)
+    {
+        accounting.commit(size).await;
+    }
     schedule_lifecycle_replication_delete_if_needed(oi, &dobj).await;
 
     // The object (or all its versions, for delete_all) was expired; evict any
@@ -12891,6 +12904,130 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn ordinary_expiry_accounts_the_committed_receipt_once() {
+        use crate::bucket::quota::{QuotaOperation, checker::QuotaChecker};
+        use crate::data_usage::{apply_bucket_usage_memory_overlay, replace_bucket_usage_memory_from_info};
+        use rustfs_data_usage::{BucketUsageInfo, DataUsageInfo};
+
+        let (_disk_paths, ecstore) = setup_test_env().await;
+        for size in [0usize, 42] {
+            let bytes = u64::try_from(size).expect("fixture size fits");
+            let bucket = format!("expiry-accounting-{}", Uuid::new_v4().simple());
+            create_test_bucket(&ecstore, &bucket).await;
+            let mut reader = PutObjReader::from_vec(vec![0; size]);
+            let mut queued = ecstore
+                .put_object(&bucket, "expire", &mut reader, &ObjectOptions::default())
+                .await
+                .expect("create expiring object");
+            let mut reader = PutObjReader::from_vec(vec![0; size]);
+            ecstore
+                .put_object(&bucket, "keep", &mut reader, &ObjectOptions::default())
+                .await
+                .expect("create retained object");
+            let mut baseline = DataUsageInfo {
+                last_update: Some(std::time::SystemTime::now()),
+                usage_snapshot_complete: true,
+                buckets_count: 1,
+                ..Default::default()
+            };
+            baseline.buckets_usage.insert(
+                bucket.clone(),
+                BucketUsageInfo {
+                    objects_count: 2,
+                    versions_count: 2,
+                    size: bytes * 2,
+                    ..Default::default()
+                },
+            );
+            baseline.bucket_sizes.insert(bucket.clone(), bytes * 2);
+            baseline.calculate_totals();
+            replace_bucket_usage_memory_from_info(&baseline).await;
+            // Legacy quota admission consumes the same conservative overlay.
+            // Durable reservations have their own storage-commit accounting.
+            metadata_sys::update(
+                &bucket,
+                rustfs_config::QUOTA_CONFIG_FILE,
+                format!(r#"{{"quota":{},"quota_type":"Hard"}}"#, bytes * 2).into_bytes(),
+            )
+            .await
+            .expect("configure the legacy quota");
+            let checker =
+                QuotaChecker::new(Arc::new(tokio::sync::RwLock::new(metadata_sys::BucketMetadataSys::new(ecstore.clone()))));
+            let attempted_size = bytes.max(1);
+            let quota = checker
+                .check_quota(&bucket, QuotaOperation::PutObject, attempted_size)
+                .await
+                .expect("check quota before expiration");
+            assert!(!quota.allowed);
+            assert!(!quota.uses_durable_reservations);
+            assert_eq!(quota.current_usage, Some(bytes * 2));
+            // Queued metadata is not the accounting receipt from the storage commit.
+            queued.size = 1;
+            let event = lifecycle::Event {
+                action: IlmAction::DeleteAction,
+                ..Default::default()
+            };
+            let incarnation = ecstore
+                .bucket_incarnation_id_from_disk(&bucket)
+                .await
+                .expect("bucket incarnation");
+            for (attempt_incarnation, expected) in [(Uuid::new_v4(), false), (incarnation, true), (incarnation, false)] {
+                assert_eq!(
+                    super::apply_expiry_on_non_transitioned_objects(
+                        ecstore.clone(),
+                        &queued,
+                        &event,
+                        &LcEventSrc::Scanner,
+                        attempt_incarnation,
+                    )
+                    .await,
+                    expected
+                );
+                let mut response = baseline.clone();
+                apply_bucket_usage_memory_overlay(&mut response).await;
+                let remaining = if attempt_incarnation == incarnation { 1 } else { 2 };
+                assert_eq!(response.buckets_usage[&bucket].objects_count, remaining);
+                assert_eq!(response.buckets_usage[&bucket].size, remaining * bytes);
+                let quota = checker
+                    .check_quota(&bucket, QuotaOperation::PutObject, attempted_size)
+                    .await
+                    .expect("check quota before scanner confirmation");
+                assert!(!quota.allowed, "expiry must not release unconfirmed quota credit");
+                assert_eq!(quota.current_usage, Some(bytes * 2));
+            }
+            let mut confirmed = baseline.clone();
+            confirmed.last_update = Some(std::time::SystemTime::now());
+            confirmed.scanner_epoch = Some(7);
+            confirmed.scanner_cycle = Some(10);
+            let usage = confirmed.buckets_usage.get_mut(&bucket).expect("bucket is in the snapshot");
+            usage.size = bytes;
+            usage.objects_count = 1;
+            usage.versions_count = 0; // The scanner does not count unversioned objects as versions.
+            confirmed.bucket_sizes.insert(bucket.clone(), bytes);
+            confirmed.calculate_totals();
+            replace_bucket_usage_memory_from_info(&confirmed).await;
+            let quota = checker
+                .check_quota(&bucket, QuotaOperation::PutObject, attempted_size)
+                .await
+                .expect("check quota after scanner confirmation");
+            assert_eq!(quota.allowed, bytes > 0);
+            assert_eq!(quota.current_usage, Some(bytes));
+            let too_large = checker
+                .check_quota(&bucket, QuotaOperation::PutObject, bytes + 1)
+                .await
+                .expect("check the first byte above the quota limit");
+            assert!(!too_large.allowed);
+            assert!(
+                ecstore
+                    .get_object_info(&bucket, "keep", &ObjectOptions::default())
+                    .await
+                    .is_ok()
+            );
+        }
+    }
+
     #[tokio::test]
     #[serial]
     async fn queued_delete_all_rechecks_a_same_id_rule_moved_into_the_future() {

crates/ecstore/src/core/pools.rs

@@ -19858,9 +19858,9 @@ mod tests {
                 .reload_pool_meta()
                 .await
                 .expect("load the active generation on the remote node");
-            other_store
-                .decommission_cancel(0)
+            tokio::time::timeout(std::time::Duration::from_secs(30), other_store.decommission_cancel(0))
                 .await
+                .expect("remote cancel must persist pool metadata without re-entering its state locks")
                 .expect("the remote node should cancel the old generation");
             other_store
                 .clear_decommission(0)
@@ -20125,9 +20125,9 @@ mod tests {
             .expect("the stale operation should have a valid terminal fence plan")
             .expect("the stale operation should have an active reservation");
 
-        first_node
-            .decommission_cancel(0)
+        tokio::time::timeout(std::time::Duration::from_secs(30), first_node.decommission_cancel(0))
             .await
+            .expect("cancel must persist pool metadata without re-entering its state locks")
             .expect("cancel the first durable operation");
         first_node
             .clear_decommission(0)
@@ -20630,8 +20630,9 @@ mod tests {
             .await
             .expect("second target mutation task should not panic")
             .expect("second target mutation should finalize");
-        cancel
+        tokio::time::timeout(std::time::Duration::from_secs(30), cancel)
             .await
+            .expect("cancel must finish after target mutations release their gates")
             .expect("decommission cancellation task should not panic")
             .expect("decommission cancellation should finish after target mutations");
 

crates/ecstore/src/data_usage/mod.rs

@@ -43,7 +43,7 @@ use std::{
     future::Future,
     sync::{
         Arc, LazyLock, OnceLock,
-        atomic::{AtomicU64, Ordering},
+        atomic::{AtomicU64, AtomicUsize, Ordering},
     },
     time::{Duration, SystemTime},
 };
@@ -83,6 +83,79 @@ struct CachedBucketUsage {
     // necessarily a complete scanner reconciliation and therefore creates a
     // fresh cache entry with no pending hold.
     pending_negative_delta: u64,
+    pending_negative_updated_at: Option<SystemTime>,
+    pending_negative_reconciliation: Option<PendingDeleteReconciliation>,
+    in_flight_expirations: Arc<AtomicUsize>,
+}
+
+#[derive(Debug, Clone)]
+struct PendingDeleteReconciliation {
+    bytes: u64,
+    deleted_at: SystemTime,
+    observed: Option<((u64, u64), SystemTime)>,
+}
+
+impl CachedBucketUsage {
+    fn reconcile_pending_deletes(&mut self, position: Option<(u64, u64)>, updated_at: SystemTime) {
+        let Some(position) = position else { return };
+        let Some(pending) = self.pending_negative_reconciliation.as_mut() else { return };
+        match pending.observed {
+            None if updated_at >= pending.deleted_at => pending.observed = Some((position, updated_at)),
+            Some((previous, observed_at)) if position.0 >= previous.0 && position.1 > previous.1 && updated_at >= observed_at => {
+                self.pending_negative_delta = self.pending_negative_delta.saturating_sub(pending.bytes);
+                self.pending_negative_reconciliation = self
+                    .pending_negative_updated_at
+                    .filter(|_| self.pending_negative_delta > 0)
+                    .map(|deleted_at| PendingDeleteReconciliation {
+                        bytes: self.pending_negative_delta,
+                        deleted_at,
+                        observed: (updated_at >= deleted_at).then_some((position, updated_at)),
+                    });
+            }
+            _ => {}
+        }
+    }
+}
+
+/// Keeps a scanner refresh from including a delete before its memory receipt.
+pub(crate) struct ExpiryUsageAccounting<'a> {
+    bucket: &'a str,
+    in_flight: Arc<AtomicUsize>,
+}
+
+impl ExpiryUsageAccounting<'_> {
+    pub(crate) async fn commit(self, deleted_size: u64) {
+        let mut cache = memory_cache().write().await;
+        if let Some(entry) = cache.get_mut(self.bucket)
+            && Arc::ptr_eq(&entry.in_flight_expirations, &self.in_flight)
+        {
+            apply_bucket_object_delete_memory(entry, deleted_size, true);
+        }
+    }
+}
+
+impl Drop for ExpiryUsageAccounting<'_> {
+    fn drop(&mut self) {
+        self.in_flight.fetch_sub(1, Ordering::AcqRel);
+    }
+}
+
+pub(crate) async fn begin_expiry_usage_accounting(bucket: &str) -> ExpiryUsageAccounting<'_> {
+    ensure_bucket_usage_cached(bucket).await;
+    let mut cache = memory_cache().write().await;
+    // The caller owns the bucket name through commit. Avoid allocating either
+    // a receipt name or a map key for every expiry in an already cached bucket.
+    let in_flight = match cache.get(bucket) {
+        Some(entry) => Arc::clone(&entry.in_flight_expirations),
+        None => {
+            let entry = cached_bucket_usage_now(BucketUsageInfo::default());
+            let in_flight = Arc::clone(&entry.in_flight_expirations);
+            cache.insert(bucket.to_owned(), entry);
+            in_flight
+        }
+    };
+    in_flight.fetch_add(1, Ordering::AcqRel);
+    ExpiryUsageAccounting { bucket, in_flight }
 }
 
 type UsageMemoryCache = Arc<RwLock<HashMap<String, CachedBucketUsage>>>;
@@ -1945,6 +2018,9 @@ fn cached_bucket_usage_from_backend(usage: BucketUsageInfo, updated_at: SystemTi
         stale_snapshot_pending: false,
         pending_scanner_position: None,
         pending_negative_delta: 0,
+        pending_negative_updated_at: None,
+        pending_negative_reconciliation: None,
+        in_flight_expirations: Arc::new(AtomicUsize::new(0)),
     }
 }
 
@@ -1959,6 +2035,9 @@ fn cached_bucket_usage_now(usage: BucketUsageInfo) -> CachedBucketUsage {
         stale_snapshot_pending: false,
         pending_scanner_position: None,
         pending_negative_delta: 0,
+        pending_negative_updated_at: None,
+        pending_negative_reconciliation: None,
+        in_flight_expirations: Arc::new(AtomicUsize::new(0)),
     }
 }
 
@@ -2110,6 +2189,10 @@ pub async fn record_bucket_object_delete_memory(bucket: &str, deleted_size: u64,
         .entry(bucket.to_string())
         .or_insert_with(|| cached_bucket_usage_now(BucketUsageInfo::default()));
 
+    apply_bucket_object_delete_memory(entry, deleted_size, removed_current_object);
+}
+
+fn apply_bucket_object_delete_memory(entry: &mut CachedBucketUsage, deleted_size: u64, removed_current_object: bool) {
     entry.usage.size = entry.usage.size.saturating_sub(deleted_size);
     entry.pending_negative_delta = entry.pending_negative_delta.saturating_add(deleted_size);
     if removed_current_object {
@@ -2118,6 +2201,16 @@ pub async fn record_bucket_object_delete_memory(bucket: &str, deleted_size: u64,
     }
 
     let now = SystemTime::now();
+    entry.pending_negative_updated_at = Some(now);
+    // Freeze this tranche even before its first observation. Later deletes
+    // must not keep moving the timestamp that the scanner needs to catch up to.
+    if entry.pending_negative_reconciliation.is_none() && entry.pending_negative_delta > 0 {
+        entry.pending_negative_reconciliation = Some(PendingDeleteReconciliation {
+            bytes: entry.pending_negative_delta,
+            deleted_at: now,
+            observed: None,
+        });
+    }
     entry.refreshed_at = now;
     entry.usage_updated_at = now;
     entry.dirty = true;
@@ -2257,7 +2350,19 @@ async fn replace_bucket_usage_memory_from_info_if_generation(data_usage_info: &D
     if expected_generation.is_some_and(|expected| usage_memory_generation() != expected) {
         return;
     }
-    for (bucket, existing) in cache.iter() {
+    for (bucket, existing) in cache.iter_mut() {
+        existing.reconcile_pending_deletes(snapshot_position, usage_updated_at);
+        if existing.in_flight_expirations.load(Ordering::Acquire) > 0 {
+            next_cache.insert(bucket.clone(), existing.clone());
+            continue;
+        }
+        // Leadership fencing can advance the epoch without scanning. Only a
+        // later cycle can reconcile a dirty overlay that disagreed with the
+        // first complete observation after its last request mutation.
+        let reconciled_dirty_usage = existing
+            .pending_scanner_position
+            .zip(snapshot_position)
+            .is_some_and(|(previous, current)| current.0 >= previous.0 && current.1 > previous.1);
         match next_cache.entry(bucket.clone()) {
             Entry::Occupied(mut candidate) => {
                 if let Some(preserved) = preserve_unknown_dirty_usage(existing, snapshot_position, usage_updated_at) {
@@ -2275,7 +2380,10 @@ async fn replace_bucket_usage_memory_from_info_if_generation(data_usage_info: &D
                     candidate.insert(existing.clone());
                     continue;
                 }
-                if existing.authoritative && existing.dirty && !bucket_usage_counts_match(&existing.usage, &candidate.get().usage)
+                if existing.authoritative
+                    && existing.dirty
+                    && !reconciled_dirty_usage
+                    && !bucket_usage_counts_match(&existing.usage, &candidate.get().usage)
                 {
                     // A scanner snapshot can be saved after newer writes but still miss them if it listed the bucket earlier.
                     let mut preserved = existing.clone();
@@ -2285,8 +2393,10 @@ async fn replace_bucket_usage_memory_from_info_if_generation(data_usage_info: &D
                         // counts until they also converge, but release the
                         // conservative quota hold.
                         preserved.pending_negative_delta = 0;
+                        preserved.pending_negative_reconciliation = None;
                     }
                     preserved.stale_snapshot_pending = true;
+                    preserved.pending_scanner_position = preserved.pending_scanner_position.or(snapshot_position);
                     candidate.insert(preserved);
                 }
             }
@@ -2295,14 +2405,16 @@ async fn replace_bucket_usage_memory_from_info_if_generation(data_usage_info: &D
                     candidate.insert(preserved);
                 } else if existing.authoritative && existing.usage_updated_at > usage_updated_at {
                     candidate.insert(existing.clone());
-                } else if existing.authoritative && existing.dirty {
+                } else if existing.authoritative && existing.dirty && !reconciled_dirty_usage {
                     let mut preserved = existing.clone();
                     if preserved.pending_negative_delta > 0 && preserved.usage.size == 0 {
                         // Omitting the bucket from a complete snapshot confirms
                         // that its post-delete byte total is zero.
                         preserved.pending_negative_delta = 0;
+                        preserved.pending_negative_reconciliation = None;
                     }
                     preserved.stale_snapshot_pending = true;
+                    preserved.pending_scanner_position = preserved.pending_scanner_position.or(snapshot_position);
                     candidate.insert(preserved);
                 }
             }
@@ -5405,11 +5517,12 @@ mod tests {
             .expect("delete usage should remain cached")
             .usage_updated_at;
 
-        let stale = data_usage_info_for_test("bucket-a", 2, 768, mutation_update + Duration::from_nanos(1));
+        // Windows SystemTime has 100 ns resolution; retain strict ordering there.
+        let stale = data_usage_info_for_test("bucket-a", 2, 768, mutation_update + Duration::from_micros(1));
         replace_bucket_usage_memory_from_info(&stale).await;
         assert_eq!(get_bucket_usage_memory("bucket-a").await, Some(768));
 
-        let mut too_old_matching = data_usage_info_for_test("bucket-a", 1, 256, mutation_update - Duration::from_nanos(1));
+        let mut too_old_matching = data_usage_info_for_test("bucket-a", 1, 256, mutation_update - Duration::from_micros(1));
         too_old_matching
             .buckets_usage
             .get_mut("bucket-a")
@@ -5423,7 +5536,7 @@ mod tests {
             "a matching byte total from before the delete cannot release the hold"
         );
 
-        let mut reconciled = data_usage_info_for_test("bucket-a", 1, 256, mutation_update + Duration::from_nanos(2));
+        let mut reconciled = data_usage_info_for_test("bucket-a", 1, 256, mutation_update + Duration::from_micros(2));
         reconciled
             .buckets_usage
             .get_mut("bucket-a")
@@ -5759,6 +5872,333 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn complete_scanner_cycles_reconcile_expired_dirty_usage() {
+        for (remaining_objects, omit_bucket) in [(0, false), (1, false), (0, true)] {
+            clear_usage_memory_cache_for_test().await;
+            let bucket = "expired-dirty-usage";
+            let baseline = data_usage_info_for_test(bucket, 0, 0, SystemTime::now());
+            replace_bucket_usage_memory_from_info(&baseline).await;
+            record_bucket_object_write_memory(bucket, None, 42).await;
+            record_bucket_object_write_memory(bucket, None, 42).await;
+
+            // Lifecycle deletion bypasses request-layer memory accounting.
+            // Repeated complete scans must eventually replace that dirty overlay.
+            for cycle in 1..=3 {
+                let mut snapshot = data_usage_info_for_test(bucket, remaining_objects, remaining_objects * 42, SystemTime::now());
+                snapshot.scanner_epoch = Some(7);
+                snapshot.scanner_cycle = Some(cycle);
+                if omit_bucket {
+                    snapshot.buckets_usage.clear();
+                    snapshot.bucket_sizes.clear();
+                    snapshot.buckets_count = 0;
+                    snapshot.calculate_totals();
+                }
+                replace_bucket_usage_memory_from_info(&snapshot).await;
+                if cycle == 3 {
+                    apply_bucket_usage_memory_overlay_if_authoritative(&mut snapshot, true).await;
+                    assert_eq!(snapshot.objects_total_count, remaining_objects);
+                    assert_eq!(snapshot.objects_total_size, remaining_objects * 42);
+                    if omit_bucket {
+                        assert!(!snapshot.buckets_usage.contains_key(bucket));
+                    } else {
+                        assert_eq!(snapshot.buckets_usage[bucket].objects_count, remaining_objects);
+                        assert_eq!(snapshot.buckets_usage[bucket].size, remaining_objects * 42);
+                    }
+                }
+            }
+        }
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn expiry_receipts_exclude_refresh_until_committed_or_cancelled() {
+        clear_usage_memory_cache_for_test().await;
+        let bucket = "expiry-receipt";
+        replace_bucket_usage_memory_from_info(&data_usage_info_for_test(bucket, 2, 84, SystemTime::now())).await;
+        let first = begin_expiry_usage_accounting(bucket).await;
+        let cancelled = begin_expiry_usage_accounting(bucket).await;
+        let after_delete = data_usage_info_for_test(bucket, 1, 42, SystemTime::now());
+        replace_bucket_usage_memory_from_info(&after_delete).await;
+        assert_eq!(memory_cache().read().await[bucket].usage.size, 84);
+        first.commit(42).await;
+        assert_eq!(memory_cache().read().await[bucket].usage.size, 42);
+
+        let mut omitted = data_usage_info_for_test(bucket, 0, 0, SystemTime::now());
+        omitted.buckets_usage.clear();
+        omitted.bucket_sizes.clear();
+        omitted.buckets_count = 0;
+        omitted.calculate_totals();
+        replace_bucket_usage_memory_from_info(&omitted).await;
+        assert_eq!(memory_cache().read().await[bucket].usage.size, 42);
+        drop(cancelled);
+        assert_eq!(
+            memory_cache().read().await[bucket]
+                .in_flight_expirations
+                .load(Ordering::Acquire),
+            0
+        );
+
+        replace_bucket_usage_memory_from_info(&data_usage_info_for_test(bucket, 1, 42, SystemTime::now())).await;
+        assert_eq!(get_bucket_usage_memory(bucket).await, Some(42));
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn cancelled_expiry_receipt_reconciles_without_another_mutation() {
+        clear_usage_memory_cache_for_test().await;
+        let bucket = "expiry-cancelled-after-delete";
+        replace_bucket_usage_memory_from_info(&data_usage_info_for_test(bucket, 0, 0, SystemTime::now())).await;
+        record_bucket_object_write_memory(bucket, None, 42).await;
+        record_bucket_object_write_memory(bucket, None, 42).await;
+        let receipt = begin_expiry_usage_accounting(bucket).await;
+
+        // Storage has deleted one object, but the task has not committed its
+        // memory receipt. Even two complete scans must not replace its baseline.
+        let mut scanned = data_usage_info_for_test(bucket, 1, 42, SystemTime::now());
+        scanned.scanner_epoch = Some(7);
+        for cycle in [10, 11] {
+            scanned.scanner_cycle = Some(cycle);
+            replace_bucket_usage_memory_from_info(&scanned).await;
+            assert_eq!(get_bucket_usage_memory(bucket).await, Some(84));
+        }
+        drop(receipt);
+
+        for (cycle, expected) in [(12, 84), (13, 42)] {
+            scanned.scanner_cycle = Some(cycle);
+            replace_bucket_usage_memory_from_info(&scanned).await;
+            assert_eq!(get_bucket_usage_memory(bucket).await, Some(expected));
+            let mut displayed = scanned.clone();
+            apply_bucket_usage_memory_overlay(&mut displayed).await;
+            assert_eq!(displayed.buckets_usage[bucket].size, expected);
+            assert_eq!(displayed.buckets_usage[bucket].objects_count, expected / 42);
+        }
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn expiry_restart_reloads_durable_usage_without_persisting_the_overlay() {
+        use crate::bucket::metadata_sys::test_support::isolated_store_over_temp_disks;
+
+        let (_dirs, store) = isolated_store_over_temp_disks().await;
+        let bucket = "expiry-restart";
+        clear_usage_memory_cache_for_test().await;
+        let baseline = data_usage_info_for_test(bucket, 2, 84, SystemTime::now());
+        store_data_usage_in_backend(baseline.clone(), store.clone())
+            .await
+            .expect("persist the pre-expiration scanner snapshot");
+        replace_bucket_usage_memory_from_info(&baseline).await;
+        begin_expiry_usage_accounting(bucket).await.commit(42).await;
+        let mut displayed = baseline.clone();
+        apply_bucket_usage_memory_overlay(&mut displayed).await;
+        assert_eq!(displayed.buckets_usage[bucket].size, 42);
+        assert_eq!(get_bucket_usage_memory(bucket).await, Some(84));
+
+        // Recreate only process-local usage state, then use the real backend
+        // loader. This is a restart-state regression, not a process restart E2E.
+        clear_usage_memory_cache_for_test().await;
+        invalidate_data_usage_snapshot_cache().await;
+        let recovered = load_data_usage_from_backend_cached(store.clone())
+            .await
+            .expect("reload the durable pre-expiration snapshot");
+        assert_eq!(recovered.buckets_usage[bucket].size, 84);
+        replace_bucket_usage_memory_from_info(&recovered).await;
+        assert_eq!(get_bucket_usage_memory(bucket).await, Some(84));
+
+        let confirmed = data_usage_info_for_test(bucket, 1, 42, SystemTime::now());
+        store_data_usage_in_backend(confirmed, store.clone())
+            .await
+            .expect("persist scanner confirmation without another object mutation");
+        let recovered = load_data_usage_from_backend_cached(store.clone())
+            .await
+            .expect("reload the confirmed snapshot");
+        replace_bucket_usage_memory_from_info(&recovered).await;
+        assert_eq!(get_bucket_usage_memory(bucket).await, Some(42));
+
+        // Historical overlay drift must not contaminate the persisted scan.
+        record_bucket_object_write_memory(bucket, None, 42).await;
+        clear_usage_memory_cache_for_test().await;
+        invalidate_data_usage_snapshot_cache().await;
+        let mut recovered = load_data_usage_from_backend_cached(store)
+            .await
+            .expect("reload the confirmed snapshot after losing the stale overlay");
+        replace_bucket_usage_memory_from_info(&recovered).await;
+        apply_bucket_usage_memory_overlay(&mut recovered).await;
+        assert_eq!(recovered.buckets_usage[bucket].objects_count, 1);
+        assert_eq!(recovered.buckets_usage[bucket].size, 42);
+        assert_eq!(get_bucket_usage_memory(bucket).await, Some(42));
+        clear_usage_memory_cache_for_test().await;
+        invalidate_data_usage_snapshot_cache().await;
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn expiry_receipt_cannot_credit_a_replaced_bucket_cache() {
+        clear_usage_memory_cache_for_test().await;
+        let bucket = "expiry-recreated";
+        replace_bucket_usage_memory_from_info(&data_usage_info_for_test(bucket, 1, 42, SystemTime::now())).await;
+        let receipt = begin_expiry_usage_accounting(bucket).await;
+        clear_usage_memory_cache_for_test().await;
+        replace_bucket_usage_memory_from_info(&data_usage_info_for_test(bucket, 2, 84, SystemTime::now())).await;
+        receipt.commit(42).await;
+        assert_eq!(memory_cache().read().await[bucket].usage.objects_count, 2);
+        assert_eq!(get_bucket_usage_memory(bucket).await, Some(84));
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn quota_delete_tranches_advance_despite_newer_mutations() {
+        clear_usage_memory_cache_for_test().await;
+        let bucket = "expiry-busy-quota";
+        replace_bucket_usage_memory_from_info(&data_usage_info_for_test(bucket, 10, 420, SystemTime::now())).await;
+        record_bucket_object_delete_memory(bucket, 42, true).await;
+        let mut observation = data_usage_info_for_test(bucket, 9, 378, SystemTime::now());
+        observation.scanner_epoch = Some(7);
+        observation.scanner_cycle = Some(10);
+        record_bucket_object_write_memory(bucket, None, 42).await;
+        replace_bucket_usage_memory_from_info(&observation).await;
+        assert_eq!(get_bucket_usage_memory(bucket).await, Some(462));
+
+        for (epoch, cycle, partial, old_time) in [
+            (8, 10, false, false),
+            (6, 11, false, false),
+            (7, 11, true, false),
+            (7, 11, false, true),
+        ] {
+            let mut rejected = observation.clone();
+            rejected.scanner_epoch = Some(epoch);
+            rejected.scanner_cycle = Some(cycle);
+            rejected.usage_snapshot_complete = !partial;
+            rejected.usage_snapshot_partial = partial;
+            if old_time {
+                rejected.last_update = Some(SystemTime::UNIX_EPOCH);
+            }
+            replace_bucket_usage_memory_from_info(&rejected).await;
+            assert_eq!(memory_cache().read().await[bucket].pending_negative_delta, 42);
+        }
+
+        observation.last_update = Some(SystemTime::now());
+        observation.scanner_cycle = Some(11);
+        // A newer delete must stay charged, without delaying the older receipt.
+        record_bucket_object_delete_memory(bucket, 42, true).await;
+        record_bucket_object_write_memory(bucket, None, 42).await;
+        replace_bucket_usage_memory_from_info(&observation).await;
+        assert_eq!(memory_cache().read().await[bucket].pending_negative_delta, 42);
+        assert_eq!(get_bucket_usage_memory(bucket).await, Some(462));
+
+        for cycle in [12, 13] {
+            observation.last_update = Some(SystemTime::now());
+            observation.scanner_cycle = Some(cycle);
+            record_bucket_object_write_memory(bucket, None, 42).await;
+            replace_bucket_usage_memory_from_info(&observation).await;
+        }
+        assert_eq!(memory_cache().read().await[bucket].pending_negative_delta, 0);
+        assert_eq!(get_bucket_usage_memory(bucket).await, Some(504));
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn quota_delete_hold_stays_bounded_with_deletes_after_every_observation() {
+        clear_usage_memory_cache_for_test().await;
+        let bucket = "expiry-continuous-quota";
+        replace_bucket_usage_memory_from_info(&data_usage_info_for_test(bucket, 10, 420, SystemTime::now())).await;
+        record_bucket_object_delete_memory(bucket, 42, true).await;
+        for cycle in 1..=12 {
+            let mut observed = data_usage_info_for_test(bucket, 9, 378, SystemTime::now());
+            observed.scanner_epoch = Some(7);
+            observed.scanner_cycle = Some(cycle);
+            record_bucket_object_write_memory(bucket, None, 42).await;
+            record_bucket_object_delete_memory(bucket, 42, true).await;
+            replace_bucket_usage_memory_from_info(&observed).await;
+            let cache = memory_cache().read().await;
+            assert_eq!(cache[bucket].usage.size, 378);
+            assert!(
+                cache[bucket].pending_negative_delta <= 126,
+                "confirmed older deletes must leave the quota hold"
+            );
+        }
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn dirty_usage_reconciliation_requires_a_later_complete_observation() {
+        clear_usage_memory_cache_for_test().await;
+        let bucket = "dirty-observation-fence";
+        let baseline = data_usage_info_for_test(bucket, 0, 0, SystemTime::now());
+        replace_bucket_usage_memory_from_info(&baseline).await;
+        record_bucket_object_write_memory(bucket, None, 42).await;
+        let mutation_at = memory_cache().read().await[bucket].usage_updated_at;
+        let mut first = data_usage_info_for_test(bucket, 0, 0, mutation_at);
+        first.scanner_epoch = Some(7);
+        first.scanner_cycle = Some(10);
+        let mut partial = first.clone();
+        partial.usage_snapshot_partial = true;
+        partial.usage_snapshot_complete = false;
+        replace_bucket_usage_memory_from_info(&partial).await;
+        assert_eq!(memory_cache().read().await[bucket].pending_scanner_position, None);
+        replace_bucket_usage_memory_from_info(&first).await;
+
+        for (epoch, cycle, partial, update) in [
+            (Some(7), Some(10), false, mutation_at),
+            (Some(8), Some(10), false, mutation_at),
+            (Some(6), Some(11), false, mutation_at),
+            (Some(7), Some(9), false, mutation_at),
+            (Some(7), Some(11), true, mutation_at),
+            (None, Some(11), false, mutation_at),
+            (Some(7), None, false, mutation_at),
+            // Windows `SystemTime` stores 100 ns intervals, so a 1 ns offset is a no-op there.
+            (Some(7), Some(11), false, mutation_at - Duration::from_micros(1)),
+        ] {
+            let mut snapshot = first.clone();
+            snapshot.scanner_epoch = epoch;
+            snapshot.scanner_cycle = cycle;
+            snapshot.usage_snapshot_partial = partial;
+            snapshot.usage_snapshot_complete = !partial;
+            snapshot.last_update = Some(update);
+            replace_bucket_usage_memory_from_info(&snapshot).await;
+            apply_bucket_usage_memory_overlay_if_authoritative(&mut snapshot, true).await;
+            assert_eq!(snapshot.objects_total_count, 1, "epoch={epoch:?}, cycle={cycle:?}, partial={partial}");
+            assert_eq!(snapshot.objects_total_size, 42);
+        }
+
+        let mut successor = first;
+        successor.scanner_epoch = Some(8);
+        successor.scanner_cycle = Some(11);
+        replace_bucket_usage_memory_from_info(&successor).await;
+        apply_bucket_usage_memory_overlay_if_authoritative(&mut successor, true).await;
+        assert_eq!(successor.objects_total_count, 0);
+        assert_eq!(successor.objects_total_size, 0);
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn a_new_authoritative_mutation_restarts_reconciliation() {
+        clear_usage_memory_cache_for_test().await;
+        let bucket = "dirty-mutation-fence";
+        replace_bucket_usage_memory_from_info(&data_usage_info_for_test(bucket, 0, 0, SystemTime::now())).await;
+        record_bucket_object_write_memory(bucket, None, 42).await;
+        let mut snapshot = data_usage_info_for_test(bucket, 0, 0, SystemTime::now());
+        snapshot.scanner_epoch = Some(7);
+        snapshot.scanner_cycle = Some(10);
+        replace_bucket_usage_memory_from_info(&snapshot).await;
+
+        record_bucket_object_write_memory(bucket, None, 42).await;
+        snapshot.last_update = Some(SystemTime::now());
+        snapshot.scanner_cycle = Some(11);
+        replace_bucket_usage_memory_from_info(&snapshot).await;
+        let mut response = snapshot.clone();
+        apply_bucket_usage_memory_overlay_if_authoritative(&mut response, true).await;
+        assert_eq!(response.objects_total_count, 2);
+        assert_eq!(response.objects_total_size, 84);
+
+        snapshot.scanner_cycle = Some(12);
+        replace_bucket_usage_memory_from_info(&snapshot).await;
+        apply_bucket_usage_memory_overlay_if_authoritative(&mut snapshot, true).await;
+        assert_eq!(snapshot.objects_total_count, 0);
+        assert_eq!(snapshot.objects_total_size, 0);
+    }
+
     #[tokio::test]
     #[serial]
     async fn authoritative_refresh_can_reduce_memory_usage() {

crates/ecstore/src/ecstore_validation_blackbox.rs

@@ -408,6 +408,7 @@ async fn blackbox_heal_requests_preserve_repair_scope() {
                         &ObjectOptions {
                             no_lock: true,
                             versioned: true,
+                            expected_bucket_incarnation_id: Some(put_incarnation),
                             ..Default::default()
                         },
                     )

crates/scanner/src/lib.rs

@@ -87,13 +87,13 @@ pub use runtime_config::{apply_scanner_runtime_config, scanner_runtime_config_st
 pub use rustfs_scanner_metrics::last_minute;
 pub use scanner::{
     SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD, ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus,
-    ScannerCycleScheduleStatus, ScannerPauseBacklogAlertReason, ScannerPauseBacklogPhase, ScannerPauseBacklogStatus,
-    ScannerPauseBacklogThresholds, ScannerRecoveryIntentAcceptResult, ScannerRecoveryIntentConflict, ScannerRecoveryIntentRecord,
-    ScannerRecoveryIntentRequest, ScannerUsageStateResetResult, accept_scanner_usage_recovery_intent,
-    get_scanner_usage_recovery_intent, init_data_scanner, init_scanner_with_recovery, register_scanner_pause_backlog_retirement,
-    reset_scanner_cycle_recovery, reset_scanner_usage_state_for_full_rebuild, run_scanner_usage_recovery_intent,
-    scanner_cycle_recovery_status, scanner_cycle_schedule_status, scanner_pause_backlog_status, scanner_recovery_actor_sha256,
-    scanner_topology_digest,
+    ScannerCycleScheduleStatus, ScannerPauseBacklogAlertReason, ScannerPauseBacklogPhase, ScannerPauseBacklogReplicaStatus,
+    ScannerPauseBacklogReplicaStatusState, ScannerPauseBacklogStatus, ScannerPauseBacklogThresholds,
+    ScannerRecoveryIntentAcceptResult, ScannerRecoveryIntentConflict, ScannerRecoveryIntentRecord, ScannerRecoveryIntentRequest,
+    ScannerUsageStateResetResult, accept_scanner_usage_recovery_intent, get_scanner_usage_recovery_intent, init_data_scanner,
+    init_scanner_with_recovery, register_scanner_pause_backlog_retirement, reset_scanner_cycle_recovery,
+    reset_scanner_usage_state_for_full_rebuild, run_scanner_usage_recovery_intent, scanner_cycle_recovery_status,
+    scanner_cycle_schedule_status, scanner_pause_backlog_status, scanner_recovery_actor_sha256, scanner_topology_digest,
 };
 pub use scanner_io::{
     ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageClearObserver, ScannerDirtyUsageMutationObserver,

crates/scanner/src/scanner.rs

@@ -3800,6 +3800,7 @@ mod usage_store;
 
 use activity::*;
 use backlog::*;
+pub use backlog::{ScannerPauseBacklogReplicaStatus, ScannerPauseBacklogReplicaStatusState};
 use cycle_state::*;
 use leadership::*;
 pub(crate) use usage_store::RootPublicationProof;

crates/scanner/src/scanner/backlog.rs

@@ -144,12 +144,38 @@ pub struct ScannerPauseBacklogStatus {
     pub replica_count: usize,
     pub healthy_replicas: usize,
     pub stale_or_unavailable_replicas: usize,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub replica_diagnostics: Option<Vec<ScannerPauseBacklogReplicaStatus>>,
     pub alerting: bool,
     pub alert_reasons: Vec<ScannerPauseBacklogAlertReason>,
     pub thresholds: ScannerPauseBacklogThresholds,
     pub error: Option<String>,
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ScannerPauseBacklogReplicaStatusState {
+    Missing,
+    Valid,
+    Invalid,
+    FutureSchema,
+    Unavailable,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+pub struct ScannerPauseBacklogReplicaStatus {
+    pub pool_index: usize,
+    pub set_index: usize,
+    pub state: ScannerPauseBacklogReplicaStatusState,
+    pub revision: Option<String>,
+    pub stable_generation: Option<u64>,
+    pub stable_writer_epoch: Option<u64>,
+    pub committed_generation: Option<u64>,
+    pub committed_writer_epoch: Option<u64>,
+    pub committed_replica_count: Option<usize>,
+    pub error: Option<String>,
+}
+
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
 #[serde(deny_unknown_fields)]
 struct ScannerPauseBacklogLedger {
@@ -769,6 +795,9 @@ impl LoadedScannerPauseBacklog {
     fn status(&self, now: u64, error: Option<String>) -> ScannerPauseBacklogStatus {
         let persistence_unavailable = error.is_some();
         let replica_degraded = self.durable && self.stale_or_unavailable_replicas > 0;
+        let diagnostics_needed =
+            persistence_unavailable || !self.durable || replica_degraded || self.persistence_state != "healthy";
+        let replica_diagnostics = diagnostics_needed.then(|| scanner_pause_backlog_replica_diagnostics(&self.replicas));
         status_from_ledger(
             &self.ledger,
             now,
@@ -780,6 +809,7 @@ impl LoadedScannerPauseBacklog {
             replica_degraded,
             persistence_unavailable,
             error,
+            replica_diagnostics,
         )
     }
 }
@@ -796,6 +826,7 @@ fn status_from_ledger(
     replica_degraded: bool,
     persistence_unavailable: bool,
     error: Option<String>,
+    replica_diagnostics: Option<Vec<ScannerPauseBacklogReplicaStatus>>,
 ) -> ScannerPauseBacklogStatus {
     let alert_reasons = ledger.alert_reasons(now, replica_degraded, persistence_unavailable);
     let status = ScannerPauseBacklogStatus {
@@ -833,6 +864,7 @@ fn status_from_ledger(
         replica_count,
         healthy_replicas,
         stale_or_unavailable_replicas,
+        replica_diagnostics,
         alerting: !alert_reasons.is_empty(),
         alert_reasons,
         thresholds: ScannerPauseBacklogThresholds::default(),
@@ -947,6 +979,83 @@ async fn read_scanner_pause_backlog_replica(store: Arc<SetDisks>) -> ScannerPaus
     ScannerPauseBacklogReplica { id, revision, state }
 }
 
+fn scanner_pause_backlog_replica_diagnostics(replicas: &[ScannerPauseBacklogReplica]) -> Vec<ScannerPauseBacklogReplicaStatus> {
+    replicas
+        .iter()
+        .map(|replica| {
+            let revision = match &replica.revision {
+                Some(DataUsageCacheRevision::Etag(etag)) => Some(etag.clone()),
+                Some(DataUsageCacheRevision::Missing) | None => None,
+            };
+            let (
+                state,
+                stable_generation,
+                stable_writer_epoch,
+                committed_generation,
+                committed_writer_epoch,
+                committed_replica_count,
+                error,
+            ) = match &replica.state {
+                ScannerPauseBacklogReplicaState::Missing => {
+                    (ScannerPauseBacklogReplicaStatusState::Missing, None, None, None, None, None, None)
+                }
+                ScannerPauseBacklogReplicaState::Valid(record) => {
+                    let stable = record.stable.as_ref();
+                    let committed = record.committed.as_ref();
+                    (
+                        ScannerPauseBacklogReplicaStatusState::Valid,
+                        stable.map(|ledger| ledger.generation),
+                        stable.map(|ledger| ledger.writer_epoch),
+                        committed.map(|commit| commit.ledger.generation),
+                        committed.map(|commit| commit.ledger.writer_epoch),
+                        committed.map(|commit| commit.replicas.len()),
+                        None,
+                    )
+                }
+                ScannerPauseBacklogReplicaState::Invalid(reason) => (
+                    ScannerPauseBacklogReplicaStatusState::Invalid,
+                    None,
+                    None,
+                    None,
+                    None,
+                    None,
+                    Some(reason.clone()),
+                ),
+                ScannerPauseBacklogReplicaState::FutureSchema(version) => (
+                    ScannerPauseBacklogReplicaStatusState::FutureSchema,
+                    None,
+                    None,
+                    None,
+                    None,
+                    None,
+                    Some(format!("unsupported future schema {version}")),
+                ),
+                ScannerPauseBacklogReplicaState::Unavailable(reason) => (
+                    ScannerPauseBacklogReplicaStatusState::Unavailable,
+                    None,
+                    None,
+                    None,
+                    None,
+                    None,
+                    Some(reason.clone()),
+                ),
+            };
+            ScannerPauseBacklogReplicaStatus {
+                pool_index: replica.id.pool_index,
+                set_index: replica.id.set_index,
+                state,
+                revision,
+                stable_generation,
+                stable_writer_epoch,
+                committed_generation,
+                committed_writer_epoch,
+                committed_replica_count,
+                error,
+            }
+        })
+        .collect()
+}
+
 fn decode_scanner_pause_backlog_ledger(data: &[u8]) -> ScannerPauseBacklogReplicaState {
     let value = match serde_json::from_slice::<serde_json::Value>(data) {
         Ok(value) => value,
@@ -1016,6 +1125,33 @@ enum ScannerPauseBacklogSelectionError {
     InvalidRecord(String),
 }
 
+#[derive(Clone)]
+struct ScannerPauseBacklogSelectionFailure {
+    error: ScannerPauseBacklogSelectionError,
+    replicas: Vec<ScannerPauseBacklogReplica>,
+}
+
+impl fmt::Debug for ScannerPauseBacklogSelectionFailure {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ScannerPauseBacklogSelectionFailure")
+            .field("error", &self.error)
+            .finish()
+    }
+}
+
+impl fmt::Display for ScannerPauseBacklogSelectionFailure {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt::Display::fmt(&self.error, f)
+    }
+}
+
+fn scanner_pause_backlog_selection_failure(
+    replicas: Vec<ScannerPauseBacklogReplica>,
+    error: ScannerPauseBacklogSelectionError,
+) -> ScannerPauseBacklogSelectionFailure {
+    ScannerPauseBacklogSelectionFailure { error, replicas }
+}
+
 impl fmt::Display for ScannerPauseBacklogSelectionError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
@@ -1034,11 +1170,12 @@ impl From<ScannerPauseBacklogCommitSelectionError> for ScannerPauseBacklogSelect
 }
 
 fn scanner_pause_backlog_retirement_selection_error(
-    err: ScannerPauseBacklogSelectionError,
+    error: ScannerPauseBacklogSelectionError,
 ) -> ScannerPauseBacklogRetirementError {
-    match err {
+    let reason = error.to_string();
+    match error {
         ScannerPauseBacklogSelectionError::RetryableHandoffConflict => {
-            ScannerPauseBacklogRetirementError::HandoffConflict { reason: err.to_string() }
+            ScannerPauseBacklogRetirementError::HandoffConflict { reason }
         }
         ScannerPauseBacklogSelectionError::AuthorityConflict(reason) => {
             ScannerPauseBacklogRetirementError::AuthorityConflict { reason }
@@ -1123,24 +1260,33 @@ fn select_scanner_pause_backlog_commit(
 
 fn select_scanner_pause_backlog_replicas(
     replicas: Vec<ScannerPauseBacklogReplica>,
-) -> Result<LoadedScannerPauseBacklog, ScannerPauseBacklogSelectionError> {
+) -> Result<LoadedScannerPauseBacklog, ScannerPauseBacklogSelectionFailure> {
     if replicas.is_empty() {
-        return Err(ScannerPauseBacklogSelectionError::AuthorityConflict(
-            "scanner pause backlog has no storage replicas".to_string(),
+        return Err(scanner_pause_backlog_selection_failure(
+            replicas,
+            ScannerPauseBacklogSelectionError::AuthorityConflict("scanner pause backlog has no storage replicas".to_string()),
         ));
     }
-    for replica in &replicas {
-        if let ScannerPauseBacklogReplicaState::FutureSchema(version) = &replica.state {
-            return Err(ScannerPauseBacklogSelectionError::InvalidRecord(format!(
+    if let Some((replica_id, version)) = replicas.iter().find_map(|replica| match &replica.state {
+        ScannerPauseBacklogReplicaState::FutureSchema(version) => Some((replica.id, *version)),
+        _ => None,
+    }) {
+        return Err(scanner_pause_backlog_selection_failure(
+            replicas,
+            ScannerPauseBacklogSelectionError::InvalidRecord(format!(
                 "scanner pause backlog pool {} set {} uses future schema {version}",
-                replica.id.pool_index, replica.id.set_index
-            )));
-        }
+                replica_id.pool_index, replica_id.set_index
+            )),
+        ));
     }
 
     let replica_ids = scanner_pause_backlog_replica_ids(&replicas);
-    let authoritative_commit =
-        select_scanner_pause_backlog_commit(&replicas, &replica_ids).map_err(ScannerPauseBacklogSelectionError::from)?;
+    let authoritative_commit = match select_scanner_pause_backlog_commit(&replicas, &replica_ids) {
+        Ok(commit) => commit,
+        Err(error) => {
+            return Err(scanner_pause_backlog_selection_failure(replicas, error.into()));
+        }
+    };
     let stable_consensus = if replicas.iter().all(|replica| {
         matches!(
             &replica.state,
@@ -1175,16 +1321,21 @@ fn select_scanner_pause_backlog_replicas(
                     "scanner pause backlog pool {} set {} is unavailable: {reason}",
                     replica.id.pool_index, replica.id.set_index
                 );
-                return Err(match &replica.state {
+                let error = match &replica.state {
                     ScannerPauseBacklogReplicaState::Invalid(_) => ScannerPauseBacklogSelectionError::InvalidRecord(reason),
                     _ => ScannerPauseBacklogSelectionError::AuthorityConflict(reason),
-                });
+                };
+                return Err(scanner_pause_backlog_selection_failure(replicas, error));
             }
             match &stable_consensus {
                 Ok(stable) => stable.clone(),
                 Err(()) => {
-                    return Err(ScannerPauseBacklogSelectionError::AuthorityConflict(
-                        "scanner pause backlog has neither a surviving membership commit nor a stable rollback point".to_string(),
+                    return Err(scanner_pause_backlog_selection_failure(
+                        replicas,
+                        ScannerPauseBacklogSelectionError::AuthorityConflict(
+                            "scanner pause backlog has neither a surviving membership commit nor a stable rollback point"
+                                .to_string(),
+                        ),
                     ));
                 }
             }
@@ -1340,13 +1491,15 @@ fn verify_scanner_pause_backlog_retirement(
         .filter(|replica| replica.id.pool_index != source_pool_index)
         .cloned()
         .collect::<Vec<_>>();
-    let selected = select_scanner_pause_backlog_replicas(surviving).map_err(scanner_pause_backlog_retirement_selection_error)?;
+    let selected = select_scanner_pause_backlog_replicas(surviving)
+        .map_err(|failure| scanner_pause_backlog_retirement_selection_error(failure.error))?;
     if !selected.durable || !selected.stable_matches_ledger {
         return Err(scanner_pause_backlog_retirement_authority_conflict(
             "scanner pause backlog retirement requires stable authority on the complete surviving membership",
         ));
     }
-    let before = select_scanner_pause_backlog_replicas(decoded).map_err(scanner_pause_backlog_retirement_selection_error)?;
+    let before = select_scanner_pause_backlog_replicas(decoded)
+        .map_err(|failure| scanner_pause_backlog_retirement_selection_error(failure.error))?;
     if !before.durable || before.ledger != selected.ledger {
         return Err(scanner_pause_backlog_retirement_authority_conflict(
             "scanner pause backlog retirement would change the native ledger authority",
@@ -1439,7 +1592,7 @@ fn plan_scanner_pause_backlog_retirement(
         // Use the same native selector that validates the full commit, rather
         // than inferring authority from source epoch or physical object time.
         full_selection
-            .map_err(scanner_pause_backlog_retirement_selection_error)?
+            .map_err(|failure| scanner_pause_backlog_retirement_selection_error(failure.error))?
             .ledger
     } else {
         // An interrupted cohort switch can invalidate the old full commit
@@ -1448,7 +1601,7 @@ fn plan_scanner_pause_backlog_retirement(
         // must have been rejected before this recovery path.
         let current = selected
             .as_ref()
-            .map_err(|err| scanner_pause_backlog_retirement_selection_error(err.clone()))?;
+            .map_err(|failure| scanner_pause_backlog_retirement_selection_error(failure.error.clone()))?;
         if !current.durable {
             return Err(scanner_pause_backlog_retirement_authority_conflict(
                 "scanner pause backlog retirement has no proven native authority",
@@ -1548,12 +1701,19 @@ async fn load_scanner_pause_backlog<S>(storeapi: Arc<S>) -> Result<LoadedScanner
 where
     S: ScannerStorage,
 {
-    let writable = storeapi.scanner_pause_backlog_writable_set_disks().await;
-    if writable.is_empty() {
+    let replicas = read_scanner_pause_backlog_replicas(storeapi).await;
+    if replicas.is_empty() {
         return Err("scanner pause backlog has no surviving storage replicas".to_string());
     }
-    let replicas = join_all(writable.into_iter().map(read_scanner_pause_backlog_replica)).await;
-    select_scanner_pause_backlog_replicas(replicas).map_err(|err| err.to_string())
+    select_scanner_pause_backlog_replicas(replicas).map_err(|failure| failure.to_string())
+}
+
+async fn read_scanner_pause_backlog_replicas<S>(storeapi: Arc<S>) -> Vec<ScannerPauseBacklogReplica>
+where
+    S: ScannerStorage,
+{
+    let writable = storeapi.scanner_pause_backlog_writable_set_disks().await;
+    join_all(writable.into_iter().map(read_scanner_pause_backlog_replica)).await
 }
 
 async fn write_scanner_pause_backlog_record<S>(
@@ -1929,21 +2089,25 @@ fn runtime_error() -> Option<String> {
 
 pub async fn scanner_pause_backlog_status(storeapi: Arc<ECStore>) -> ScannerPauseBacklogStatus {
     let now = unix_now();
-    match load_scanner_pause_backlog(storeapi).await {
+    let replicas = read_scanner_pause_backlog_replicas(storeapi).await;
+    let replica_count = replicas.len();
+    match select_scanner_pause_backlog_replicas(replicas) {
         Ok(loaded) => loaded.status(now, runtime_error()),
-        Err(error) => {
-            let error = runtime_error().map_or(error.clone(), |runtime| format!("{error}; {runtime}"));
+        Err(failure) => {
+            let replica_diagnostics = scanner_pause_backlog_replica_diagnostics(&failure.replicas);
+            let error = runtime_error().map_or(failure.to_string(), |runtime| format!("{failure}; {runtime}"));
             status_from_ledger(
                 &ScannerPauseBacklogLedger::default(),
                 now,
                 "unavailable".to_string(),
                 false,
+                replica_count,
                 0,
-                0,
-                0,
+                replica_count,
                 false,
                 true,
                 Some(error),
+                Some(replica_diagnostics),
             )
         }
     }
@@ -1998,6 +2162,221 @@ mod tests {
         (root, store)
     }
 
+    #[test]
+    #[serial_test::serial]
+    fn native_status_reports_missing_and_conflicting_pause_backlog_replicas() {
+        run_native_retirement_test(async || {
+            set_runtime_error(None);
+            let (_root, store) = native_retirement_store().await;
+
+            let empty = scanner_pause_backlog_status(Arc::clone(&store)).await;
+            assert_eq!(empty.persistence_state, "missing");
+            assert_eq!(empty.replica_count, 6);
+            assert_eq!(
+                empty.replica_diagnostics.as_ref().map(Vec::len),
+                Some(6),
+                "initial all-missing membership should be visible without being treated as an error"
+            );
+            assert!(
+                empty
+                    .replica_diagnostics
+                    .as_ref()
+                    .expect("missing replicas should be diagnosed")
+                    .iter()
+                    .all(|replica| replica.state == ScannerPauseBacklogReplicaStatusState::Missing)
+            );
+
+            let stable = durable_ledger(unix_now());
+            let members = (0..3)
+                .flat_map(|pool_index| (0..2).map(move |set_index| replica_id(pool_index, set_index)))
+                .collect::<Vec<_>>();
+            let committed = ScannerPauseBacklogCommitRecord::new(stable.clone(), members);
+            let record = ScannerPauseBacklogReplicaRecord::new(Some(stable), Some(committed));
+            let bytes = encode_scanner_pause_backlog_record(&record).expect("stable replica should encode");
+            Arc::clone(&store)
+                .save_scanner_pause_backlog_replica(0, 0, bytes, DataUsageCacheRevision::Missing.preconditions())
+                .await
+                .expect("native fixture should publish its one stable replica");
+
+            let conflicted = scanner_pause_backlog_status(Arc::clone(&store)).await;
+            assert_eq!(conflicted.persistence_state, "unavailable");
+            assert_eq!(conflicted.replica_count, 6, "failed selection must preserve the actual replica count");
+            assert!(
+                conflicted
+                    .error
+                    .as_deref()
+                    .is_some_and(|error| error.contains("neither a surviving membership commit"))
+            );
+            let diagnostics = conflicted
+                .replica_diagnostics
+                .as_ref()
+                .expect("authority conflict should expose per-replica diagnostics");
+            assert_eq!(diagnostics.len(), 6);
+            assert!(diagnostics.iter().any(|replica| {
+                replica.pool_index == 0
+                    && replica.set_index == 0
+                    && replica.state == ScannerPauseBacklogReplicaStatusState::Valid
+                    && replica.stable_generation == Some(1)
+                    && replica.committed_generation == Some(1)
+                    && replica.committed_replica_count == Some(6)
+                    && replica.revision.is_some()
+            }));
+            assert_eq!(
+                diagnostics
+                    .iter()
+                    .filter(|replica| replica.state == ScannerPauseBacklogReplicaStatusState::Missing)
+                    .count(),
+                5
+            );
+            let conflicted_json = serde_json::to_value(&conflicted).expect("status diagnostics should encode as JSON");
+            let conflicted_diagnostics = conflicted_json["replica_diagnostics"]
+                .as_array()
+                .expect("replica diagnostics should serialize as an array");
+            assert!(
+                conflicted_diagnostics
+                    .iter()
+                    .any(|replica| { replica["pool_index"] == 0 && replica["set_index"] == 0 && replica["state"] == "valid" })
+            );
+
+            for (pool_index, set_index, bytes) in [
+                (0, 1, b"invalid backlog JSON".to_vec()),
+                (1, 0, br#"{"replica_schema_version":999}"#.to_vec()),
+            ] {
+                Arc::clone(&store)
+                    .save_scanner_pause_backlog_replica(
+                        pool_index,
+                        set_index,
+                        bytes,
+                        DataUsageCacheRevision::Missing.preconditions(),
+                    )
+                    .await
+                    .expect("native fixture should publish malformed and future-schema replicas");
+            }
+
+            let classified = scanner_pause_backlog_status(Arc::clone(&store)).await;
+            assert!(
+                classified
+                    .error
+                    .as_deref()
+                    .is_some_and(|error| error.contains("future schema 999"))
+            );
+            let classified_json = serde_json::to_value(&classified).expect("classified status should encode as JSON");
+            let classified_diagnostics = classified_json["replica_diagnostics"]
+                .as_array()
+                .expect("classified diagnostics should serialize as an array");
+            assert!(
+                classified_diagnostics
+                    .iter()
+                    .any(|replica| { replica["pool_index"] == 0 && replica["set_index"] == 1 && replica["state"] == "invalid" })
+            );
+            assert!(classified_diagnostics.iter().any(|replica| {
+                replica["pool_index"] == 1 && replica["set_index"] == 0 && replica["state"] == "future_schema"
+            }));
+            assert!(classified_diagnostics.iter().any(|replica| {
+                replica["pool_index"] == 0
+                    && replica["set_index"] == 1
+                    && replica["error"].as_str().is_some_and(|reason| !reason.is_empty())
+            }));
+
+            let unavailable = scanner_pause_backlog_replica_diagnostics(&[ScannerPauseBacklogReplica {
+                id: replica_id(9, 9),
+                revision: None,
+                state: ScannerPauseBacklogReplicaState::Unavailable("read failed".to_string()),
+            }]);
+            assert_eq!(unavailable[0].state, ScannerPauseBacklogReplicaStatusState::Unavailable);
+            assert_eq!(unavailable[0].error.as_deref(), Some("read failed"));
+
+            shutdown_native_retirement_store(store).await;
+            set_runtime_error(None);
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn native_status_omits_replica_diagnostics_for_healthy_backlog() {
+        run_native_retirement_test(async || {
+            set_runtime_error(None);
+            let (_root, store) = native_retirement_store().await;
+            ScannerPauseBacklogController::claim(Arc::clone(&store), unix_now())
+                .await
+                .expect("all-missing native backlog should claim and stabilize");
+
+            let status = scanner_pause_backlog_status(Arc::clone(&store)).await;
+            assert_eq!(status.persistence_state, "healthy");
+            assert!(status.durable);
+            assert_eq!(status.replica_count, 6);
+            assert!(status.replica_diagnostics.is_none());
+            let status_json = serde_json::to_value(status).expect("healthy status should encode");
+            assert!(status_json.get("replica_diagnostics").is_none());
+
+            shutdown_native_retirement_store(store).await;
+            set_runtime_error(None);
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn native_status_reports_competing_maximum_membership_commits() {
+        run_native_retirement_test(async || {
+            set_runtime_error(None);
+            let (_root, store) = native_retirement_store().await;
+            let controller = ScannerPauseBacklogController::claim(Arc::clone(&store), unix_now())
+                .await
+                .expect("all-missing native backlog should claim and stabilize");
+            let stable = controller.loaded.ledger.clone();
+            let replica_ids = scanner_pause_backlog_replica_ids(&controller.loaded.replicas);
+            drop(controller);
+
+            let split = replica_ids.len() / 2;
+            let first_membership = replica_ids[..split].to_vec();
+            let second_membership = replica_ids[split..].to_vec();
+            let now = stable.last_updated_at_unix_secs.saturating_add(10);
+            let first_ledger = claim_scanner_pause_backlog_writer(&stable, now).expect("first proof should advance");
+            let second_ledger = claim_scanner_pause_backlog_writer(&stable, now + 1).expect("second proof should advance");
+            assert_ne!(first_ledger, second_ledger);
+            let first_commit = ScannerPauseBacklogCommitRecord::new(first_ledger, first_membership);
+            let second_commit = ScannerPauseBacklogCommitRecord::new(second_ledger, second_membership);
+
+            for (index, id) in replica_ids.iter().enumerate() {
+                let committed = if index < split { &first_commit } else { &second_commit };
+                replace_native_pause_backlog_replica(
+                    &store,
+                    *id,
+                    ScannerPauseBacklogReplicaRecord::new(Some(stable.clone()), Some(committed.clone())),
+                )
+                .await;
+            }
+
+            let status = scanner_pause_backlog_status(Arc::clone(&store)).await;
+            assert_eq!(status.persistence_state, "unavailable");
+            assert_eq!(status.replica_count, replica_ids.len());
+            assert!(
+                status
+                    .error
+                    .as_deref()
+                    .is_some_and(|error| error.contains("conflicting maximum-membership commit proofs"))
+            );
+            let diagnostics = status
+                .replica_diagnostics
+                .as_ref()
+                .expect("competing proofs should expose replica diagnostics");
+            assert_eq!(diagnostics.len(), replica_ids.len());
+            assert!(
+                diagnostics
+                    .iter()
+                    .all(|replica| replica.committed_replica_count == Some(split) && replica.revision.is_some())
+            );
+            let revisions = diagnostics
+                .iter()
+                .filter_map(|replica| replica.revision.as_ref())
+                .collect::<std::collections::BTreeSet<_>>();
+            assert_eq!(revisions.len(), 2, "each competing proof cohort should have its own record revision");
+
+            shutdown_native_retirement_store(store).await;
+            set_runtime_error(None);
+        });
+    }
+
     async fn shutdown_native_retirement_store(store: Arc<ECStore>) {
         if let Some(token) = store.background_cancel_token() {
             token.cancel();
@@ -2027,6 +2406,21 @@ mod tests {
         (bytes, revision)
     }
 
+    async fn replace_native_pause_backlog_replica(
+        store: &Arc<ECStore>,
+        id: ScannerPauseBacklogReplicaId,
+        record: ScannerPauseBacklogReplicaRecord,
+    ) {
+        let set = Arc::clone(&store.pools[id.pool_index].disk_set[id.set_index]);
+        let current = read_scanner_pause_backlog_replica(set).await;
+        let revision = current.revision.expect("native backlog record should have a CAS revision");
+        let bytes = encode_scanner_pause_backlog_record(&record).expect("native backlog record should encode");
+        Arc::clone(store)
+            .save_scanner_pause_backlog_replica(id.pool_index, id.set_index, bytes, revision.preconditions())
+            .await
+            .expect("native backlog replica should accept its current CAS revision");
+    }
+
     async fn assert_native_source_missing(store: &ECStore, set_index: usize) {
         let source = store.pools[0].disk_set[set_index]
             .get_object_reader(
@@ -3969,11 +4363,13 @@ mod tests {
                 assert_eq!(ledger.phase, ScannerPauseBacklogPhase::CatchingUp);
                 assert!(ledger.pending_full_scan);
                 assert_eq!(ledger.next_attempt_at_unix_secs, 721);
-                let status = status_from_ledger(&ledger, 422, "healthy".to_string(), true, 1, 1, 0, false, false, None);
+                let status = status_from_ledger(&ledger, 422, "healthy".to_string(), true, 1, 1, 0, false, false, None, None);
                 assert_eq!(status.phase, ScannerPauseBacklogPhase::CatchingUp);
                 assert!(status.pending_full_scan);
                 assert_eq!(status.pending_work_items, 1);
                 assert!(status.rate_limited);
+                let status_json = serde_json::to_value(status).expect("healthy status should encode");
+                assert!(status_json.get("replica_diagnostics").is_none());
             }
         }
 

crates/scanner/src/scanner_io/tests.rs

@@ -45,6 +45,7 @@ use uuid::Uuid;
 
 mod scoped_entry_fallback;
 mod service_cohort;
+mod usage_expiration;
 
 #[derive(Clone)]
 struct FixedWorkloadProvider {

crates/scanner/src/scanner_io/tests/usage_expiration.rs

@@ -0,0 +1,238 @@
+// Copyright 2026 RustFS Team
+// SPDX-License-Identifier: Apache-2.0
+
+use super::*;
+use crate::storage_api::owner::ObjectOperations as _;
+use crate::storage_api::scanner_io::{
+    BUCKET_LIFECYCLE_CONFIG, apply_bucket_usage_memory_overlay, init_background_expiry,
+    load_admin_data_usage_from_backend_cached, load_data_usage_from_backend, record_bucket_object_delete_memory,
+    record_bucket_object_write_memory, replace_bucket_usage_memory_from_info, seed_persisted_usage, update_bucket_metadata,
+};
+
+// Lifecycle expiry goes through the process-global store, which the first
+// store created in the process claims. Nextest runs this test in its own
+// process; sharing one store also lets all scenarios run in one filtered
+// `cargo test` process.
+#[tokio::test]
+#[serial]
+async fn lifecycle_expiration_usage_converges_across_complete_scans() {
+    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
+    let next_cycle = check_persisted_stale_usage_recovery(&store, 1).await;
+    let next_cycle = check_usage_after_lifecycle_expiration(&store, "idle", false, next_cycle).await;
+    check_usage_after_lifecycle_expiration(&store, "busy", true, next_cycle).await;
+}
+
+async fn check_usage_after_lifecycle_expiration(
+    store: &Arc<ECStore>,
+    scenario: &str,
+    write_between_scans: bool,
+    first_cycle: u64,
+) -> u64 {
+    clear_dirty_usage_buckets_for_tests();
+    let cases = [
+        ("ilm-empty", true, false),
+        ("ilm-partial", true, true),
+        ("delete-empty", false, false),
+        ("delete-partial", false, true),
+    ]
+    .map(|(name, lifecycle, keep_one)| (format!("{scenario}-{name}"), lifecycle, keep_one));
+    let usage_matches = |snapshot: &DataUsageInfo, retained_writes: u64| {
+        cases.iter().all(|(bucket, _, keep_one)| {
+            snapshot.buckets_usage.get(bucket).is_some_and(|usage| {
+                usage.objects_count == u64::from(*keep_one) + retained_writes
+                    && usage.size == (u64::from(*keep_one) + retained_writes) * 42
+            })
+        })
+    };
+    let mut baseline = DataUsageInfo {
+        last_update: Some(SystemTime::now()),
+        usage_snapshot_complete: true,
+        ..Default::default()
+    };
+    for (bucket, _, _) in &cases {
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create usage bucket");
+        baseline.buckets_usage.insert(bucket.clone(), BucketUsageInfo::default());
+        baseline.bucket_sizes.insert(bucket.clone(), 0);
+    }
+    baseline.buckets_count = 4;
+    replace_bucket_usage_memory_from_info(&baseline).await;
+
+    for (bucket, lifecycle, keep_one) in &cases {
+        for object in ["expire/first", if *keep_one { "keep/second" } else { "expire/second" }] {
+            let mut reader = ScannerPutObjReader::from_vec(vec![0; 42]);
+            store
+                .put_object(bucket, object, &mut reader, &ScannerObjectOptions::default())
+                .await
+                .expect("write object before expiration");
+            record_bucket_object_write_memory(bucket, None, 42).await;
+        }
+        if *lifecycle {
+            let config = br#"<LifecycleConfiguration><Rule><ID>expire</ID><Status>Enabled</Status><Filter><Prefix>expire/</Prefix></Filter><Expiration><Days>0</Days></Expiration></Rule></LifecycleConfiguration>"#;
+            update_bucket_metadata(bucket, BUCKET_LIFECYCLE_CONFIG, config.to_vec())
+                .await
+                .expect("configure immediate expiration after writes");
+        } else {
+            for object in ["expire/first", "expire/second"]
+                .into_iter()
+                .take(if *keep_one { 1 } else { 2 })
+            {
+                store
+                    .delete_object(bucket, object, ScannerObjectOptions::default())
+                    .await
+                    .expect("ordinary deletion should succeed");
+                record_bucket_object_delete_memory(bucket, 42, true).await;
+            }
+        }
+    }
+    init_background_expiry(store.clone()).await;
+    wait_for_namespace_commit_tails(store.as_ref()).await;
+
+    let mut reconciled = false;
+    let mut observed_expiration = false;
+    let mut retained_writes = 0;
+    let mut reconciled_cycles = 0;
+    let mut cycle = first_cycle;
+    while cycle < first_cycle + 8 {
+        let ctx = CancellationToken::new();
+        let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
+        let (updates, mut receiver) = mpsc::channel(4);
+        let result = tokio::time::timeout(
+            Duration::from_secs(30),
+            ScannerIOCycle::nsscanner_with_status(store.as_ref(), ctx, budget, updates, cycle, 11, HealScanMode::Normal),
+        )
+        .await
+        .expect("scan should finish")
+        .expect("scan should succeed");
+        while let Some(mut snapshot) = receiver.recv().await {
+            if result.status != ScannerCycleStatus::Complete || !snapshot.is_complete_bucket_usage_snapshot() {
+                continue;
+            }
+            if usage_matches(&snapshot, retained_writes) {
+                observed_expiration = true;
+                if write_between_scans {
+                    for (bucket, _, _) in &cases {
+                        let mut reader = ScannerPutObjReader::from_vec(vec![0; 42]);
+                        store
+                            .put_object(bucket, &format!("keep/write-{cycle}"), &mut reader, &ScannerObjectOptions::default())
+                            .await
+                            .expect("write between scan observation and publication");
+                        record_bucket_object_write_memory(bucket, None, 42).await;
+                    }
+                    retained_writes += 1;
+                    wait_for_namespace_commit_tails(store.as_ref()).await;
+                }
+                replace_bucket_usage_memory_from_info(&snapshot).await;
+                apply_bucket_usage_memory_overlay(&mut snapshot).await;
+                if usage_matches(&snapshot, retained_writes) {
+                    reconciled_cycles += 1;
+                    reconciled = !write_between_scans || reconciled_cycles >= 3;
+                    break;
+                }
+            }
+        }
+        cycle += 1;
+        if reconciled {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+    assert!(
+        observed_expiration,
+        "{scenario}: complete scans must observe successful lifecycle deletions"
+    );
+    assert!(
+        reconciled,
+        "{scenario}: usage overlay must converge to the complete scanner counts and bytes"
+    );
+    cycle
+}
+
+async fn check_persisted_stale_usage_recovery(store: &Arc<ECStore>, first_cycle: u64) -> u64 {
+    let cases = [("historical-empty", 0_u64), ("historical-retained", 1_u64)];
+    let mut stale = DataUsageInfo {
+        last_update: Some(SystemTime::now()),
+        usage_snapshot_complete: true,
+        usage_snapshot_converged: Some(true),
+        buckets_count: 2,
+        scanner_epoch: Some(11),
+        scanner_cycle: Some(first_cycle - 1),
+        ..Default::default()
+    };
+    for (bucket, retained) in cases {
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create historical usage bucket");
+        if retained > 0 {
+            let mut reader = ScannerPutObjReader::from_vec(vec![0; 42]);
+            store
+                .put_object(bucket, "retained", &mut reader, &ScannerObjectOptions::default())
+                .await
+                .expect("create the only live object before seeding stale usage");
+        }
+        stale.buckets_usage.insert(
+            bucket.to_owned(),
+            BucketUsageInfo {
+                objects_count: 436 + retained,
+                size: 4_587_688_512 + retained * 42,
+                ..Default::default()
+            },
+        );
+        stale.bucket_sizes.insert(bucket.to_owned(), 4_587_688_512 + retained * 42);
+    }
+    stale.calculate_totals();
+    wait_for_namespace_commit_tails(store.as_ref()).await;
+    seed_persisted_usage(stale, store.clone())
+        .await
+        .expect("persist an incorrect complete aggregate");
+    let loaded = load_data_usage_from_backend(store.clone())
+        .await
+        .expect("load the incorrect durable baseline");
+    assert!(loaded.is_complete_bucket_usage_snapshot());
+    for (bucket, retained) in cases {
+        assert_eq!(loaded.buckets_usage[bucket].objects_count, 436 + retained);
+        assert_eq!(loaded.buckets_usage[bucket].size, 4_587_688_512 + retained * 42);
+    }
+    // Model startup from the stale durable aggregate. No object mutation or
+    // expiry receipt is issued after this point, and no corrected snapshot is
+    // constructed by the test. This does not model corrupt per-set caches.
+    replace_bucket_usage_memory_from_info(&loaded).await;
+    for cycle in first_cycle..first_cycle + 2 {
+        let ctx = CancellationToken::new();
+        let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
+        let (updates, receiver) = mpsc::channel(4);
+        let persist = tokio::spawn(crate::scanner::store_data_usage_in_backend(ctx.clone(), store.clone(), receiver));
+        let result = tokio::time::timeout(
+            Duration::from_secs(30),
+            ScannerIOCycle::nsscanner_with_status(store.as_ref(), ctx, budget, updates, cycle, 11, HealScanMode::Normal),
+        )
+        .await
+        .expect("historical recovery scan should finish")
+        .expect("historical recovery scan should succeed");
+        assert_eq!(result.status, ScannerCycleStatus::Complete);
+        tokio::time::timeout(Duration::from_secs(30), persist)
+            .await
+            .expect("scanner publication should finish")
+            .expect("scanner publication task should not panic");
+
+        let durable = load_data_usage_from_backend(store.clone())
+            .await
+            .expect("reload scanner-produced durable usage");
+        assert!(durable.is_complete_bucket_usage_snapshot());
+        assert_eq!(durable.scanner_cycle, Some(cycle));
+        let mut displayed = load_admin_data_usage_from_backend_cached(store.clone())
+            .await
+            .expect("load usage through the admin snapshot path");
+        apply_bucket_usage_memory_overlay(&mut displayed).await;
+        for (bucket, retained) in cases {
+            for snapshot in [&durable, &displayed] {
+                assert_eq!(snapshot.buckets_usage[bucket].objects_count, retained);
+                assert_eq!(snapshot.buckets_usage[bucket].size, retained * 42);
+            }
+        }
+    }
+    first_cycle + 2
+}

crates/scanner/src/storage_api.rs

@@ -346,6 +346,18 @@ pub(crate) mod scanner_io {
     pub(crate) use super::storage_contracts::{BucketInfo, BucketOptions};
     #[cfg(test)]
     pub(crate) use super::storage_contracts::{HTTPRangeSpec, ObjectIO};
+    #[cfg(test)]
+    pub(crate) use rustfs_ecstore::api::{
+        bucket::{
+            lifecycle::bucket_lifecycle_ops::init_background_expiry, metadata::BUCKET_LIFECYCLE_CONFIG,
+            metadata_sys::update as update_bucket_metadata,
+        },
+        data_usage::{
+            apply_bucket_usage_memory_overlay, load_admin_data_usage_from_backend_cached, load_data_usage_from_backend,
+            record_bucket_object_delete_memory, record_bucket_object_write_memory, replace_bucket_usage_memory_from_info,
+            store_data_usage_in_backend as seed_persisted_usage,
+        },
+    };
 }
 
 pub(crate) type ScannerBucketListing = EcstoreScannerBucketListing;

docs/operations/cluster-lifecycle-operations.md

@@ -217,6 +217,7 @@ If a set has lost more drives than its parity, heal cannot rebuild; recover from
 - **Ordinary restart.** No topology change in flight, identical arguments on every node, one node at a time per [rolling-restart.md](rolling-restart.md). Startup reloads pool metadata, `rebalance.bin`, the decommission queue, and heal recovery records, and resumes workers by leader and owner rules.
 - **Rebalance.** Read `/rebalance/status` first. Handle active or stopping pools, `lastError`, cleanup warnings, and stop propagation before deciding to stop or restart. Do not delete `rebalance.bin`.
 - **Decommission.** Non-terminal entries resume; failed and canceled entries survive restarts until cleared or retried through the allowed path (§9).
+- **Scanner pause backlog.** For an unavailable or degraded pause backlog, query `/rustfs/admin/v3/scanner/status` on each node and inspect `pause_backlog.replica_diagnostics`. The entries identify each writable pool/set, its `missing`, `valid`, `invalid`, `future_schema`, or `unavailable` state, object revision (the CAS token for that record), stable and committed generations, committed member count, and any read/decode error. Compare revisions and generations to locate divergent proofs, and compare pool order and endpoints across nodes before diagnosing membership conflicts. Keep the existing backlog records and attach the complete status output when escalating; do not delete or reset `.scanner-pause-backlog.json`. There is no supported force-reset for an authority conflict because it may discard pending data-movement work. A conflict remains fail-closed until an authority-safe recovery is available.
 - **Root heal.** A graceful shutdown persists an unfinished cluster heal as `root-heal-<task-id>.json` in `.rustfs.sys` (`RootHealRecovery` in `crates/heal/src/heal/manager/root_recovery.rs`, 64 KiB bound) and replays it with the same task id. Terminal records use `terminal-root-heal-`; invalid, oversized, or unsupported records move to `quarantined-root-heal-intent-` or `quarantined-root-heal-terminal-` and are skipped with a log line until an operator repairs them. Do not delete quarantine markers or reuse a task id.
 - **Unclean shutdown.** When the previous run left its marker behind, the heal manager's `process_unclean_shutdown` (`crates/heal/src/heal/manager/unclean_shutdown.rs`) enqueues a full erasure-set heal for every local set; graceful shutdown clears the marker. See [durability-modes.md](durability-modes.md) for which sync tier can leave what behind.
 

rustfs/src/connect/diagnostics/job.rs

@@ -1679,6 +1679,7 @@ mod tests {
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[serial_test::serial]
     async fn executes_thread_profile_jobs_against_the_service_process() {
         for scope in [ThreadProfileScope::NativeThreads, ThreadProfileScope::TokioRuntime] {
             let now = Utc::now();

rustfs/src/connect/diagnostics/job_delivery.rs

@@ -44,6 +44,9 @@ const MAX_STATE_BYTES: u64 = 4 * 1024 * 1024;
 const MAX_ARTIFACT_BYTES: usize = 524_288;
 const MAX_STATE_FILES: usize = 65_536;
 
+// The executable identity is stable for this process; avoid hashing the full binary for every profile request.
+static EXECUTABLE_PROVENANCE: tokio::sync::OnceCell<ProfileProvenance> = tokio::sync::OnceCell::const_new();
+
 #[derive(Clone)]
 pub(crate) struct DiagnosticJobRuntime {
     config: HeartbeatConfig,
@@ -386,16 +389,21 @@ fn target_and_identity(config: &HeartbeatConfig) -> Result<(DiagnosticJobTarget,
 }
 
 pub(super) async fn executable_provenance() -> Result<ProfileProvenance, &'static str> {
-    let digest = tokio::task::spawn_blocking(hash_current_executable)
-        .await
-        .map_err(|_| "PROVENANCE_FAILED")?
-        .map_err(|_| "PROVENANCE_FAILED")?;
-    Ok(ProfileProvenance::new(
-        crate::version::build::COMMIT_HASH,
-        digest,
-        env!("CARGO_PKG_VERSION"),
-        enabled_build_features(),
-    ))
+    let provenance = EXECUTABLE_PROVENANCE
+        .get_or_try_init(|| async {
+            let digest = tokio::task::spawn_blocking(hash_current_executable)
+                .await
+                .map_err(|_| "PROVENANCE_FAILED")?
+                .map_err(|_| "PROVENANCE_FAILED")?;
+            Ok::<_, &'static str>(ProfileProvenance::new(
+                crate::version::build::COMMIT_HASH,
+                digest,
+                env!("CARGO_PKG_VERSION"),
+                enabled_build_features(),
+            ))
+        })
+        .await?;
+    Ok(provenance.clone())
 }
 
 fn hash_current_executable() -> Result<String, ()> {

rustfs/src/connect/diagnostics/schedule.rs

@@ -683,13 +683,13 @@ mod tests {
     }
 
     async fn wait_running(status: &mut watch::Receiver<DiagnosticScheduleStatus>) {
-        tokio::time::timeout(Duration::from_secs(2), async {
+        tokio::time::timeout(Duration::from_secs(15), async {
             while !matches!(&*status.borrow(), DiagnosticScheduleStatus::Running { .. }) {
                 status.changed().await.expect("schedule remains active");
             }
         })
         .await
-        .expect("running");
+        .unwrap_or_else(|_| panic!("timed out waiting for schedule to run; current status: {:?}", status.borrow()));
     }
 
     fn blocking_runner() -> Runner {

rustfs/src/connect/diagnostics/trace_runtime.rs

@@ -3144,18 +3144,47 @@ mod tests {
     async fn local_runtime_profile_cancellation_waits_for_lease_release() {
         let state = tempfile::tempdir().unwrap();
         std::fs::set_permissions(state.path(), std::fs::Permissions::from_mode(0o700)).unwrap();
+        let provenance = super::super::job_delivery::executable_provenance()
+            .await
+            .expect("test executable provenance should be available");
+        assert!(provenance.is_valid(), "profile capture requires valid executable provenance");
         let (mut request, _) = runtime_request(state.path());
         request.duration_millis = 5_000;
-        let runtime = spawn_local_trace_capture_runtime(state.path(), &CancellationToken::new()).unwrap();
+        let server_state_root = state.path().to_path_buf();
+        let (ready, wait) = tokio::sync::oneshot::channel();
+        let (stop, stopped) = tokio::sync::oneshot::channel();
+        let server = std::thread::spawn(move || {
+            tokio::runtime::Builder::new_multi_thread()
+                .worker_threads(2)
+                .enable_all()
+                .build()
+                .unwrap()
+                .block_on(async {
+                    let server_shutdown = CancellationToken::new();
+                    let runtime = spawn_local_trace_capture_runtime(&server_state_root, &server_shutdown).unwrap();
+                    ready.send(()).unwrap();
+                    let _ = stopped.await;
+                    runtime.shutdown().await;
+                });
+        });
+        wait.await.unwrap();
         let cancel = CancellationToken::new();
         let task_cancel = cancel.clone();
         let state_root = state.path().to_path_buf();
-        let task = tokio::spawn(async move { super::request_local_runtime_profile(&state_root, request, &task_cancel).await });
+        let mut task =
+            tokio::spawn(async move { super::request_local_runtime_profile(&state_root, request, &task_cancel).await });
         tokio::time::timeout(Duration::from_secs(30), async {
             loop {
                 if super::super::profile_cpu::CollectorLease::acquire().is_err() {
                     break;
                 }
+                if task.is_finished() {
+                    let detail = match (&mut task).await.expect("request task should not panic") {
+                        Ok(_) => "request completed successfully".to_owned(),
+                        Err(error) => format!("request failed: {error}"),
+                    };
+                    panic!("runtime profile finished before acquiring its collector lease: {detail}");
+                }
                 tokio::time::sleep(Duration::from_millis(5)).await;
             }
         })
@@ -3165,7 +3194,8 @@ mod tests {
         assert!(matches!(task.await.unwrap(), Err(LocalTraceCaptureError::RuntimeProfile(code)) if code == "CANCELLED"));
         let lease = super::super::profile_cpu::CollectorLease::acquire().expect("client cancellation joins collector");
         drop(lease);
-        runtime.shutdown().await;
+        stop.send(()).unwrap();
+        server.join().unwrap();
     }
     #[tokio::test]
     #[serial]