Merge branch 'main' into fix/scanner-preserve-maintenance-digest

crates/scanner/src/scanner.rs

@@ -1910,10 +1910,10 @@ where
         .as_ref()
         .map(|(notification_system, grants)| (Arc::clone(notification_system), grants.clone()));
     let remote_lease_release_safe = Arc::new(AtomicBool::new(true));
-    let mut usage_publication_result = match publication_defer_reason {
+    let mut usage_persist_outcome = match publication_defer_reason {
         Some(reason) => {
             drop(receiver);
-            DataUsagePublicationResult::from(DataUsagePersistOutcome::Deferred(reason))
+            DataUsagePersistOutcome::Deferred(reason)
         }
         None => {
             // ScannerIO emits its complete or observational update only after
@@ -1928,11 +1928,6 @@ where
                 .as_ref()
                 .map(|(_, grants)| grants.iter().map(|grant| grant.lease.token).collect())
                 .unwrap_or_default();
-            let ack_expectation = scan_result
-                .as_ref()
-                .ok()
-                .filter(|result| result.has_dirty_usage_to_acknowledge())
-                .and_then(ScannerCycleResult::publication_expectation);
             let mut usage_persist_task = AbortOnDropHandle::new(tokio::spawn(async move {
                 store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence(
                     ctx_clone,
@@ -1945,7 +1940,6 @@ where
                         remote_lease_deadline,
                         remote_lease_fence,
                     )
-                    .with_ack_expectation(ack_expectation)
                     .with_remote_lease_tokens(remote_lease_tokens)
                     .with_lease_release_flag(remote_lease_release_safe_for_task),
                     move || {
@@ -1979,7 +1973,7 @@ where
                         error = %err,
                         "Scanner data usage persistence task failed"
                     );
-                    DataUsagePublicationResult::from(DataUsagePersistOutcome::Failed)
+                    DataUsagePersistOutcome::Failed
                 }
                 DataUsagePersistTaskResult::Cancelled => {
                     debug!(
@@ -1991,7 +1985,7 @@ where
                         state = "usage_persist_task_cancelled",
                         "Scanner data usage persistence task cancelled"
                     );
-                    DataUsagePublicationResult::from(DataUsagePersistOutcome::Failed)
+                    DataUsagePersistOutcome::Failed
                 }
                 DataUsagePersistTaskResult::TimedOut => {
                     error!(
@@ -2004,12 +1998,11 @@ where
                         state = "usage_persist_task_timed_out",
                         "Scanner data usage persistence task timed out"
                     );
-                    DataUsagePublicationResult::from(DataUsagePersistOutcome::Failed)
+                    DataUsagePersistOutcome::Failed
                 }
             }
         }
     };
-    let mut usage_persist_outcome = usage_publication_result.outcome();
     let lease_expired = remote_publication_leases
         .as_ref()
         .is_some_and(|(_, grants)| grants.iter().any(|grant| !grant.lease.is_valid()));
@@ -2209,9 +2202,8 @@ where
         };
     }
 
-    usage_publication_result.restrict_outcome(usage_persist_outcome);
     let (completion_outcome, scanner_pending_maintenance_work, remote_dirty_usage_acknowledgements) =
-        finalize_scanner_cycle_result(scan_cycle_result, usage_publication_result);
+        finalize_scanner_cycle_result(scan_cycle_result, usage_persist_outcome);
     let remote_dirty_usage_pending = if remote_dirty_usage_acknowledgements.is_empty() {
         false
     } else if let Some(notification_system) = storeapi.scanner_notification_system() {
@@ -3445,35 +3437,21 @@ fn scanner_cycle_completion_outcome(
 
 fn finalize_scanner_cycle_result(
     scan_cycle_result: crate::scanner_io::ScannerCycleResult,
-    publication: DataUsagePublicationResult,
+    usage_persist_outcome: DataUsagePersistOutcome,
 ) -> (ScannerCycleOutcome, bool, Vec<ScannerDirtyUsageAcknowledgement>) {
-    let (usage_persist_outcome, proof) = publication.into_parts();
     let completion_outcome = scanner_cycle_completion_outcome_for_result(&scan_cycle_result, usage_persist_outcome);
+    let pending_maintenance_work = scan_cycle_result.has_pending_maintenance_work();
     let durable_complete_snapshot = scan_cycle_result.status == ScannerCycleStatus::Complete
         && matches!(
             usage_persist_outcome,
             DataUsagePersistOutcome::Saved | DataUsagePersistOutcome::AlreadyDurable
-        )
-        && scan_cycle_result.publication_expectation().as_ref().is_some_and(|expected| {
-            proof
-                .as_ref()
-                .is_some_and(|proof| proof.verified_version_for(expected).is_some())
-        });
-    let pending_maintenance_work = scan_cycle_result.has_pending_maintenance_work()
-        || (scan_cycle_result.has_dirty_usage_to_acknowledge() && !durable_complete_snapshot);
+        );
     let remote_dirty_usage_acknowledgements = if durable_complete_snapshot {
-        match proof {
-            Some(proof) => scan_cycle_result.acknowledge_durable_usage(&proof),
-            None => Vec::new(),
-        }
+        scan_cycle_result.acknowledge_durable_usage()
     } else {
         Vec::new()
     };
-    (
-        completion_outcome,
-        pending_maintenance_work || crate::scanner_io::dirty_usage_buckets_pending(),
-        remote_dirty_usage_acknowledgements,
-    )
+    (completion_outcome, pending_maintenance_work, remote_dirty_usage_acknowledgements)
 }
 
 fn scanner_cycle_completion_outcome_for_result(
@@ -3573,7 +3551,6 @@ use activity::*;
 use backlog::*;
 use cycle_state::*;
 use leadership::*;
-pub(crate) use usage_store::RootPublicationProof;
 use usage_store::*;
 
 pub use activity::scanner_topology_digest;

crates/scanner/src/scanner/tests.rs

@@ -37,7 +37,6 @@ use tokio::time::{Duration, advance};
 const TEST_DEFAULT_SCANNER_CYCLE_SECS: u64 = 24 * 60 * 60;
 
 mod recovery_control;
-mod scoped_ack_publication;
 
 async fn setup_scanner_cycle_store() -> (tempfile::TempDir, Arc<ECStore>) {
     setup_scanner_cycle_store_with_usage_baseline(true).await
@@ -6185,7 +6184,7 @@ async fn coordinator_classifies_an_expired_publication_lease() {
         .await;
 
     assert_eq!(
-        outcome.outcome(),
+        outcome,
         DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded)
     );
     assert!(store.put_counts.lock().await.is_empty(), "expired lease must prevent a PUT");
@@ -7326,7 +7325,7 @@ fn scanner_cycle_cache_floor_stays_pending_during_deferred_usage_publication() {
 
 #[test]
 #[serial]
-fn finalizing_a_saved_enum_without_proof_keeps_dirty_pending() {
+fn finalizing_a_saved_cycle_acknowledges_its_exact_dirty_snapshot() {
     crate::scanner_io::clear_dirty_usage_bucket("photos");
     crate::scanner_io::record_dirty_usage_bucket("photos");
     let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests();
@@ -7338,19 +7337,17 @@ fn finalizing_a_saved_enum_without_proof_keeps_dirty_pending() {
     };
     let unsaved = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot.clone()))
         .with_remote_dirty_usage_acknowledgements(vec![remote_acknowledgement.clone()]);
-    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(unsaved, DataUsagePersistOutcome::NoUpdate.into());
+    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(unsaved, DataUsagePersistOutcome::NoUpdate);
     assert_eq!(outcome, ScannerCycleOutcome::Failed);
     assert!(acknowledgements.is_empty());
     assert!(crate::scanner_io::dirty_usage_buckets_pending());
 
     let saved = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot))
-        .with_remote_dirty_usage_acknowledgements(vec![remote_acknowledgement]);
-    let (outcome, pending, acknowledgements) = finalize_scanner_cycle_result(saved, DataUsagePersistOutcome::Saved.into());
+        .with_remote_dirty_usage_acknowledgements(vec![remote_acknowledgement.clone()]);
+    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(saved, DataUsagePersistOutcome::Saved);
     assert_eq!(outcome, ScannerCycleOutcome::Completed);
-    assert!(acknowledgements.is_empty());
-    assert!(pending);
-    assert!(crate::scanner_io::dirty_usage_buckets_pending());
-    crate::scanner_io::clear_dirty_usage_bucket("photos");
+    assert_eq!(acknowledgements, vec![remote_acknowledgement]);
+    assert!(!crate::scanner_io::dirty_usage_buckets_pending());
 }
 
 #[test]
@@ -7362,7 +7359,7 @@ fn finalizing_a_deferred_usage_save_keeps_dirty_work_pending() {
     let deferred = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot));
 
     let (outcome, _, acknowledgements) =
-        finalize_scanner_cycle_result(deferred, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement).into());
+        finalize_scanner_cycle_result(deferred, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement));
 
     assert_eq!(outcome, ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::DataMovement));
     assert!(acknowledgements.is_empty());
@@ -7382,7 +7379,7 @@ fn finalizing_post_scan_observation_advances_partially_without_dirty_ack() {
     )
     .with_observational_snapshot_published(true);
 
-    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(observed, DataUsagePersistOutcome::Saved.into());
+    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(observed, DataUsagePersistOutcome::Saved);
 
     assert_eq!(outcome, ScannerCycleOutcome::Partial);
     assert!(acknowledgements.is_empty());
@@ -7418,20 +7415,17 @@ async fn scanner_cycle_keeps_remote_pending_acknowledgement() {
 
 #[test]
 #[serial]
-fn finalizing_an_already_durable_enum_without_proof_keeps_dirty_pending() {
+fn finalizing_an_already_durable_cycle_acknowledges_its_exact_dirty_snapshot() {
     crate::scanner_io::clear_dirty_usage_bucket("photos");
     crate::scanner_io::record_dirty_usage_bucket("photos");
     let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests();
 
     let durable = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot));
-    let (outcome, pending, acknowledgements) =
-        finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::AlreadyDurable.into());
+    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::AlreadyDurable);
 
     assert_eq!(outcome, ScannerCycleOutcome::Completed);
     assert!(acknowledgements.is_empty());
-    assert!(pending);
-    assert!(crate::scanner_io::dirty_usage_buckets_pending());
-    crate::scanner_io::clear_dirty_usage_bucket("photos");
+    assert!(!crate::scanner_io::dirty_usage_buckets_pending());
 }
 
 #[test]
@@ -7442,8 +7436,7 @@ fn finalizing_a_prior_same_cycle_snapshot_keeps_new_dirty_work_pending() {
     let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests();
 
     let durable = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot));
-    let (outcome, _, acknowledgements) =
-        finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::PriorCycleDurable.into());
+    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::PriorCycleDurable);
 
     assert_eq!(outcome, ScannerCycleOutcome::Completed);
     assert!(acknowledgements.is_empty());
@@ -7459,7 +7452,7 @@ fn finalizing_a_durable_superseded_snapshot_keeps_dirty_work_pending() {
     let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests();
 
     let superseded = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Superseded, Some(dirty_snapshot));
-    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(superseded, DataUsagePersistOutcome::Saved.into());
+    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(superseded, DataUsagePersistOutcome::Saved);
 
     assert_eq!(outcome, ScannerCycleOutcome::Superseded);
     assert!(acknowledgements.is_empty());
@@ -8888,7 +8881,7 @@ fn post_lease_activity_proof_rejects_a_put_tail_that_finished_before_lease_acqui
     ]);
     let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(
         result,
-        DataUsagePersistOutcome::Deferred(reason.expect("changed namespace should defer publication")).into(),
+        DataUsagePersistOutcome::Deferred(reason.expect("changed namespace should defer publication")),
     );
     assert_eq!(
         outcome,

crates/scanner/src/scanner/tests/scoped_ack_publication.rs

@@ -1,490 +0,0 @@
-// Copyright 2026 RustFS Team
-// Licensed under the Apache License, Version 2.0.
-
-use super::super::usage_store::DataUsagePublicationResult;
-use super::*;
-use crate::scanner_io::ScannerBucketScanScope;
-use rustfs_utils::path::path_join_buf;
-use sha2::Digest;
-use std::time::SystemTime;
-
-const PROOF_BUCKET: &str = "publication-proof-bucket";
-const PROOF_EPOCH: u64 = 7;
-const PROOF_CYCLE: u64 = 11;
-
-async fn settle_namespace_commits(store: &ECStore) {
-    tokio::time::timeout(Duration::from_secs(30), async {
-        while store.scanner_data_usage_publication_blocked().await {
-            tokio::time::sleep(Duration::from_millis(1)).await;
-        }
-    })
-    .await
-    .expect("fixture namespace commits must settle before collecting complete coverage");
-}
-
-async fn complete_candidate(store: &Arc<ECStore>, cycle: u64) -> (crate::scanner_io::ScannerCycleResult, DataUsageInfo) {
-    settle_namespace_commits(store).await;
-    let ctx = CancellationToken::new();
-    let budget = ScannerCycleBudget::new_with_progress_tracking(
-        &ctx,
-        ScannerCycleBudgetConfig {
-            max_objects: Some(8),
-            ..Default::default()
-        },
-    );
-    let (updates, mut receiver) = mpsc::channel(1);
-    let result = crate::scanner_io::nsscanner_with_storage_status_scoped(
-        store.as_ref(),
-        crate::scanner_io::ScannerCycleRequest {
-            ctx,
-            budget,
-            updates,
-            want_cycle: cycle,
-            leader_epoch: PROOF_EPOCH,
-            scan_mode: HealScanMode::Normal,
-            scan_scope: ScannerBucketScanScope::default(),
-            persisted_usage_baseline: None,
-            observed_usage_candidate: None,
-            requires_full_scan: true,
-            service_cohort: None,
-            resolved_scope_observer: None,
-        },
-    )
-    .await
-    .expect("real scanner must produce the fixture candidate");
-    assert_eq!(result.status, ScannerCycleStatus::Complete);
-    let candidate = receiver.recv().await.expect("complete scanner snapshot");
-    assert!(candidate.usage_snapshot_complete);
-    assert_eq!(candidate.usage_snapshot_converged, Some(true));
-    assert_eq!(candidate.scanner_cycle, Some(cycle));
-    assert_eq!(candidate.scanner_epoch, Some(PROOF_EPOCH));
-    (result, candidate)
-}
-
-async fn candidate_store() -> (tempfile::TempDir, Arc<ECStore>) {
-    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
-    let (directory, store) = setup_scanner_cycle_store_with_usage_baseline(false).await;
-    store
-        .make_bucket(PROOF_BUCKET, &crate::storage_api::scan::MakeBucketOptions::default())
-        .await
-        .expect("create proof fixture bucket through the owner");
-    let mut reader = PutObjReader::from_vec(b"proof".to_vec());
-    store.pools[0].disk_set[0]
-        .put_object(PROOF_BUCKET, "initial", &mut reader, &ObjectOptions::default())
-        .await
-        .expect("persist fixture object through the owner");
-    crate::scanner_io::record_dirty_usage_bucket(PROOF_BUCKET);
-    settle_namespace_commits(&store).await;
-    (directory, store)
-}
-
-async fn read_root(store: &Arc<ECStore>) -> (Option<Vec<u8>>, DataUsageCacheRevision) {
-    read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
-        .await
-        .expect("read actual v2 root bytes and revision")
-}
-
-async fn publish_candidate(
-    store: &Arc<ECStore>,
-    scan: &crate::scanner_io::ScannerCycleResult,
-    candidate: DataUsageInfo,
-    baseline: Option<DataUsagePersistBaseline>,
-) -> DataUsagePublicationResult {
-    let expectation = scan.publication_expectation();
-    assert!(expectation.is_some(), "only a real complete scan may supply the expectation");
-    let (sender, receiver) = mpsc::channel(1);
-    sender.send(candidate).await.expect("enqueue the real scan candidate");
-    drop(sender);
-    store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence(
-        CancellationToken::new(),
-        store.clone(),
-        receiver,
-        Some(PROOF_EPOCH),
-        baseline,
-        ScannerPublicationFence::new(scan.publication_epoch(), None, None).with_ack_expectation(expectation),
-        || async { None },
-    )
-    .await
-}
-
-#[tokio::test]
-#[serial]
-async fn scoped_ack_publication_companion_only_does_not_authorize_root_ack() {
-    for companion in [
-        format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()),
-        LEGACY_DATA_USAGE_OBJ_NAME_PATH.to_string(),
-        format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
-    ] {
-        let (_directory, store) = candidate_store().await;
-        let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
-        let bytes = serde_json::to_vec(&candidate).expect("actual candidate JSON");
-        save_config(store.clone(), &companion, bytes.clone())
-            .await
-            .expect("persist the companion on real disks");
-        let baseline = read_data_usage_persist_baseline(store.clone())
-            .await
-            .expect("companion fallback baseline");
-        assert_eq!(baseline.data.as_deref(), Some(bytes.as_slice()));
-        assert_eq!(baseline.revision, DataUsageCacheRevision::Missing);
-        assert_eq!(read_root(&store).await.0, None);
-        let dirty = crate::scanner_io::dirty_usage_buckets_for_tests();
-
-        let publication = publish_candidate(&store, &scan, candidate, Some(baseline)).await;
-        assert_eq!(publication.outcome(), DataUsagePersistOutcome::AlreadyDurable);
-        let (_, pending, acknowledgements) = finalize_scanner_cycle_result(scan, publication);
-        assert!(pending, "unacknowledged durable companion work must remain pending");
-        assert!(acknowledgements.is_empty());
-        assert_eq!(
-            crate::scanner_io::dirty_usage_buckets_for_tests(),
-            dirty,
-            "a companion is not the v2 root target"
-        );
-        assert_eq!(read_root(&store).await, (None, DataUsageCacheRevision::Missing));
-        assert_eq!(read_config(store.clone(), &companion).await.expect("companion retained"), bytes);
-    }
-    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
-}
-
-#[tokio::test]
-#[serial]
-async fn scoped_ack_publication_actual_root_readback_accepts_semantic_json_equivalence() {
-    let (_directory, store) = candidate_store().await;
-    let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
-    let canonical = serde_json::to_vec(&candidate).expect("candidate encoding");
-    let mut value = serde_json::to_value(&candidate).expect("candidate value");
-    value
-        .as_object_mut()
-        .expect("usage object")
-        .insert("fixture_unknown_field".into(), serde_json::json!({"retained": true}));
-    let different_bytes = serde_json::to_vec_pretty(&value).expect("noncanonical primary JSON");
-    assert_ne!(different_bytes, canonical);
-    assert_eq!(
-        serde_json::from_slice::<DataUsageInfo>(&different_bytes).expect("semantic primary"),
-        candidate
-    );
-    save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), different_bytes.clone())
-        .await
-        .expect("persist actual primary representation");
-    let before = read_root(&store).await;
-    assert!(matches!(&before.1, DataUsageCacheRevision::Etag(etag) if !etag.is_empty()));
-    let baseline = read_data_usage_persist_baseline(store.clone())
-        .await
-        .expect("real primary revision");
-    assert!(crate::scanner_io::dirty_usage_buckets_pending());
-
-    let publication = publish_candidate(&store, &scan, candidate.clone(), Some(baseline)).await;
-    assert_eq!(publication.outcome(), DataUsagePersistOutcome::AlreadyDurable);
-    let (_, proof) = publication.into_parts();
-    let proof = proof.expect("actual primary readback must produce its own root proof");
-    let expected = scan.publication_expectation().expect("real scan expectation");
-    let (etag, raw_digest) = proof.verified_version_for(&expected).expect("proof must bind this candidate");
-    let DataUsageCacheRevision::Etag(expected_etag) = &before.1 else { panic!("actual root ETag") };
-    assert_eq!(etag, expected_etag);
-    let expected_digest: [u8; 32] = sha2::Sha256::digest(&different_bytes).into();
-    assert_eq!(
-        *raw_digest, expected_digest,
-        "proof must record actual bytes, not reserialized candidate bytes"
-    );
-    // Obtain another proof through the same real readback path rather than
-    // fabricating a publication result from the inspected proof above.
-    let publication = publish_candidate(&store, &scan, candidate, None).await;
-    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(scan, publication);
-    assert_eq!(outcome, ScannerCycleOutcome::Completed);
-    assert!(acknowledgements.is_empty(), "the single-node fixture has no remote targets");
-    assert!(
-        !crate::scanner_io::dirty_usage_buckets_pending(),
-        "actual root bytes plus a real revision authorize this scan"
-    );
-    assert_eq!(read_root(&store).await, before, "readback must not rewrite unknown fields or whitespace");
-}
-
-#[tokio::test]
-#[serial]
-async fn scoped_ack_publication_successful_root_cas_authorizes_its_scan() {
-    let (_directory, store) = candidate_store().await;
-    let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
-    let baseline = read_data_usage_persist_baseline(store.clone())
-        .await
-        .expect("initial root revision");
-    assert_eq!(baseline.revision, DataUsageCacheRevision::Missing);
-    assert!(crate::scanner_io::dirty_usage_buckets_pending());
-    let publication = publish_candidate(&store, &scan, candidate.clone(), Some(baseline)).await;
-    assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved);
-    let (bytes, revision) = read_root(&store).await;
-    assert!(matches!(revision, DataUsageCacheRevision::Etag(etag) if !etag.is_empty()));
-    assert_eq!(
-        serde_json::from_slice::<DataUsageInfo>(&bytes.expect("actual saved root")).expect("root JSON"),
-        candidate
-    );
-    let (outcome, pending, acknowledgements) = finalize_scanner_cycle_result(scan, publication);
-    assert_eq!(outcome, ScannerCycleOutcome::Completed);
-    assert!(!pending);
-    assert!(acknowledgements.is_empty(), "the single-node fixture has no remote targets");
-    assert!(
-        !crate::scanner_io::dirty_usage_buckets_pending(),
-        "the real root CAS must authorize its matching scan"
-    );
-}
-
-#[tokio::test]
-#[serial]
-async fn scoped_ack_publication_observed_candidate_reuse_requires_a_new_root_proof() {
-    let (_directory, store) = candidate_store().await;
-    let bootstrap = scanner_usage_bootstrap_marker(SystemTime::now(), Some(PROOF_EPOCH));
-    save_config(
-        store.clone(),
-        DATA_USAGE_OBJ_NAME_PATH.as_str(),
-        serde_json::to_vec(&bootstrap).expect("bootstrap root encoding"),
-    )
-    .await
-    .expect("persist authoritative bootstrap root");
-    let (prior_scan, mut observed_candidate) = complete_candidate(&store, PROOF_CYCLE).await;
-    // Seed a complete but unconverged observation from real scanner coverage;
-    // the production writer attaches its authoritative baseline identity.
-    observed_candidate.usage_snapshot_converged = Some(false);
-    let observation = publish_candidate(&store, &prior_scan, observed_candidate, None).await;
-    let (outcome, proof) = observation.into_parts();
-    assert_eq!(outcome, DataUsagePersistOutcome::Saved);
-    assert!(proof.is_none(), "an observational write cannot authorize a root ACK");
-    let (root_before, revision_before) = read_root(&store).await;
-    let observed = read_config(store.clone(), DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str())
-        .await
-        .expect("read real persisted observation");
-    let ctx = CancellationToken::new();
-    let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
-    let (updates, mut receiver) = mpsc::channel(1);
-    let (observer, selected) = tokio::sync::oneshot::channel();
-    let scan = crate::scanner_io::nsscanner_with_storage_status_scoped(
-        store.as_ref(),
-        crate::scanner_io::ScannerCycleRequest {
-            ctx,
-            budget,
-            updates,
-            want_cycle: PROOF_CYCLE + 1,
-            leader_epoch: PROOF_EPOCH,
-            scan_mode: HealScanMode::Normal,
-            scan_scope: ScannerBucketScanScope::default(),
-            persisted_usage_baseline: root_before.clone().map(Bytes::from),
-            observed_usage_candidate: Some(Bytes::from(observed)),
-            requires_full_scan: false,
-            service_cohort: None,
-            resolved_scope_observer: Some(observer),
-        },
-    )
-    .await
-    .expect("observation-backed scope must run through the real scanner");
-    let scope = selected.await.expect("production resolver decision");
-    assert_eq!(scope.selected_buckets_for_tests(), Some(&HashSet::from([PROOF_BUCKET.to_string()])));
-    assert_eq!(scan.status, ScannerCycleStatus::Complete);
-    let expectation = scan.publication_expectation().expect("reused coverage must be revalidated");
-    assert!(
-        !expectation.same_candidate(&prior_scan.publication_expectation().expect("prior real candidate")),
-        "the observation cannot transfer the previous scan's expectation"
-    );
-    assert_eq!(read_root(&store).await, (root_before, revision_before));
-    assert!(crate::scanner_io::dirty_usage_buckets_pending());
-    let candidate = receiver.recv().await.expect("new validated root candidate");
-    assert_eq!(candidate.scanner_cycle, Some(PROOF_CYCLE + 1));
-    assert_eq!(candidate.usage_snapshot_converged, Some(true));
-    let publication = publish_candidate(&store, &scan, candidate, None).await;
-    assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved);
-    let (outcome, pending, acknowledgements) = finalize_scanner_cycle_result(scan, publication);
-    assert_eq!(outcome, ScannerCycleOutcome::Completed);
-    assert!(!pending);
-    assert!(acknowledgements.is_empty());
-    assert!(!crate::scanner_io::dirty_usage_buckets_pending());
-}
-
-#[tokio::test]
-#[serial]
-async fn scoped_ack_publication_stale_root_cas_keeps_dirty_after_bucket_save() {
-    let (_directory, store) = candidate_store().await;
-    let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
-    let mut bucket_cache = DataUsageCache::default();
-    bucket_cache
-        .load(store.pools[0].disk_set[0].clone(), &path_join_buf(&[PROOF_BUCKET, DATA_USAGE_CACHE_NAME]))
-        .await
-        .expect("real bucket checkpoint must be persisted before root publication");
-    assert!(bucket_cache.info.snapshot_complete);
-    assert_eq!(
-        bucket_cache
-            .checked_flatten(PROOF_BUCKET)
-            .expect("persisted bucket root")
-            .objects,
-        1
-    );
-    let stale_baseline = read_data_usage_persist_baseline(store.clone())
-        .await
-        .expect("missing root revision");
-    assert_eq!(stale_baseline.revision, DataUsageCacheRevision::Missing);
-    let mut competing = candidate.clone();
-    competing.scanner_epoch = Some(PROOF_EPOCH + 1);
-    competing.scanner_cycle = Some(PROOF_CYCLE + 1);
-    for state in &mut competing.usage_snapshot_set_states {
-        state.scanner_epoch = Some(PROOF_EPOCH + 1);
-        state.scanner_cycle = Some(PROOF_CYCLE + 1);
-    }
-    let competing_bytes = serde_json::to_vec(&competing).expect("competing root");
-    save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), competing_bytes.clone())
-        .await
-        .expect("another publisher wins the actual root slot");
-    let before = read_root(&store).await;
-    let dirty = crate::scanner_io::dirty_usage_buckets_for_tests();
-
-    let publication = publish_candidate(&store, &scan, candidate, Some(stale_baseline)).await;
-    assert_eq!(
-        publication.outcome(),
-        DataUsagePersistOutcome::Current,
-        "the old missing revision loses CAS and reconciles the newer root"
-    );
-    let (_, _, acknowledgements) = finalize_scanner_cycle_result(scan, publication);
-    assert!(acknowledgements.is_empty());
-    assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty);
-    assert_eq!(
-        read_root(&store).await,
-        before,
-        "bucket durability must not authorize replacing the winning root"
-    );
-    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
-}
-
-#[tokio::test]
-#[serial]
-async fn scoped_ack_publication_cannot_transfer_proof_between_real_scan_results() {
-    let (_directory, store) = candidate_store().await;
-    let (first_scan, first_candidate) = complete_candidate(&store, PROOF_CYCLE).await;
-    let (second_scan, second_candidate) = complete_candidate(&store, PROOF_CYCLE).await;
-    assert_eq!(first_candidate.scanner_epoch, second_candidate.scanner_epoch);
-    assert_eq!(first_candidate.scanner_cycle, second_candidate.scanner_cycle);
-    assert_eq!(first_candidate.objects_total_count, second_candidate.objects_total_count);
-    let baseline = read_data_usage_persist_baseline(store.clone())
-        .await
-        .expect("initial root revision");
-    let dirty = crate::scanner_io::dirty_usage_buckets_for_tests();
-    let publication = publish_candidate(&store, &first_scan, first_candidate, Some(baseline)).await;
-    assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved);
-    assert!(read_root(&store).await.0.is_some(), "the first scan really published its root");
-
-    let (_, pending, acknowledgements) = finalize_scanner_cycle_result(second_scan, publication);
-    assert!(pending, "another scan's publication must not finish this scan's dirty maintenance work");
-    assert!(acknowledgements.is_empty());
-    assert_eq!(
-        crate::scanner_io::dirty_usage_buckets_for_tests(),
-        dirty,
-        "same counters and cycle cannot transfer another scan's proof"
-    );
-
-    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
-}
-
-#[tokio::test]
-#[serial]
-async fn scoped_ack_publication_stale_baseline_cannot_prove_a_replaced_root() {
-    let (_directory, store) = candidate_store().await;
-    let (first_scan, first_candidate) = complete_candidate(&store, PROOF_CYCLE).await;
-    save_config(
-        store.clone(),
-        DATA_USAGE_OBJ_NAME_PATH.as_str(),
-        serde_json::to_vec(&first_candidate).expect("first candidate"),
-    )
-    .await
-    .expect("persist the first candidate on real disks");
-    let stale_baseline = read_data_usage_persist_baseline(store.clone())
-        .await
-        .expect("capture the genuine first root revision");
-    let dirty = crate::scanner_io::dirty_usage_buckets_for_tests();
-    let mut reader = PutObjReader::from_vec(b"second".to_vec());
-    store.pools[0].disk_set[0]
-        .put_object(PROOF_BUCKET, "second", &mut reader, &ObjectOptions::default())
-        .await
-        .expect("commit a real namespace change");
-    assert_eq!(
-        crate::scanner_io::dirty_usage_buckets_for_tests(),
-        dirty,
-        "direct storage writes leave this fixture's scanner hint generation unchanged"
-    );
-    let (_, replacement) = complete_candidate(&store, PROOF_CYCLE).await;
-    assert_eq!(first_candidate.scanner_epoch, replacement.scanner_epoch);
-    assert_eq!(first_candidate.scanner_cycle, replacement.scanner_cycle);
-    assert_eq!((first_candidate.objects_total_count, replacement.objects_total_count), (1, 2));
-    save_config(
-        store.clone(),
-        DATA_USAGE_OBJ_NAME_PATH.as_str(),
-        serde_json::to_vec(&replacement).expect("replacement candidate"),
-    )
-    .await
-    .expect("publish the replacement root");
-    let current = read_root(&store).await;
-    assert_ne!(current.1, stale_baseline.revision);
-
-    // The supplied baseline still equals candidate A, but the actual target
-    // now contains B. Compatibility's AlreadyDurable outcome is not proof.
-    let publication = publish_candidate(&store, &first_scan, first_candidate, Some(stale_baseline)).await;
-    assert_eq!(publication.outcome(), DataUsagePersistOutcome::AlreadyDurable);
-    let (_, pending, acknowledgements) = finalize_scanner_cycle_result(first_scan, publication);
-    assert!(pending);
-    assert!(acknowledgements.is_empty());
-    assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty);
-    assert_eq!(read_root(&store).await, current);
-    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
-}
-
-#[tokio::test]
-#[serial]
-async fn scoped_ack_publication_rejects_builder_mutation_after_real_root_publish() {
-    for mutation in ["remote_ack_target", "publication_epoch", "remote_lease_targets"] {
-        let (_directory, store) = candidate_store().await;
-        let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
-        let baseline = read_data_usage_persist_baseline(store.clone())
-            .await
-            .expect("initial root revision");
-        let dirty = crate::scanner_io::dirty_usage_buckets_for_tests();
-        let changed_generation = dirty
-            .get(PROOF_BUCKET)
-            .expect("the real scan has dirty work")
-            .checked_add(1)
-            .expect("bounded fixture generation");
-        let changed_epoch = scan
-            .publication_epoch()
-            .expect("real scan publication epoch")
-            .checked_add(1)
-            .expect("bounded fixture epoch");
-        let publication = publish_candidate(&store, &scan, candidate.clone(), Some(baseline)).await;
-        assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved, "{mutation}");
-        let root_before = read_root(&store).await;
-        assert_eq!(
-            serde_json::from_slice::<DataUsageInfo>(root_before.0.as_deref().expect("actual saved root"))
-                .expect("persisted root JSON"),
-            candidate,
-            "{mutation}: the original candidate really reached root storage"
-        );
-
-        let changed = match mutation {
-            "remote_ack_target" => scan.with_remote_dirty_usage_acknowledgements(vec![ScannerDirtyUsageAcknowledgement {
-                host: "proof-peer:9000".to_string(),
-                instance_id: crate::scanner_activity_epoch().to_string(),
-                generation: changed_generation,
-            }]),
-            "publication_epoch" => scan.with_publication_epoch(Some(changed_epoch)),
-            "remote_lease_targets" => scan.with_remote_publication_lease_targets(vec![(
-                "proof-peer:9000".to_string(),
-                crate::scanner_activity_epoch().to_string(),
-                changed_generation,
-            )]),
-            _ => unreachable!("fixed mutation cases"),
-        };
-        let (_, pending, acknowledgements) = finalize_scanner_cycle_result(changed, publication);
-        assert!(
-            acknowledgements.is_empty(),
-            "{mutation}: the old root proof must not authorize changed ACK work"
-        );
-        assert!(pending, "{mutation}: changed maintenance work must remain pending");
-        assert_eq!(
-            crate::scanner_io::dirty_usage_buckets_for_tests(),
-            dirty,
-            "{mutation}: the changed scan must not clear local dirty work"
-        );
-        assert_eq!(read_root(&store).await, root_before, "{mutation}: the durable original root is retained");
-    }
-    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
-}

crates/scanner/src/scanner/usage_store.rs

@@ -34,139 +34,6 @@ pub(super) enum DataUsagePersistOutcome {
     Failed,
 }
 
-#[derive(Debug)]
-pub(crate) struct RootPublicationProof {
-    candidate: crate::scanner_io::ScannerPublicationExpectation,
-    root_version: (String, [u8; 32]),
-}
-
-impl RootPublicationProof {
-    pub(crate) fn verified_version_for(
-        &self,
-        expected: &crate::scanner_io::ScannerPublicationExpectation,
-    ) -> Option<&(String, [u8; 32])> {
-        self.candidate.same_candidate(expected).then_some(&self.root_version)
-    }
-}
-
-#[derive(Debug)]
-pub(super) struct DataUsagePublicationResult {
-    outcome: DataUsagePersistOutcome,
-    proof: Option<RootPublicationProof>,
-}
-
-impl From<DataUsagePersistOutcome> for DataUsagePublicationResult {
-    fn from(outcome: DataUsagePersistOutcome) -> Self {
-        Self { outcome, proof: None }
-    }
-}
-
-impl DataUsagePublicationResult {
-    pub(super) fn outcome(&self) -> DataUsagePersistOutcome {
-        self.outcome
-    }
-    pub(super) fn restrict_outcome(&mut self, outcome: DataUsagePersistOutcome) {
-        if outcome != self.outcome {
-            self.proof = None;
-        }
-        self.outcome = outcome;
-    }
-    pub(super) fn into_parts(self) -> (DataUsagePersistOutcome, Option<RootPublicationProof>) {
-        (self.outcome, self.proof)
-    }
-}
-
-fn root_ack_write_is_confirmed<T, E>(
-    result: &std::result::Result<T, E>,
-    state: Option<ScannerPublicationCommitState>,
-    written_etag: Option<&str>,
-) -> bool {
-    result.is_ok() && state == Some(ScannerPublicationCommitState::Committed) && written_etag.is_some_and(|etag| !etag.is_empty())
-}
-
-#[cfg(test)]
-mod root_publication_confirmation_tests {
-    use super::*;
-
-    #[test]
-    fn root_publication_confirmation_requires_committed_state_and_write_revision() {
-        let saved = Ok::<(), ()>(());
-        for state in [
-            None,
-            Some(ScannerPublicationCommitState::Admitted),
-            Some(ScannerPublicationCommitState::InFlight),
-            Some(ScannerPublicationCommitState::AbortedBeforeCommit),
-            Some(ScannerPublicationCommitState::Indeterminate),
-        ] {
-            assert!(!root_ack_write_is_confirmed(&saved, state, Some("revision")));
-        }
-        for etag in [None, Some("")] {
-            assert!(!root_ack_write_is_confirmed(&saved, Some(ScannerPublicationCommitState::Committed), etag));
-        }
-        assert!(root_ack_write_is_confirmed(
-            &saved,
-            Some(ScannerPublicationCommitState::Committed),
-            Some("revision")
-        ));
-    }
-
-    #[test]
-    fn root_publication_confirmation_does_not_carry_state_across_cas_attempts() {
-        let attempts = [
-            (Err(()), Some(ScannerPublicationCommitState::Committed), Some("first")),
-            (Ok(()), Some(ScannerPublicationCommitState::AbortedBeforeCommit), Some("second")),
-            (Ok(()), None, Some("legacy")),
-            (Ok(()), Some(ScannerPublicationCommitState::Committed), Some("confirmed")),
-        ];
-        let confirmations = attempts
-            .iter()
-            .map(|(result, state, etag)| root_ack_write_is_confirmed(result, *state, *etag))
-            .collect::<Vec<_>>();
-        assert_eq!(confirmations, [false, false, false, true]);
-    }
-}
-
-async fn read_root_publication_proof<S: ScannerObjectIO + ScannerConfigObjectDelete>(
-    store: Arc<S>,
-    ctx: &CancellationToken,
-    deadline: tokio::time::Instant,
-    epoch: u64,
-    expected: &crate::scanner_io::ScannerPublicationExpectation,
-    candidate: &DataUsageInfo,
-    written_etag: Option<&str>,
-) -> Option<RootPublicationProof> {
-    let read = async {
-        let _admission = scanner_publication_admission_for_epoch(store.clone(), epoch).await?;
-        let (bytes, revision) = read_config_with_revision(store, DATA_USAGE_OBJ_NAME_PATH.as_str())
-            .await
-            .ok()?;
-        let bytes = bytes?;
-        let DataUsageCacheRevision::Etag(etag) = revision else {
-            return None;
-        };
-        if etag.is_empty() || written_etag.is_some_and(|written| written != etag) {
-            return None;
-        }
-        let persisted: DataUsageInfo = serde_json::from_slice(&bytes).ok()?;
-        if &persisted != candidate {
-            return None;
-        }
-        let root_digest = Sha256::digest(&bytes).into();
-        if ctx.is_cancelled() || tokio::time::Instant::now() >= deadline {
-            return None;
-        }
-        Some(RootPublicationProof {
-            candidate: expected.clone(),
-            root_version: (etag, root_digest),
-        })
-    };
-    tokio::select! {
-        biased;
-        _ = ctx.cancelled() => None,
-        result = tokio::time::timeout_at(deadline, read) => result.ok().flatten(),
-    }
-}
-
 fn remote_lease_expired(deadline: Option<std::time::Instant>) -> bool {
     deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline)
 }
@@ -299,7 +166,6 @@ pub(super) struct ScannerPublicationFence {
     pub(super) scanner_publication_lease_fence: Option<String>,
     pub(super) remote_lease_tokens: Vec<Uuid>,
     pub(super) lease_release_safe: Arc<AtomicBool>,
-    pub(super) ack_expectation: Option<crate::scanner_io::ScannerPublicationExpectation>,
 }
 
 impl ScannerPublicationFence {
@@ -314,7 +180,6 @@ impl ScannerPublicationFence {
             scanner_publication_lease_fence,
             remote_lease_tokens: Vec::new(),
             lease_release_safe: Arc::new(AtomicBool::new(true)),
-            ack_expectation: None,
         }
     }
 
@@ -327,26 +192,21 @@ impl ScannerPublicationFence {
         self.lease_release_safe = lease_release_safe;
         self
     }
-
-    pub(super) fn with_ack_expectation(mut self, expected: Option<crate::scanner_io::ScannerPublicationExpectation>) -> Self {
-        self.ack_expectation = expected;
-        self
-    }
 }
 
 #[derive(Debug)]
-pub(super) enum DataUsagePersistTaskResult<T = DataUsagePersistOutcome> {
-    Completed(T),
+pub(super) enum DataUsagePersistTaskResult {
+    Completed(DataUsagePersistOutcome),
     Cancelled,
     TimedOut,
     JoinFailed(tokio::task::JoinError),
 }
 
-pub(super) async fn wait_for_data_usage_persist_task<T>(
+pub(super) async fn wait_for_data_usage_persist_task(
     ctx: &CancellationToken,
-    task: &mut AbortOnDropHandle<T>,
+    task: &mut AbortOnDropHandle<DataUsagePersistOutcome>,
     timeout: Duration,
-) -> DataUsagePersistTaskResult<T> {
+) -> DataUsagePersistTaskResult {
     tokio::select! {
         biased;
         result = &mut *task => match result {
@@ -460,7 +320,6 @@ where
         route_probe,
     )
     .await
-    .outcome()
 }
 
 pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence<
@@ -474,7 +333,7 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel
     initial_baseline: Option<DataUsagePersistBaseline>,
     publication_fence: ScannerPublicationFence,
     route_probe: F,
-) -> DataUsagePublicationResult
+) -> DataUsagePersistOutcome
 where
     F: Fn() -> Fut + Send + Sync,
     Fut: Future<Output = Option<ScannerCycleDeferReason>> + Send,
@@ -485,15 +344,11 @@ where
         scanner_publication_lease_fence,
         remote_lease_tokens,
         lease_release_safe,
-        ack_expectation,
     } = publication_fence;
-    let ack_deadline = scanner_publication_scope_deadline(data_usage_persist_timeout(), remote_lease_deadline);
     let mut outcome = DataUsagePersistOutcome::NoUpdate;
-    let mut proof = None;
     let mut next_baseline = initial_baseline;
 
     'updates: while let Some(mut data_usage_info) = receiver.recv().await {
-        proof = None;
         let _activity_guard = ScannerActivityGuard::new();
         if ctx.is_cancelled() {
             break;
@@ -668,14 +523,10 @@ where
                 continue;
             }
         };
-        let data_digest: [u8; 32] = Sha256::digest(&data).into();
-        let sha256hex = (!data.is_empty()).then(|| hex_simd::encode_to_string(data_digest, hex_simd::AsciiCase::Lower));
+        let sha256hex = (!data.is_empty()).then(|| hex_simd::encode_to_string(Sha256::digest(&data), hex_simd::AsciiCase::Lower));
         let data = Bytes::from(data);
         let backup_due = !observational && data_usage_backup_due(&data_usage_info);
         let mut cas_retry = 0usize;
-        let mut ack_epoch = None;
-        let mut write_confirmed = false;
-        let mut written_etag = None;
         let save_outcome = loop {
             if ctx.is_cancelled() {
                 break 'updates;
@@ -706,7 +557,6 @@ where
             } else {
                 None
             };
-            ack_epoch = Some(publication_epoch_for_save);
             let (existing_data, revision) = match baseline {
                 Some(baseline) => (baseline.data, baseline.revision),
                 None => match read_config_with_revision(storeapi.clone(), target_path).await {
@@ -795,7 +645,7 @@ where
             }
 
             let done_save = Metrics::time(Metric::SaveUsage);
-            let (save_result, commit_state) = {
+            let save_result = {
                 let publication_scope = storeapi
                     .scanner_data_usage_publication_commit_scope_with_release_flag(
                         publication_epoch_for_save,
@@ -831,33 +681,24 @@ where
                 .await;
                 drop(legacy_publication_admission);
                 if let Some(scope) = publication_scope {
-                    let state = scope.wait_for_completion().await;
-                    let result = match state {
-                        ScannerPublicationCommitState::Committed => save_result,
-                        ScannerPublicationCommitState::AbortedBeforeCommit => save_result,
+                    match scope.wait_for_completion().await {
+                        ScannerPublicationCommitState::Committed | ScannerPublicationCommitState::AbortedBeforeCommit => {
+                            save_result
+                        }
                         ScannerPublicationCommitState::Indeterminate
                         | ScannerPublicationCommitState::Admitted
                         | ScannerPublicationCommitState::InFlight => Err(EcstoreError::other(
                             "scanner publication commit scope did not reach a safe terminal state",
                         )),
-                    };
-                    (result, Some(state))
+                    }
                 } else {
-                    (save_result, None)
+                    save_result
                 }
             };
             done_save();
 
-            let attempt_confirmed = root_ack_write_is_confirmed(
-                &save_result,
-                commit_state,
-                save_result.as_ref().ok().and_then(|info| info.etag.as_deref()),
-            );
-
             match save_result {
                 Ok(object_info) => {
-                    write_confirmed = attempt_confirmed;
-                    written_etag = object_info.etag.as_ref().filter(|etag| !etag.is_empty()).cloned();
                     if !observational {
                         next_baseline = object_info
                             .etag
@@ -1068,27 +909,9 @@ where
                 break 'updates;
             }
         }
-        if !observational
-            && data_usage_info.usage_snapshot_converged == Some(true)
-            && matches!(outcome, DataUsagePersistOutcome::Saved | DataUsagePersistOutcome::AlreadyDurable)
-            && (outcome == DataUsagePersistOutcome::AlreadyDurable || write_confirmed)
-            && let (Some(expected), Some(epoch)) = (ack_expectation.as_ref(), ack_epoch)
-            && expected.matches_encoded_candidate(&data_digest)
-        {
-            proof = read_root_publication_proof(
-                storeapi.clone(),
-                &ctx,
-                ack_deadline,
-                epoch,
-                expected,
-                &data_usage_info,
-                written_etag.as_deref(),
-            )
-            .await;
-        }
     }
 
-    DataUsagePublicationResult { outcome, proof }
+    outcome
 }
 
 async fn cleanup_observed_data_usage_snapshot_for_epoch_and_lease(

crates/scanner/src/scanner_io.rs

@@ -114,11 +114,6 @@ pub(crate) struct ScannerBucketScanScope {
 }
 
 impl ScannerBucketScanScope {
-    #[cfg(test)]
-    pub(crate) fn selected_buckets_for_tests(&self) -> Option<&HashSet<String>> {
-        self.selected_buckets.as_deref()
-    }
-
     fn is_default(&self) -> bool {
         self.selected_buckets.is_none() && self.baseline_scan_plan_digest.is_none()
     }
@@ -223,8 +218,8 @@ fn complete_scanner_cache_snapshot_plan_digest(
 
 fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<'_>) -> Option<DataUsageScanPlanDigest> {
     let authoritative = serde_json::from_slice::<DataUsageInfo>(proof.authoritative_data?).ok()?;
-    if let Some(validated_digest) = complete_scanner_cache_snapshot_plan_digest(&authoritative, proof, true) {
-        return Some(validated_digest);
+    if let Some(plan_digest) = complete_scanner_cache_snapshot_plan_digest(&authoritative, proof, true) {
+        return Some(plan_digest);
     }
 
     // A complete but superseded observation may reuse its per-set cache only
@@ -901,7 +896,6 @@ pub(crate) struct ScannerCycleResult {
     failed_dirty_usage: bool,
     pending_maintenance_work: bool,
     required_cycle_floor: Option<u64>,
-    publication_expectation: Option<ScannerPublicationExpectation>,
 }
 
 impl ScannerCycleResult {
@@ -917,12 +911,10 @@ impl ScannerCycleResult {
             failed_dirty_usage: false,
             pending_maintenance_work: false,
             required_cycle_floor: None,
-            publication_expectation: None,
         }
     }
 
     pub(crate) fn with_publication_epoch(mut self, publication_epoch: Option<u64>) -> Self {
-        self.publication_expectation = None;
         self.publication_epoch = publication_epoch;
         self
     }
@@ -932,7 +924,6 @@ impl ScannerCycleResult {
     }
 
     fn with_activity_digest(mut self, activity_digest: [u8; 32]) -> Self {
-        self.publication_expectation = None;
         self.activity_digest = Some(activity_digest);
         self
     }
@@ -942,7 +933,6 @@ impl ScannerCycleResult {
     }
 
     pub(crate) fn with_observational_snapshot_published(mut self, published: bool) -> Self {
-        self.publication_expectation = None;
         self.observational_snapshot_published = published;
         self
     }
@@ -952,19 +942,16 @@ impl ScannerCycleResult {
     }
 
     fn with_failed_dirty_usage(mut self, failed_dirty_usage: bool) -> Self {
-        self.publication_expectation = None;
         self.failed_dirty_usage = failed_dirty_usage;
         self
     }
 
     fn with_pending_maintenance_work(mut self, pending_maintenance_work: bool) -> Self {
-        self.publication_expectation = None;
         self.pending_maintenance_work = pending_maintenance_work;
         self
     }
 
     fn with_required_cycle_floor(mut self, required_cycle_floor: Option<u64>) -> Self {
-        self.publication_expectation = None;
         self.required_cycle_floor = required_cycle_floor;
         self
     }
@@ -973,13 +960,11 @@ impl ScannerCycleResult {
         mut self,
         acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
     ) -> Self {
-        self.publication_expectation = None;
         self.remote_dirty_usage_acknowledgements = acknowledgements;
         self
     }
 
     pub(crate) fn with_remote_publication_lease_targets(mut self, targets: Vec<(String, String, u64)>) -> Self {
-        self.publication_expectation = None;
         self.remote_publication_lease_targets = targets;
         self
     }
@@ -988,32 +973,7 @@ impl ScannerCycleResult {
         &self.remote_publication_lease_targets
     }
 
-    pub(crate) fn publication_expectation(&self) -> Option<ScannerPublicationExpectation> {
-        self.publication_expectation.clone()
-    }
-
-    fn with_publication_expectation(mut self, expectation: Option<ScannerPublicationExpectation>) -> Self {
-        // Seal only after all coverage and acknowledgement inputs are final.
-        self.publication_expectation = expectation;
-        self
-    }
-
-    pub(crate) fn acknowledge_durable_usage(
-        self,
-        proof: &crate::scanner::RootPublicationProof,
-    ) -> Vec<crate::scanner::ScannerDirtyUsageAcknowledgement> {
-        if self.status != ScannerCycleStatus::Complete
-            || self
-                .publication_expectation
-                .as_ref()
-                .is_none_or(|expected| proof.verified_version_for(expected).is_none())
-        {
-            return Vec::new();
-        }
-        self.clear_verified_usage()
-    }
-
-    fn clear_verified_usage(self) -> Vec<crate::scanner::ScannerDirtyUsageAcknowledgement> {
+    pub(crate) fn acknowledge_durable_usage(self) -> Vec<crate::scanner::ScannerDirtyUsageAcknowledgement> {
         if let Some(snapshot) = self.dirty_usage_clear {
             clear_dirty_usage_buckets(&snapshot);
         }
@@ -1053,7 +1013,6 @@ mod publish_gate_tests;
 #[cfg(test)]
 mod tests;
 
-pub(crate) use cache::ScannerPublicationExpectation;
 use cache::*;
 use dirty_usage::*;
 use guards::*;

crates/scanner/src/scanner_io/cache.rs

@@ -308,86 +308,6 @@ impl<'a> ValidatedScannerSnapshot<'a> {
     }
 }
 
-#[derive(Clone, Debug)]
-pub(crate) struct ScannerPublicationExpectation {
-    candidate: Arc<([u8; 32], DataUsageScanPlanDigest)>,
-}
-
-impl ScannerPublicationExpectation {
-    pub(crate) fn matches_encoded_candidate(&self, digest: &[u8; 32]) -> bool {
-        &self.candidate.0 == digest
-    }
-
-    pub(crate) fn same_candidate(&self, other: &Self) -> bool {
-        Arc::ptr_eq(&self.candidate, &other.candidate) && self.candidate.1 == other.candidate.1
-    }
-}
-
-pub(super) struct ValidatedUsageCandidate {
-    data: DataUsageInfo,
-    #[cfg(test)]
-    last_update: SystemTime,
-    coverage_digest: DataUsageScanPlanDigest,
-}
-
-pub(super) fn empty_namespace_usage_candidate(
-    all_buckets: &[BucketInfo],
-    sources: &HashSet<DataUsageCacheSource>,
-    buckets_by_source: &HashMap<DataUsageCacheSource, Vec<BucketInfo>>,
-    identity: ScannerSnapshotIdentity,
-) -> Option<ValidatedUsageCandidate> {
-    if !all_buckets.is_empty()
-        || sources.is_empty()
-        || sources.len() != buckets_by_source.len()
-        || sources
-            .iter()
-            .any(|source| buckets_by_source.get(source).is_none_or(|buckets| !buckets.is_empty()))
-    {
-        return None;
-    }
-    let last_update = SystemTime::now();
-    Some(ValidatedUsageCandidate {
-        data: DataUsageInfo {
-            last_update: Some(last_update),
-            scanner_cycle: Some(identity.cycle),
-            scanner_epoch: Some(identity.leader_epoch),
-            usage_snapshot_complete: true,
-            ..Default::default()
-        },
-        #[cfg(test)]
-        last_update,
-        coverage_digest: identity.coverage_digest,
-    })
-}
-
-impl ValidatedUsageCandidate {
-    pub(super) fn prepare(mut self, status: ScannerCycleStatus) -> (DataUsageInfo, Option<ScannerPublicationExpectation>) {
-        self.data.usage_snapshot_converged = Some(status == ScannerCycleStatus::Complete);
-        let expectation = if status == ScannerCycleStatus::Complete {
-            struct DigestWriter(Sha256);
-            impl std::io::Write for DigestWriter {
-                fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
-                    self.0.update(bytes);
-                    Ok(bytes.len())
-                }
-                fn flush(&mut self) -> std::io::Result<()> {
-                    Ok(())
-                }
-            }
-            let mut writer = DigestWriter(Sha256::new());
-            serde_json::to_writer(&mut writer, &self.data)
-                .ok()
-                .map(|()| ScannerPublicationExpectation {
-                    candidate: Arc::new((writer.0.finalize().into(), self.coverage_digest)),
-                })
-        } else {
-            None
-        };
-        (self.data, expectation)
-    }
-}
-
-#[cfg(test)]
 pub(super) fn completed_data_usage_info(
     results: &[DataUsageCache],
     scope: &ScannerSnapshotScope<'_>,
@@ -396,18 +316,6 @@ pub(super) fn completed_data_usage_info(
     budget_elapsed: bool,
     cancelled: bool,
 ) -> Option<(DataUsageInfo, SystemTime)> {
-    completed_usage_candidate(results, scope, tier_registry_names, bucket_plan_complete, budget_elapsed, cancelled)
-        .map(|candidate| (candidate.data, candidate.last_update))
-}
-
-pub(super) fn completed_usage_candidate(
-    results: &[DataUsageCache],
-    scope: &ScannerSnapshotScope<'_>,
-    tier_registry_names: &[String],
-    bucket_plan_complete: bool,
-    budget_elapsed: bool,
-    cancelled: bool,
-) -> Option<ValidatedUsageCandidate> {
     if !bucket_plan_complete {
         return None;
     }
@@ -485,12 +393,7 @@ pub(super) fn completed_usage_candidate(
         usage_snapshot_set_states,
         ..Default::default()
     };
-    Some(ValidatedUsageCandidate {
-        data: data_usage_info,
-        #[cfg(test)]
-        last_update: merged_last_update,
-        coverage_digest: scope.identity.coverage_digest,
-    })
+    Some((data_usage_info, merged_last_update))
 }
 
 fn tier_accounting_proof_is_publishable(

crates/scanner/src/scanner_io/io_cycle.rs

@@ -338,21 +338,12 @@ where
             dirty_usage_status,
             activity_status,
         );
-        let Some(candidate) = empty_namespace_usage_candidate(
-            &all_buckets,
-            &expected_sources,
-            &buckets_by_source,
-            ScannerSnapshotIdentity {
-                cycle: want_cycle,
-                leader_epoch,
-                plan_digest: scan_plan_digest,
-                coverage_digest: bucket_coverage_digest,
-                tier_registry_generation: Some(tier_registry_generation),
-            },
-        ) else {
-            return Ok(ScannerCycleResult::new(ScannerCycleStatus::Incomplete, None).with_publication_epoch(publication_epoch));
+        let empty_usage = DataUsageInfo {
+            last_update: Some(SystemTime::now()),
+            scanner_cycle: Some(want_cycle),
+            usage_snapshot_complete: true,
+            ..Default::default()
         };
-        let (empty_usage, publication_expectation) = candidate.prepare(status);
         let observational_snapshot_published = if should_publish_observational_snapshot(status) {
             publish_observational_snapshot(&updates, empty_usage).await?
         } else {
@@ -375,8 +366,7 @@ where
             .with_activity_digest(activity_digest)
             .with_observational_snapshot_published(observational_snapshot_published)
             .with_remote_publication_lease_targets(remote_publication_lease_targets)
-            .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements)
-            .with_publication_expectation(publication_expectation));
+            .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements));
     }
 
     let total_results = expected_sources.len();
@@ -605,7 +595,7 @@ where
     let (activity_status, remote_publication_lease_targets) =
         scanner_cycle_activity_status(store, distributed, &activity_before).await;
     let all_bucket_names = all_buckets.iter().map(|bucket| bucket.name.clone()).collect::<Vec<_>>();
-    let completed_usage = completed_usage_candidate(
+    let completed_usage = completed_data_usage_info(
         &results,
         &ScannerSnapshotScope {
             sources: &expected_sources,
@@ -646,10 +636,7 @@ where
         dirty_usage_status,
         activity_status,
     );
-    let mut publication_expectation = None;
-    let observational_snapshot_published = if let Some(candidate) = completed_usage {
-        let (data_usage_info, expectation) = candidate.prepare(cycle_status);
-        publication_expectation = expectation;
+    let observational_snapshot_published = if let Some((data_usage_info, _)) = completed_usage {
         if should_publish_observational_snapshot(cycle_status) {
             publish_observational_snapshot(&updates, data_usage_info).await?
         } else {
@@ -687,6 +674,5 @@ where
         .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements)
         .with_failed_dirty_usage(!failed_buckets.is_empty())
         .with_pending_maintenance_work(pending_maintenance_work)
-        .with_required_cycle_floor(required_cycle_floor)
-        .with_publication_expectation(publication_expectation))
+        .with_required_cycle_floor(required_cycle_floor))
 }

crates/scanner/src/scanner_io/tests.rs

@@ -454,7 +454,6 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure(
                 .await
                 .expect("initial object should persist");
         }
-        wait_for_namespace_commit_tails(&store).await;
         let ctx = CancellationToken::new();
         let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
         let (updates, receiver) = mpsc::channel(1);
@@ -504,7 +503,6 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure(
             .put_object("cold-bucket", "new", &mut reader, &ScannerObjectOptions::default())
             .await
             .expect("new cold object should persist");
-        wait_for_namespace_commit_tails(&store).await;
         record_dirty_usage_bucket("hot-bucket");
         if scan_mode == HealScanMode::Normal && !requires_full_scan {
             record_dirty_usage_bucket("cold-bucket");
@@ -996,8 +994,8 @@ fn dirty_usage_snapshot_clears_a_stably_absent_bucket_after_durable_save() {
     assert!(dirty_usage_buckets().contains_key("temporarily-omitted"));
     assert_eq!(dirty_usage_snapshot_status(&snapshot), DirtyUsageSnapshotStatus::Current);
 
-    let acknowledgements =
-        ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(snapshot.buckets.as_ref().clone())).clear_verified_usage();
+    let acknowledgements = ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(snapshot.buckets.as_ref().clone()))
+        .acknowledge_durable_usage();
     assert!(acknowledgements.is_empty());
     assert!(!dirty_usage_buckets().contains_key("temporarily-omitted"));
     clear_dirty_usage_buckets_for_tests();
@@ -1103,7 +1101,7 @@ fn dirty_usage_is_acknowledged_only_after_durable_usage_confirmation() {
     assert!(dirty_usage_buckets().contains_key("photos"));
 
     let confirmed = ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(snapshot.buckets.as_ref().clone()));
-    let acknowledgements = confirmed.clear_verified_usage();
+    let acknowledgements = confirmed.acknowledge_durable_usage();
     assert!(acknowledgements.is_empty());
     assert!(!dirty_usage_buckets().contains_key("photos"));
     clear_dirty_usage_buckets_for_tests();