test(scanner): wait for EC8+4 heal control readiness (#7540)

Record scanner segment producer identities with the dirty usage generation they invalidated, then expose a cycle-local producer evidence snapshot for segment reuse activation preflight.

Production activation remains fail-closed until durable producer identity and restart-gap proof are available.

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/e2e_test/src/distributed/heal_test.rs

@@ -26,6 +26,7 @@ use std::collections::{BTreeMap, HashSet};
 use std::io::{Read, Write};
 use std::path::{Path, PathBuf};
 use std::time::Duration;
+use tokio::time::{Instant, sleep};
 
 const EC84_NODE_COUNT: usize = 3;
 const EC84_DRIVES_PER_NODE: usize = 4;
@@ -33,6 +34,8 @@ const EC84_DATA_BLOCKS: usize = 8;
 const EC84_PARITY_BLOCKS: usize = 4;
 const EC84_TARGET_DRIVE_RESTART_CASE: &str = "ec84-target-drive-restart";
 const EC84_TARGET_DRIVE_RESTART_ORACLE: &str = "ec84-target-drive-restart.json";
+const EC84_HEAL_CONTROL_READY_TIMEOUT: Duration = Duration::from_secs(45);
+const EC84_HEAL_CONTROL_RETRY_DELAY: Duration = Duration::from_millis(250);
 
 #[derive(Clone)]
 struct ExpectedShard {
@@ -211,6 +214,29 @@ fn assert_replaced_drive_empty(drive: &Path, bucket: &str, keys: &[String]) -> T
     Ok(())
 }
 
+fn is_cluster_heal_coordination_unavailable(error: &(dyn std::error::Error + Send + Sync)) -> bool {
+    let message = error.to_string();
+    message.contains("500 Internal Server Error") && message.contains("cluster heal coordination unavailable")
+}
+
+async fn start_ec84_root_heal_when_control_ready(
+    heal_url: &str,
+    heal_body: &str,
+    access_key: &str,
+    secret_key: &str,
+) -> TestResult {
+    let deadline = Instant::now() + EC84_HEAL_CONTROL_READY_TIMEOUT;
+    loop {
+        match signed_admin_post(heal_url, Some(heal_body), access_key, secret_key).await {
+            Ok(_) => return Ok(()),
+            Err(error) if is_cluster_heal_coordination_unavailable(error.as_ref()) && Instant::now() < deadline => {
+                sleep(EC84_HEAL_CONTROL_RETRY_DELAY).await;
+            }
+            Err(error) => return Err(error),
+        }
+    }
+}
+
 async fn put_large_inventory(client: &Client, bucket: &str) -> TestResult<Vec<ExpectedShard>> {
     let mut expected = Vec::new();
     for index in 0..4 {
@@ -304,7 +330,7 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
     let heal_body =
         r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#;
     let heal_url = format!("{}/rustfs/admin/v3/heal/{bucket}?forceStart=true", dist.cluster.nodes[0].url);
-    signed_admin_post(&heal_url, Some(heal_body), &dist.cluster.access_key, &dist.cluster.secret_key).await?;
+    start_ec84_root_heal_when_control_ready(&heal_url, heal_body, &dist.cluster.access_key, &dist.cluster.secret_key).await?;
 
     wait_until(
         Duration::from_secs(120),
@@ -365,3 +391,23 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn cluster_heal_coordination_retry_is_exact() {
+        let retryable: Box<dyn std::error::Error + Send + Sync> =
+            "admin POST failed: 500 Internal Server Error cluster heal coordination unavailable".into();
+        assert!(is_cluster_heal_coordination_unavailable(retryable.as_ref()));
+
+        let other_internal: Box<dyn std::error::Error + Send + Sync> =
+            "admin POST failed: 500 Internal Server Error unrelated".into();
+        assert!(!is_cluster_heal_coordination_unavailable(other_internal.as_ref()));
+
+        let wrong_status: Box<dyn std::error::Error + Send + Sync> =
+            "admin POST failed: 503 Service Unavailable cluster heal coordination unavailable".into();
+        assert!(!is_cluster_heal_coordination_unavailable(wrong_status.as_ref()));
+    }
+}

crates/e2e_test/src/on_demand_migration/interaction_test.rs

@@ -1137,7 +1137,12 @@ async fn test_odm_admin_config_is_redacted_and_status_counts_match_the_source()
         let miss = env.raw_get(bucket, miss_key).await?;
         assert_eq!(miss.status, 404, "{}", String::from_utf8_lossy(&miss.body));
     }
-    assert!(env.wait_local_listed(bucket, hit_key, SETTLE).await?);
+    let (listed, _, _) = tokio::try_join!(
+        env.wait_local_listed(bucket, hit_key, SETTLE),
+        env.wait_for_status_counter(bucket, "/counters/pulled_objects_total/inline", 1, SETTLE),
+        env.wait_for_status_counter(bucket, "/counters/pulled_bytes_total", body.len() as u64, SETTLE),
+    )?;
+    assert!(listed);
 
     let status = env.status_json(bucket).await?;
     assert_eq!(status.pointer("/configured").and_then(Value::as_bool), Some(true), "{status}");

crates/heal/src/heal/mrf_queue.rs

@@ -1137,6 +1137,8 @@ fn tick_action(dirty: bool, depth: usize, journal_on_disk: bool, retain_replay_j
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::heal::manager::HealConfig;
+    use crate::heal::storage::{ECStoreHealStorage, HealStorageAPI};
     use rustfs_common::mrf_channel::{MrfIntent, MrfKind, MrfVerifiedRepairDisposition, MrfVerifiedRepairEvent};
     use serial_test::serial;
     use std::sync::Arc as StdArc;
@@ -1353,6 +1355,147 @@ mod tests {
         assert_eq!(read_journal(MRF_JOURNAL_PATH).await, None);
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[serial]
+    async fn committed_replay_anchor_waits_for_verified_proof_before_idle_cleanup() {
+        let env = rustfs_test_utils::TestECStoreEnv::builder()
+            .prefix("rustfs_mrf_replay_proof_cleanup")
+            .build()
+            .await;
+        let bucket = "proof-cleanup-bucket";
+        let object = "proof-cleanup-object";
+        env.make_bucket(bucket, false).await;
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(ECStoreHealStorage::new(env.ecstore.clone()));
+        let manager = Arc::new(HealManager::new(
+            storage.clone(),
+            Some(HealConfig {
+                queue_size: 2,
+                heal_interval: Duration::from_secs(3600),
+                enable_auto_heal: false,
+                ..Default::default()
+            }),
+        ));
+        let disks = journal_disks().await;
+        assert!(!disks.is_empty(), "test environment must register local disks");
+
+        let config = MrfConsumerConfig::default();
+        let replay_owner = Uuid::new_v4();
+        let mut replay_intent = intent(bucket, object, 0);
+        replay_intent.kind = MrfKind::PartialWrite;
+        replay_intent.version_id = None;
+        let replay_payload = encoded_payload(&replay_intent);
+        snapshot::publish_committed_snapshot(&disks, replay_owner, 11, &replay_payload, config.journal_max_bytes)
+            .await
+            .expect("publish committed replay checkpoint");
+
+        let mut queue = MrfQueue::new(config.queue_capacity, config.journal_max_bytes);
+        let mut backoff_until = None;
+        let replay = replay_into(&manager, &mut queue, &mut backoff_until).await;
+        assert_eq!(replay.replayed, 1, "the committed replay checkpoint must decode one record");
+        assert_eq!(queue.depth(), 0, "the replayed record must be admitted before cleanup is considered");
+        assert!(backoff_until.is_none(), "the accepted replay must not arm admission backoff");
+        assert_eq!(
+            manager.operations_snapshot().await.queued_by_source.mrf,
+            1,
+            "the replayed record must be visible as an MRF manager request"
+        );
+        assert!(
+            replay.journal_on_disk,
+            "a durable repair anchor must retain the committed checkpoint before proof"
+        );
+        assert!(
+            !replay.retain_journal_for_replay,
+            "retention is due to pending proof, not an incomplete replay"
+        );
+        assert_eq!(
+            replay.durable_replay_anchors.len(),
+            1,
+            "the real bucket incarnation must create a proof anchor"
+        );
+        assert_eq!(
+            replay.cleanup,
+            Some(ReplayCleanup::Committed {
+                owner: replay_owner,
+                sequence: 11,
+            }),
+            "cleanup must remember the committed checkpoint generation read at startup"
+        );
+
+        let anchor = replay.durable_replay_anchors[0].clone();
+        let mut runtime = MrfRuntime {
+            queue,
+            config,
+            checkpoint_owner: Uuid::new_v4(),
+            next_checkpoint_sequence: replay.next_checkpoint_sequence,
+            new_since_flush: 0,
+            dirty: false,
+            journal_on_disk: replay.journal_on_disk,
+            retain_replay_journal: replay.retain_journal_for_replay,
+            durable_replay_anchors: replay.durable_replay_anchors,
+            replay_cleanup: replay.cleanup,
+            runtime_checkpoint: None,
+            backoff_until,
+        };
+        assert!(runtime.retained_replay_journal(), "proof-bearing replay anchors must block idle cleanup");
+        assert!(
+            snapshot::inspect_local_committed_snapshot(runtime.config.journal_max_bytes)
+                .await
+                .expect("inspect retained committed checkpoint")
+                .is_some(),
+            "the committed replay checkpoint must still be present before proof"
+        );
+
+        rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent {
+            kind: anchor.kind,
+            bucket: anchor.bucket.clone(),
+            object: anchor.object.clone(),
+            version_id: anchor.version_id,
+            scope: anchor.scope,
+            lease: Some(anchor.lease),
+            bucket_incarnation_id: anchor.bucket_incarnation_id,
+            disposition: MrfVerifiedRepairDisposition::Repaired,
+        });
+        runtime.discharge_durable_replay_anchors();
+        assert!(
+            !runtime.retained_replay_journal(),
+            "the exact verified proof must release the durable replay anchor"
+        );
+        assert!(
+            runtime.delete_idle_recovery_anchors().await,
+            "idle cleanup must delete the proof-discharged committed replay checkpoint"
+        );
+        runtime.journal_on_disk = false;
+        assert!(
+            snapshot::inspect_local_committed_snapshot(runtime.config.journal_max_bytes)
+                .await
+                .expect("inspect committed checkpoints after proof cleanup")
+                .is_none(),
+            "the committed replay checkpoint must be gone after proof-driven cleanup"
+        );
+
+        let restart_manager = Arc::new(HealManager::new(
+            storage,
+            Some(HealConfig {
+                queue_size: 2,
+                heal_interval: Duration::from_secs(3600),
+                enable_auto_heal: false,
+                ..Default::default()
+            }),
+        ));
+        assert_eq!(
+            replay_journal_once(&restart_manager).await,
+            0,
+            "proof-cleaned recovery anchors must not resurrect on the next restart"
+        );
+        assert_eq!(
+            restart_manager.operations_snapshot().await.queued_by_source.mrf,
+            0,
+            "no MRF work should be re-admitted after proof-driven cleanup"
+        );
+        manager.stop().await.expect("stop proof cleanup manager");
+        restart_manager.stop().await.expect("stop restart-check manager");
+    }
+
     #[test]
     fn durable_replay_acquires_a_fresh_lease_before_manager_admission() {
         let unique = uuid::Uuid::new_v4();

crates/scanner/src/data_usage_define.rs

@@ -564,6 +564,18 @@ impl DataUsageCacheSource {
 #[serde(transparent)]
 pub struct DataUsageScanPlanDigest(pub [u8; 32]);
 
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
+pub struct DataUsageSegmentInvalidationProof {
+    #[serde(default)]
+    pub process_epoch: String,
+    #[serde(default)]
+    pub generation_start: u64,
+    #[serde(default)]
+    pub generation_end: u64,
+    #[serde(default)]
+    pub producer_identity_coverage_complete: bool,
+}
+
 #[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(rename_all = "snake_case")]
 pub enum PendingScannerHealKind {
@@ -657,6 +669,11 @@ pub struct DataUsageCacheInfo {
     /// structural plan remains reusable across ordinary bucket writes.
     #[serde(default)]
     pub scan_execution_digest: Option<DataUsageScanPlanDigest>,
+    /// Process-epoch and generation window that produced a complete set cache
+    /// with all known segment invalidation producers wired. This proof is
+    /// additive compatibility metadata; absence keeps segment reuse disabled.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub segment_invalidation_proof: Option<DataUsageSegmentInvalidationProof>,
     /// Durable bucket incarnations captured for a complete set aggregate.
     /// Missing or nil entries are legacy/unproven and cannot authorize
     /// skipping an unselected bucket in a later scoped set scan.
@@ -686,6 +703,7 @@ impl Serialize for DataUsageCacheInfo {
             + usize::from(self.lkg_leader_epoch.is_some())
             + usize::from(self.lkg_scan_plan_digest.is_some())
             + usize::from(self.scan_execution_digest.is_some())
+            + usize::from(self.segment_invalidation_proof.is_some())
             + usize::from(!self.scan_bucket_incarnations.is_empty());
         let mut state = serializer.serialize_map(Some(field_count))?;
         state.serialize_entry("name", &self.name)?;
@@ -746,6 +764,9 @@ impl Serialize for DataUsageCacheInfo {
         if let Some(scan_execution_digest) = self.scan_execution_digest {
             state.serialize_entry("scan_execution_digest", &scan_execution_digest)?;
         }
+        if let Some(proof) = &self.segment_invalidation_proof {
+            state.serialize_entry("segment_invalidation_proof", proof)?;
+        }
         if !self.scan_bucket_incarnations.is_empty() {
             state.serialize_entry("scan_bucket_incarnations", &self.scan_bucket_incarnations)?;
         }

crates/scanner/src/data_usage_define/tests.rs

@@ -1095,6 +1095,7 @@ fn test_data_usage_cache_info_deserialize_defaults_scan_resume_after() {
     assert!(!decoded.snapshot_complete);
     assert!(decoded.scan_plan_digest.is_none());
     assert!(decoded.scan_execution_digest.is_none());
+    assert!(decoded.segment_invalidation_proof.is_none());
     assert_eq!(decoded.cache_key_format, 0);
 }
 
@@ -1183,6 +1184,12 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() {
             snapshot_complete: true,
             scan_plan_digest: Some(TEST_PLAN_DIGEST),
             scan_execution_digest: Some(DataUsageScanPlanDigest([42; 32])),
+            segment_invalidation_proof: Some(DataUsageSegmentInvalidationProof {
+                process_epoch: "scanner-process".to_string(),
+                generation_start: 7,
+                generation_end: 9,
+                producer_identity_coverage_complete: true,
+            }),
             cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
             ..Default::default()
         },
@@ -1212,6 +1219,15 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() {
     assert!(current.info.snapshot_complete);
     assert_eq!(current.info.scan_plan_digest, Some(TEST_PLAN_DIGEST));
     assert_eq!(current.info.scan_execution_digest, Some(DataUsageScanPlanDigest([42; 32])));
+    assert_eq!(
+        current.info.segment_invalidation_proof,
+        Some(DataUsageSegmentInvalidationProof {
+            process_epoch: "scanner-process".to_string(),
+            generation_start: 7,
+            generation_end: 9,
+            producer_identity_coverage_complete: true,
+        })
+    );
     assert_eq!(current.info.cache_key_format, DATA_USAGE_CACHE_KEY_FORMAT);
     assert_eq!(current.find("bucket").map(|entry| entry.objects), Some(3));
 

crates/scanner/src/scanner_io.rs

@@ -39,7 +39,7 @@ use s3s::dto::{
     BucketLifecycleConfiguration, ObjectLockConfiguration, ObjectLockEnabled, ReplicationConfiguration, VersioningConfiguration,
 };
 use sha2::{Digest as _, Sha256};
-use std::collections::{BTreeSet, HashMap, HashSet};
+use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 use std::future::Future;
 use std::path::Path;
 use std::pin::Pin;
@@ -204,6 +204,7 @@ pub(crate) struct DistributedSegmentInvalidationEvidence {
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub(crate) struct ScannerSegmentReuseActivationProof {
     pub(crate) production_activation: bool,
+    pub(crate) producer_identity_coverage_complete: bool,
     pub(crate) durable_producer_identity: bool,
     pub(crate) restart_gap_absent: bool,
     pub(crate) generation_window_bound: bool,
@@ -495,6 +496,7 @@ fn scanner_segment_reuse_activation_preflight_from_proof(
     ScannerSegmentReuseActivationPreflight {
         production_activation: proof.production_activation,
         scanner_segment_reuse_activated: proof.production_activation
+            && proof.producer_identity_coverage_complete
             && proof.durable_producer_identity
             && proof.restart_gap_absent
             && proof.generation_window_bound
@@ -504,7 +506,8 @@ fn scanner_segment_reuse_activation_preflight_from_proof(
         proof_inputs: &SCANNER_SEGMENT_ACTIVATION_PROOF_INPUTS,
         fail_closed_checks: &SCANNER_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS,
         fail_closed_blockers: [
-            (!proof.durable_producer_identity).then_some("missing_producer_identity"),
+            (!proof.producer_identity_coverage_complete || !proof.durable_producer_identity)
+                .then_some("missing_producer_identity"),
             (!proof.restart_gap_absent).then_some("restart_gap"),
             (!proof.generation_window_bound).then_some("generation_gap"),
             (!proof.overflow_absent).then_some("overflow"),
@@ -514,6 +517,28 @@ fn scanner_segment_reuse_activation_preflight_from_proof(
     }
 }
 
+fn scanner_segment_reuse_activation_preflight_for_cycle(
+    dirty_usage_snapshot: &DirtyUsageSnapshot,
+    dirty_usage_producer_evidence: DirtyUsageProducerEvidence,
+    distributed: bool,
+    distributed_segment_invalidation_evidence: Option<DistributedSegmentInvalidationEvidence>,
+    cold_zero_walk_oracle: bool,
+) -> ScannerSegmentReuseActivationPreflight {
+    scanner_segment_reuse_activation_preflight_from_proof(ScannerSegmentReuseActivationProof {
+        production_activation: false,
+        producer_identity_coverage_complete: dirty_usage_producer_evidence.producer_identity_coverage_complete,
+        durable_producer_identity: dirty_usage_producer_evidence.durable_producer_identity,
+        restart_gap_absent: dirty_usage_producer_evidence.restart_gap_absent,
+        generation_window_bound: dirty_usage_snapshot.covers_all_pending
+            && dirty_usage_snapshot.generation != 0
+            && dirty_usage_snapshot.generation != u64::MAX
+            && dirty_usage_producer_evidence.generation_window_bound,
+        overflow_absent: dirty_usage_snapshot.covers_all_pending,
+        cold_zero_walk_oracle,
+        distributed_peer_invalidation: !distributed || distributed_segment_invalidation_evidence.is_some(),
+    })
+}
+
 fn scanner_segment_reuse_activated() -> bool {
     scanner_segment_reuse_activation_preflight().scanner_segment_reuse_activated
 }
@@ -577,6 +602,8 @@ pub struct ScannerBucketScanPlan {
     bucket_failures: ScannerBucketFailureState,
     pending_maintenance_work: Arc<AtomicBool>,
     cache_cycle_floor: Arc<AtomicU64>,
+    cold_zero_walk_reuse_observed: Arc<AtomicBool>,
+    segment_invalidation_proof: Option<crate::DataUsageSegmentInvalidationProof>,
 }
 
 #[derive(Clone, Default)]
@@ -709,6 +736,25 @@ fn scanner_bucket_scan_status(has_failed: bool, has_partial: bool, has_namespace
     }
 }
 
+fn scanner_cycle_cold_zero_walk_oracle(
+    scan_scope: &ScannerBucketScanScope,
+    all_buckets: &[BucketInfo],
+    completed_all_sets: bool,
+    scan_scope_matches: bool,
+    bucket_scan_status: ScannerBucketScanStatus,
+    cold_zero_walk_reuse_observed: bool,
+) -> bool {
+    let Some(selected_buckets) = scan_scope.selected_buckets.as_deref() else {
+        return false;
+    };
+    cold_zero_walk_reuse_observed
+        && !selected_buckets.is_empty()
+        && completed_all_sets
+        && scan_scope_matches
+        && bucket_scan_status == ScannerBucketScanStatus::Complete
+        && all_buckets.iter().any(|bucket| !selected_buckets.contains(&bucket.name))
+}
+
 fn classify_nsscanner_cycle(
     completed_all_sets: bool,
     budget_elapsed: bool,
@@ -1163,6 +1209,7 @@ pub(crate) struct ScannerCycleResult {
     dirty_usage_clear: Option<DirtyUsageBuckets>,
     remote_dirty_usage_acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
     distributed_segment_invalidation_evidence: Option<DistributedSegmentInvalidationEvidence>,
+    segment_reuse_activation_preflight: ScannerSegmentReuseActivationPreflight,
     remote_publication_lease_targets: Vec<(String, String, u64)>,
     failed_dirty_usage: bool,
     pending_maintenance_work: bool,
@@ -1180,6 +1227,7 @@ impl ScannerCycleResult {
             dirty_usage_clear,
             remote_dirty_usage_acknowledgements: Vec::new(),
             distributed_segment_invalidation_evidence: None,
+            segment_reuse_activation_preflight: scanner_segment_reuse_activation_preflight(),
             remote_publication_lease_targets: Vec::new(),
             failed_dirty_usage: false,
             pending_maintenance_work: false,
@@ -1254,6 +1302,12 @@ impl ScannerCycleResult {
         self
     }
 
+    fn with_segment_reuse_activation_preflight(mut self, preflight: ScannerSegmentReuseActivationPreflight) -> Self {
+        self.publication_expectation = None;
+        self.segment_reuse_activation_preflight = preflight;
+        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;

crates/scanner/src/scanner_io/dirty_usage.rs

@@ -16,17 +16,16 @@ use super::*;
 
 pub(super) static DIRTY_USAGE_BUCKET_GENERATION: AtomicU64 = AtomicU64::new(0);
 pub(super) static DIRTY_USAGE_BUCKETS: LazyLock<StdMutex<DirtyUsageBuckets>> = LazyLock::new(|| StdMutex::new(HashMap::new()));
-// Lock order when both dirty maps are needed is `DIRTY_USAGE_BUCKETS` followed
-// by `DIRTY_USAGE_BUCKET_SCOPES`. Both are held only for synchronous map
-// updates, so no scanner task can observe a bucket generation without its
-// matching scope.
+// Lock order when dirty usage state is updated is `DIRTY_USAGE_BUCKETS`,
+// `DIRTY_USAGE_BUCKET_SCOPES`, then `DIRTY_USAGE_PRODUCER_IDENTITIES`. All
+// guards are held only for synchronous map updates, so no scanner task can
+// observe a bucket generation without its matching scope and producer evidence.
 pub(super) static DIRTY_USAGE_BUCKET_SCOPES: LazyLock<StdMutex<DirtyUsageBucketScopes>> =
     LazyLock::new(|| StdMutex::new(HashMap::new()));
 // Non-authoritative process-local producer coverage. Any future segment reuse
 // activation must bind this to the exact generation window and durable proof.
-pub(super) static DIRTY_USAGE_PRODUCER_IDENTITIES: LazyLock<
-    StdMutex<BTreeSet<crate::segment_invalidation::SegmentInvalidationProducerIdentity>>,
-> = LazyLock::new(|| StdMutex::new(BTreeSet::new()));
+pub(super) static DIRTY_USAGE_PRODUCER_IDENTITIES: LazyLock<StdMutex<DirtyUsageProducerIdentities>> =
+    LazyLock::new(|| StdMutex::new(BTreeMap::new()));
 pub(super) static DIRTY_USAGE_BUCKET_NOTIFY: LazyLock<Notify> = LazyLock::new(Notify::new);
 pub(super) static SCANNER_ACTIVITY_EPOCH: LazyLock<String> = LazyLock::new(|| format!("{:032x}", rand::random::<u128>()));
 pub(super) static SCANNER_MAINTENANCE_GENERATION: AtomicU64 = AtomicU64::new(0);
@@ -59,6 +58,38 @@ pub(super) type DirtyUsageBucketScopes = HashMap<String, DirtyUsageBucketScope>;
 
 const MAX_DIRTY_USAGE_TOP_LEVEL_ENTRIES_PER_BUCKET: usize = 128;
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(super) struct DirtyUsageProducerIdentityState {
+    first_generation: u64,
+    last_generation: u64,
+}
+
+pub(super) type DirtyUsageProducerIdentities =
+    BTreeMap<crate::segment_invalidation::SegmentInvalidationProducerIdentity, DirtyUsageProducerIdentityState>;
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub(super) struct DirtyUsageProducerEvidence {
+    pub(super) producer_identity_coverage_complete: bool,
+    pub(super) durable_producer_identity: bool,
+    pub(super) restart_gap_absent: bool,
+    pub(super) generation_window_bound: bool,
+    pub(super) generation_start: u64,
+    pub(super) generation_end: u64,
+}
+
+impl DirtyUsageProducerEvidence {
+    pub(super) fn segment_invalidation_proof(self) -> Option<crate::DataUsageSegmentInvalidationProof> {
+        (self.generation_window_bound && self.producer_identity_coverage_complete).then(|| {
+            crate::DataUsageSegmentInvalidationProof {
+                process_epoch: scanner_activity_epoch().to_string(),
+                generation_start: self.generation_start,
+                generation_end: self.generation_end,
+                producer_identity_coverage_complete: true,
+            }
+        })
+    }
+}
+
 /// A point-in-time view of the local dirty bucket generations.
 ///
 /// `complete == false` is an all-or-nothing overflow signal: `buckets` is
@@ -264,8 +295,7 @@ fn dirty_usage_bucket_scopes() -> MutexGuard<'static, DirtyUsageBucketScopes> {
         .unwrap_or_else(|poisoned| poisoned.into_inner())
 }
 
-fn dirty_usage_producer_identities()
--> MutexGuard<'static, BTreeSet<crate::segment_invalidation::SegmentInvalidationProducerIdentity>> {
+fn dirty_usage_producer_identities() -> MutexGuard<'static, DirtyUsageProducerIdentities> {
     DIRTY_USAGE_PRODUCER_IDENTITIES
         .lock()
         .unwrap_or_else(|poisoned| poisoned.into_inner())
@@ -286,7 +316,7 @@ pub fn record_dirty_usage_bucket(bucket: &str) {
         return;
     }
 
-    record_dirty_usage_bucket_inner(bucket);
+    record_dirty_usage_bucket_inner(bucket, std::iter::empty());
 }
 
 pub fn record_dirty_usage_bucket_from_producer(
@@ -297,8 +327,7 @@ pub fn record_dirty_usage_bucket_from_producer(
         return;
     }
 
-    record_segment_invalidation_producer_identity(producer);
-    record_dirty_usage_bucket_inner(bucket);
+    record_dirty_usage_bucket_inner(bucket, [producer]);
 }
 
 pub fn record_dirty_usage_bucket_from_producers<I>(bucket: &str, producers: I)
@@ -309,17 +338,21 @@ where
         return;
     }
 
-    record_segment_invalidation_producer_identities(producers);
-    record_dirty_usage_bucket_inner(bucket);
+    record_dirty_usage_bucket_inner(bucket, producers);
 }
 
-fn record_dirty_usage_bucket_inner(bucket: &str) {
+fn record_dirty_usage_bucket_inner<I>(bucket: &str, producers: I)
+where
+    I: IntoIterator<Item = crate::segment_invalidation::SegmentInvalidationProducerIdentity>,
+{
     let pending_buckets = {
         let mut dirty_buckets = dirty_usage_buckets();
         let mut dirty_scopes = dirty_usage_bucket_scopes();
+        let mut producer_identities = dirty_usage_producer_identities();
         let generation = advance_generation(&DIRTY_USAGE_BUCKET_GENERATION);
         dirty_buckets.insert(bucket.to_string(), generation);
         dirty_scopes.insert(bucket.to_string(), DirtyUsageBucketScope::WholeBucket);
+        record_segment_invalidation_producer_identities_for_generation(&mut producer_identities, generation, producers);
         dirty_buckets.len()
     };
     global_metrics().record_scanner_dirty_usage_pending(usize_to_u64_saturated(pending_buckets));
@@ -337,7 +370,7 @@ fn record_dirty_usage_bucket_inner(bucket: &str) {
 /// local: after restart or any unverified distributed path the scanner falls
 /// back to its ordinary bucket scan.
 pub fn record_dirty_usage_object(bucket: &str, object: &str) {
-    record_dirty_usage_object_inner(bucket, object);
+    record_dirty_usage_object_inner(bucket, object, std::iter::empty());
 }
 
 pub fn record_dirty_usage_object_from_producer(
@@ -349,13 +382,16 @@ pub fn record_dirty_usage_object_from_producer(
         return;
     }
 
-    record_segment_invalidation_producer_identity(producer);
-    record_dirty_usage_object_inner(bucket, object);
+    record_dirty_usage_object_inner(bucket, object, [producer]);
 }
 
-fn record_dirty_usage_object_inner(bucket: &str, object: &str) {
+fn record_dirty_usage_object_inner<I>(bucket: &str, object: &str, producers: I)
+where
+    I: IntoIterator<Item = crate::segment_invalidation::SegmentInvalidationProducerIdentity>,
+{
+    let producers = producers.into_iter().collect::<Vec<_>>();
     let Some(top_level_entry) = dirty_usage_top_level_entry(object) else {
-        record_dirty_usage_bucket(bucket);
+        record_dirty_usage_bucket_inner(bucket, producers);
         return;
     };
     if bucket.is_empty() {
@@ -365,6 +401,7 @@ fn record_dirty_usage_object_inner(bucket: &str, object: &str) {
     let pending_buckets = {
         let mut dirty_buckets = dirty_usage_buckets();
         let mut dirty_scopes = dirty_usage_bucket_scopes();
+        let mut producer_identities = dirty_usage_producer_identities();
         let generation = advance_generation(&DIRTY_USAGE_BUCKET_GENERATION);
         dirty_buckets.insert(bucket.to_string(), generation);
         let scope = dirty_scopes
@@ -380,6 +417,7 @@ fn record_dirty_usage_object_inner(bucket: &str, object: &str) {
         if overflowed {
             *scope = DirtyUsageBucketScope::WholeBucket;
         }
+        record_segment_invalidation_producer_identities_for_generation(&mut producer_identities, generation, producers);
         dirty_buckets.len()
     };
     global_metrics().record_scanner_dirty_usage_pending(usize_to_u64_saturated(pending_buckets));
@@ -387,25 +425,29 @@ fn record_dirty_usage_object_inner(bucket: &str, object: &str) {
     DIRTY_USAGE_BUCKET_NOTIFY.notify_one();
 }
 
-fn record_segment_invalidation_producer_identity(producer: crate::segment_invalidation::SegmentInvalidationProducerIdentity) {
-    record_segment_invalidation_producer_identities([producer]);
-}
-
-fn record_segment_invalidation_producer_identities<I>(producers: I)
-where
+fn record_segment_invalidation_producer_identities_for_generation<I>(
+    identities: &mut DirtyUsageProducerIdentities,
+    generation: u64,
+    producers: I,
+) where
     I: IntoIterator<Item = crate::segment_invalidation::SegmentInvalidationProducerIdentity>,
 {
-    let mut identities = dirty_usage_producer_identities();
     for producer in producers {
         if producer.producer().is_some() {
-            identities.insert(producer);
+            identities
+                .entry(producer)
+                .and_modify(|state| state.last_generation = state.last_generation.max(generation))
+                .or_insert(DirtyUsageProducerIdentityState {
+                    first_generation: generation,
+                    last_generation: generation,
+                });
         }
     }
 }
 
 #[cfg(test)]
 fn dirty_usage_producer_identities_for_tests() -> BTreeSet<crate::segment_invalidation::SegmentInvalidationProducerIdentity> {
-    dirty_usage_producer_identities().clone()
+    dirty_usage_producer_identities().keys().copied().collect()
 }
 
 fn dirty_usage_top_level_entry(object: &str) -> Option<String> {
@@ -680,6 +722,31 @@ pub(super) fn dirty_usage_snapshot_status(snapshot: &DirtyUsageSnapshot) -> Dirt
     }
 }
 
+pub(super) fn dirty_usage_producer_evidence(snapshot: &DirtyUsageSnapshot) -> DirtyUsageProducerEvidence {
+    let generation_window_bound = dirty_usage_snapshot_status(snapshot) == DirtyUsageSnapshotStatus::Current
+        && snapshot.generation != 0
+        && snapshot.generation != u64::MAX;
+    let identities = dirty_usage_producer_identities()
+        .iter()
+        .filter(|(_, state)| state.first_generation <= snapshot.generation)
+        .map(|(identity, _)| *identity)
+        .collect::<BTreeSet<_>>();
+    let producer_identity_coverage_complete =
+        generation_window_bound && crate::segment_invalidation::complete_segment_invalidation_producers(identities).is_ok();
+
+    DirtyUsageProducerEvidence {
+        producer_identity_coverage_complete,
+        // The current producer journal is still process-local. Keep the
+        // durable/restart gates closed until the mutation evidence is persisted
+        // and replayable across scanner restarts.
+        durable_producer_identity: false,
+        restart_gap_absent: false,
+        generation_window_bound,
+        generation_start: snapshot.generation,
+        generation_end: snapshot.generation,
+    }
+}
+
 #[cfg(test)]
 pub(super) fn dirty_usage_bucket_count() -> usize {
     dirty_usage_buckets().len()

crates/scanner/src/scanner_io/io_cache.rs

@@ -172,6 +172,8 @@ impl ScannerIOCache for SetDisks {
             bucket_failures,
             pending_maintenance_work,
             cache_cycle_floor,
+            cold_zero_walk_reuse_observed,
+            segment_invalidation_proof,
         } = scan_plan;
         let scan_plan_digest = scanner_bucket_work_digest(scan_plan_digest, scan_mode, requires_full_scan);
         let bucket_work_digest = scanner_bucket_work_digest(bucket_coverage_digest, scan_mode, requires_full_scan);
@@ -219,6 +221,9 @@ impl ScannerIOCache for SetDisks {
             },
             current_bucket_incarnations.as_ref(),
         );
+        let cold_zero_walk_reuse_candidate = scoped_scan.as_ref().is_some_and(|prepared| {
+            old_cache.info.next_cycle < want_cycle && !prepared.buckets.is_empty() && prepared.buckets.len() < all_buckets.len()
+        });
         let mut scoped_cache = scoped_scan.map(|mut prepared| {
             buckets = prepared.buckets;
             prepared.cache.info.scan_coverage_digest = Some(bucket_coverage_digest);
@@ -239,6 +244,7 @@ impl ScannerIOCache for SetDisks {
                             scan_plan_digest: Some(scan_plan_digest),
                             scan_coverage_digest: Some(bucket_coverage_digest),
                             cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
+                            segment_invalidation_proof: segment_invalidation_proof.clone(),
                             scan_bucket_incarnations: current_bucket_incarnations.clone().unwrap_or_default(),
                             ..Default::default()
                         },
@@ -254,6 +260,7 @@ impl ScannerIOCache for SetDisks {
             cache.info.last_update = Some(now);
             cache.info.snapshot_complete = true;
             cache.info.scan_execution_digest = Some(execution_digest);
+            cache.info.segment_invalidation_proof = segment_invalidation_proof.clone();
             cache.info.lkg_snapshot_complete = false;
             cache.info.lkg_next_cycle = None;
             cache.info.lkg_last_update = None;
@@ -535,6 +542,7 @@ impl ScannerIOCache for SetDisks {
                     lkg_last_update: old_cache.info.lkg_last_update,
                     lkg_leader_epoch: old_cache.info.lkg_leader_epoch,
                     lkg_scan_plan_digest: old_cache.info.lkg_scan_plan_digest,
+                    segment_invalidation_proof: None,
                     scan_bucket_incarnations: current_bucket_incarnations.clone().unwrap_or_default(),
                     ..Default::default()
                 },
@@ -1461,11 +1469,15 @@ impl ScannerIOCache for SetDisks {
                 cache.info.last_update.get_or_insert_with(SystemTime::now);
                 cache.info.snapshot_complete = true;
                 cache.info.scan_execution_digest = Some(execution_digest);
+                cache.info.segment_invalidation_proof = segment_invalidation_proof.clone();
                 cache.info.lkg_snapshot_complete = false;
                 cache.info.lkg_next_cycle = None;
                 cache.info.lkg_last_update = None;
                 cache.info.lkg_leader_epoch = None;
                 cache.info.lkg_scan_plan_digest = None;
+                if cold_zero_walk_reuse_candidate {
+                    cold_zero_walk_reuse_observed.store(true, Ordering::Release);
+                }
                 cache.clone()
             };
             let _ = persist_and_publish_cache_snapshot(
@@ -1486,6 +1498,7 @@ impl ScannerIOCache for SetDisks {
             incomplete_scope.info.tier_registry_generation = Some(tier_registry_generation);
             incomplete_scope.info.source = Some(source);
             incomplete_scope.info.snapshot_complete = false;
+            incomplete_scope.info.segment_invalidation_proof = None;
             incomplete_scope.info.scan_plan_digest = Some(scan_plan_digest);
             incomplete_scope.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT;
             if let Err(e) = updates.send(incomplete_scope).await {

crates/scanner/src/scanner_io/io_cycle.rs

@@ -411,6 +411,7 @@ where
     let remote_dirty_usage_acknowledgements = scope_resolution.remote_dirty_usage_acknowledgements;
     let distributed_segment_invalidation_evidence = scope_resolution.distributed_segment_invalidation_evidence;
     let scan_scope = scope_resolution.scope;
+    let segment_invalidation_proof = dirty_usage_producer_evidence(&dirty_usage_snapshot).segment_invalidation_proof();
     #[cfg(test)]
     if let Some(observer) = resolved_scope_observer {
         let _ = observer.send(scan_scope.clone());
@@ -467,12 +468,20 @@ where
         } else {
             Vec::new()
         };
+        let segment_reuse_activation_preflight = scanner_segment_reuse_activation_preflight_for_cycle(
+            &dirty_usage_snapshot,
+            dirty_usage_producer_evidence(&dirty_usage_snapshot),
+            distributed,
+            None,
+            false,
+        );
         return Ok(ScannerCycleResult::new(status, dirty_usage_clear)
             .with_publication_epoch(publication_epoch)
             .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_segment_reuse_activation_preflight(segment_reuse_activation_preflight)
             .with_publication_expectation(publication_expectation));
     }
 
@@ -497,6 +506,7 @@ where
     );
     let bucket_failures = ScannerBucketFailureState::default();
     let pending_maintenance_work = Arc::new(AtomicBool::new(false));
+    let cold_zero_walk_reuse_observed = Arc::new(AtomicBool::new(false));
     record_set_scan_concurrency_limit(set_scan_limit);
     debug!(
         target: "rustfs::scanner::io",
@@ -590,6 +600,8 @@ where
             bucket_failures: bucket_failures.clone(),
             pending_maintenance_work: pending_maintenance_work.clone(),
             cache_cycle_floor: cache_cycle_floor.clone(),
+            cold_zero_walk_reuse_observed: cold_zero_walk_reuse_observed.clone(),
+            segment_invalidation_proof: segment_invalidation_proof.clone(),
         };
         // Spawn task to run the scanner
         let scanner_fut = tokio::spawn(async move {
@@ -693,6 +705,21 @@ where
         scan_scope_matches && !partial_buckets.is_empty(),
         scan_scope_matches && !namespace_not_found_buckets.is_empty(),
     );
+    let cold_zero_walk_oracle = scanner_cycle_cold_zero_walk_oracle(
+        &scan_scope,
+        &all_buckets,
+        completed_all_sets,
+        scan_scope_matches,
+        bucket_scan_status,
+        cold_zero_walk_reuse_observed.load(Ordering::Acquire),
+    );
+    let segment_reuse_activation_preflight = scanner_segment_reuse_activation_preflight_for_cycle(
+        &dirty_usage_snapshot,
+        dirty_usage_producer_evidence(&dirty_usage_snapshot),
+        distributed,
+        distributed_segment_invalidation_evidence,
+        cold_zero_walk_oracle,
+    );
     let pending_maintenance_work = pending_maintenance_work_for_cycle(&pending_maintenance_work, &results);
     let observed_cycle_floor = cache_cycle_floor.load(Ordering::Acquire);
     let required_cycle_floor = (observed_cycle_floor > want_cycle).then_some(observed_cycle_floor);
@@ -786,6 +813,7 @@ where
         .with_remote_publication_lease_targets(remote_publication_lease_targets)
         .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements)
         .with_distributed_segment_invalidation_evidence(distributed_segment_invalidation_evidence)
+        .with_segment_reuse_activation_preflight(segment_reuse_activation_preflight)
         .with_failed_dirty_usage(!failed_buckets.is_empty())
         .with_pending_maintenance_work(pending_maintenance_work)
         .with_required_cycle_floor(required_cycle_floor)

crates/scanner/src/scanner_io/tests.rs

@@ -104,6 +104,7 @@ fn scanner_segment_reuse_activation_preflight_reports_release_gate_inputs() {
 fn scanner_segment_reuse_activation_requires_every_preflight_proof() {
     let complete_proof = ScannerSegmentReuseActivationProof {
         production_activation: true,
+        producer_identity_coverage_complete: true,
         durable_producer_identity: true,
         restart_gap_absent: true,
         generation_window_bound: true,
@@ -125,9 +126,13 @@ fn scanner_segment_reuse_activation_requires_every_preflight_proof() {
     assert_eq!(preflight.fail_closed_blockers().collect::<Vec<_>>(), Vec::<&str>::new());
 
     let mut missing_identity = complete_proof;
-    missing_identity.durable_producer_identity = false;
+    missing_identity.producer_identity_coverage_complete = false;
     assert_segment_reuse_activation_blocked_by(missing_identity, "missing_producer_identity");
 
+    let mut non_durable_identity = complete_proof;
+    non_durable_identity.durable_producer_identity = false;
+    assert_segment_reuse_activation_blocked_by(non_durable_identity, "missing_producer_identity");
+
     let mut restart_gap = complete_proof;
     restart_gap.restart_gap_absent = false;
     assert_segment_reuse_activation_blocked_by(restart_gap, "restart_gap");
@@ -149,6 +154,108 @@ fn scanner_segment_reuse_activation_requires_every_preflight_proof() {
     assert_segment_reuse_activation_blocked_by(missing_distributed_invalidation, "distributed_without_peer_invalidation");
 }
 
+#[test]
+fn scanner_segment_reuse_activation_preflight_for_cycle_reports_cycle_inputs_without_activation() {
+    let dirty_usage_snapshot = DirtyUsageSnapshot {
+        buckets: Arc::new(DirtyUsageBuckets::from([("photos".to_string(), 7)])),
+        scopes: Arc::new(DirtyUsageBucketScopes::default()),
+        generation: 7,
+        covers_all_pending: true,
+    };
+    let distributed_evidence = DistributedSegmentInvalidationEvidence {
+        invalidation_domain: crate::segment_invalidation::SegmentInvalidationDomain::DistributedEc,
+        distributed_ec_invalidation: true,
+        peer_count: 2,
+        dirty_peer_count: 1,
+        same_window_remote_proof: true,
+        all_peers_bound_to_generation_window: true,
+    };
+
+    let preflight = scanner_segment_reuse_activation_preflight_for_cycle(
+        &dirty_usage_snapshot,
+        complete_process_local_producer_evidence(),
+        true,
+        Some(distributed_evidence),
+        true,
+    );
+
+    assert!(!preflight.production_activation);
+    assert!(!preflight.scanner_segment_reuse_activated);
+    assert_eq!(
+        preflight.fail_closed_blockers().collect::<Vec<_>>(),
+        vec!["missing_producer_identity", "restart_gap"]
+    );
+}
+
+#[test]
+fn scanner_segment_reuse_activation_preflight_for_cycle_blocks_unbounded_inputs() {
+    let dirty_usage_snapshot = DirtyUsageSnapshot {
+        buckets: Arc::new(DirtyUsageBuckets::default()),
+        scopes: Arc::new(DirtyUsageBucketScopes::default()),
+        generation: u64::MAX,
+        covers_all_pending: false,
+    };
+
+    let preflight = scanner_segment_reuse_activation_preflight_for_cycle(
+        &dirty_usage_snapshot,
+        DirtyUsageProducerEvidence::default(),
+        true,
+        None,
+        false,
+    );
+
+    assert!(!preflight.production_activation);
+    assert!(!preflight.scanner_segment_reuse_activated);
+    assert_eq!(
+        preflight.fail_closed_blockers().collect::<Vec<_>>(),
+        SCANNER_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS
+    );
+}
+
+#[test]
+fn scanner_segment_reuse_activation_preflight_for_cycle_skips_distributed_blocker_for_local_scan() {
+    let dirty_usage_snapshot = DirtyUsageSnapshot {
+        buckets: Arc::new(DirtyUsageBuckets::from([("photos".to_string(), 7)])),
+        scopes: Arc::new(DirtyUsageBucketScopes::default()),
+        generation: 7,
+        covers_all_pending: true,
+    };
+
+    let preflight = scanner_segment_reuse_activation_preflight_for_cycle(
+        &dirty_usage_snapshot,
+        complete_process_local_producer_evidence(),
+        false,
+        None,
+        true,
+    );
+
+    assert!(!preflight.production_activation);
+    assert!(!preflight.scanner_segment_reuse_activated);
+    assert_eq!(
+        preflight.fail_closed_blockers().collect::<Vec<_>>(),
+        vec!["missing_producer_identity", "restart_gap"]
+    );
+}
+
+#[test]
+fn scanner_cycle_result_returns_segment_reuse_activation_preflight() {
+    let proof = ScannerSegmentReuseActivationProof {
+        production_activation: true,
+        producer_identity_coverage_complete: true,
+        durable_producer_identity: true,
+        restart_gap_absent: true,
+        generation_window_bound: true,
+        overflow_absent: true,
+        cold_zero_walk_oracle: true,
+        distributed_peer_invalidation: true,
+    };
+    let preflight = scanner_segment_reuse_activation_preflight_from_proof(proof);
+
+    let result = ScannerCycleResult::new(ScannerCycleStatus::Complete, None).with_segment_reuse_activation_preflight(preflight);
+
+    assert_eq!(result.segment_reuse_activation_preflight, preflight);
+}
+
 fn assert_segment_reuse_activation_blocked_by(proof: ScannerSegmentReuseActivationProof, blocker: &'static str) {
     let preflight = scanner_segment_reuse_activation_preflight_from_proof(proof);
 
@@ -157,6 +264,17 @@ fn assert_segment_reuse_activation_blocked_by(proof: ScannerSegmentReuseActivati
     assert_eq!(preflight.fail_closed_blockers().collect::<Vec<_>>(), vec![blocker]);
 }
 
+fn complete_process_local_producer_evidence() -> DirtyUsageProducerEvidence {
+    DirtyUsageProducerEvidence {
+        producer_identity_coverage_complete: true,
+        durable_producer_identity: false,
+        restart_gap_absent: false,
+        generation_window_bound: true,
+        generation_start: 7,
+        generation_end: 7,
+    }
+}
+
 async fn setup_two_pool_scanner_store() -> (tempfile::TempDir, Arc<ECStore>) {
     init_ecstore_config_for_scanner_tests();
     let temp_dir = tempfile::tempdir().expect("multi-pool scanner test directory should be created");
@@ -1111,6 +1229,29 @@ fn dirty_usage_snapshot_detects_uncovered_generation() {
     clear_dirty_usage_buckets_for_tests();
 }
 
+#[test]
+#[serial]
+fn dirty_usage_producer_evidence_tracks_process_local_coverage_without_durable_restart_authority() {
+    use crate::segment_invalidation::SegmentInvalidationProducerIdentity;
+
+    clear_dirty_usage_buckets_for_tests();
+    record_dirty_usage_bucket_from_producers("photos", SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION);
+    let snapshot = snapshot_dirty_usage_buckets(&[bucket_info("photos")], dirty_usage_generation());
+
+    let evidence = dirty_usage_producer_evidence(&snapshot);
+
+    assert!(evidence.generation_window_bound);
+    assert!(evidence.producer_identity_coverage_complete);
+    assert!(!evidence.durable_producer_identity);
+    assert!(!evidence.restart_gap_absent);
+
+    record_dirty_usage_bucket_from_producer("videos", SegmentInvalidationProducerIdentity::PutObject);
+    let stale_evidence = dirty_usage_producer_evidence(&snapshot);
+    assert!(!stale_evidence.generation_window_bound);
+    assert!(!stale_evidence.producer_identity_coverage_complete);
+    clear_dirty_usage_buckets_for_tests();
+}
+
 #[test]
 fn generation_saturates_instead_of_wrapping() {
     let generation = AtomicU64::new(u64::MAX - 1);
@@ -1418,6 +1559,12 @@ async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers
 
     let ctx = CancellationToken::new();
     let empty_execution = DataUsageScanPlanDigest([5; 32]);
+    let segment_invalidation_proof = crate::DataUsageSegmentInvalidationProof {
+        process_epoch: scanner_activity_epoch().to_string(),
+        generation_start: 8,
+        generation_end: 8,
+        producer_identity_coverage_complete: true,
+    };
     set.nsscanner_cache(
         ctx.clone(),
         ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default()),
@@ -1437,6 +1584,8 @@ async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers
             bucket_failures: ScannerBucketFailureState::default(),
             pending_maintenance_work: Arc::new(AtomicBool::new(false)),
             cache_cycle_floor: Arc::new(AtomicU64::new(8)),
+            cold_zero_walk_reuse_observed: Arc::new(AtomicBool::new(false)),
+            segment_invalidation_proof: Some(segment_invalidation_proof.clone()),
         },
         tx,
         8,
@@ -1446,6 +1595,7 @@ async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers
     .expect("empty set scope should replace its prior nonempty cache");
     let empty = rx.try_recv().expect("empty set snapshot should be published");
     assert_eq!(empty.info.scan_execution_digest, Some(empty_execution));
+    assert_eq!(empty.info.segment_invalidation_proof, Some(segment_invalidation_proof));
     assert!(empty.info.snapshot_complete);
     let root = empty.checked_flatten(DATA_USAGE_ROOT).expect("complete empty root");
     assert_eq!((root.size, root.objects), (0, 0));

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

@@ -138,6 +138,7 @@ async fn run_entry(store: &Arc<ECStore>, cycle: u64, selected: Option<&str>, exp
     .expect("entry cycle should finish within the fixture deadline")
     .expect("entry cycle should succeed");
     assert_eq!(result.status, ScannerCycleStatus::Complete);
+    let activation_preflight = result.segment_reuse_activation_preflight;
     let scope = observed.await.expect("production resolver should report its decision");
     assert_eq!(
         scope.selected_buckets.as_deref(),
@@ -174,6 +175,20 @@ async fn run_entry(store: &Arc<ECStore>, cycle: u64, selected: Option<&str>, exp
         actual, expected_walks,
         "each listed source/bucket must have exactly the expected real walks"
     );
+    assert!(!activation_preflight.production_activation);
+    assert!(!activation_preflight.scanner_segment_reuse_activated);
+    let activation_blockers = activation_preflight.fail_closed_blockers().collect::<Vec<_>>();
+    if selected.is_some() && expect_walks {
+        assert!(
+            !activation_blockers.contains(&"missing_cold_zero_walk_oracle"),
+            "a complete scoped reuse cycle must carry the cold zero-walk oracle: cycle={cycle} selected={selected:?} blockers={activation_blockers:?}"
+        );
+    } else {
+        assert!(
+            activation_blockers.contains(&"missing_cold_zero_walk_oracle"),
+            "unscoped or same-cycle cache reuse must not claim the cold zero-walk oracle: cycle={cycle} selected={selected:?} expect_walks={expect_walks} blockers={activation_blockers:?}"
+        );
+    }
     assert_eq!(
         read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
             .await

scripts/check_test_wiring.py

@@ -1517,11 +1517,63 @@ def is_json_artifact_format(value: str) -> bool:
     return normalized == "json" or normalized.endswith("+json")
 
 
+def release_bundle_json_artifact_mirrored_fields(gate: str, field: str) -> tuple[str, ...]:
+    fields: list[str] = []
+    if gate in ("G03", "G09", "R-L"):
+        fields.extend(("versions", "mixed_version_role"))
+    if gate in ("G04", "G07", "R-E", "R-L"):
+        fields.append("crash_points")
+    if gate == "G03":
+        fields.append("scoped_ack_cases")
+        if field == "durable_root_publication_proof":
+            fields.extend(("root_cas_observed", "root_readback_observed"))
+        if field == "scoped_ack_request_identity":
+            fields.append("whole_cycle_fallback_observed")
+    if gate == "G04" and field == "root_floor_intent_crash_evidence":
+        fields.extend(("durable_intent_cases", "persist_failure_blocks_acceptance"))
+    if gate == "G07":
+        fields.append({
+            "mrf_responsibility_oracle": "mrf_responsibility_cases",
+            "commit_boundary_crash_matrix": "commit_crash_cases",
+        }[field])
+    if gate == "G08":
+        fields.append({
+            "mrf_capacity_evidence": "capacity_cases",
+            "disk_full_matrix": "disk_full_cases",
+            "replica_loss_matrix": "replica_loss_cases",
+        }[field])
+    if gate == "G09":
+        fields.append("mixed_version_cases")
+        if field == "rollback_payload_evidence":
+            fields.append("rollback_payload_replayed")
+    if (gate, field) in SCANNER_HEAL_RELEASE_MRF_DURABLE_REPLAY_FIELDS:
+        fields.extend(("replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"))
+    if gate == "P4" and field == "retained_responsibility_evidence":
+        fields.extend((
+            "duration_seconds",
+            "retained_responsibility_cases",
+            "retention_window_seconds",
+            "idle_cleanup_observed",
+            "verified_proof_discharge_observed",
+        ))
+    if gate == "P4" and field == "mrf_cleanup_gc_soak_evidence":
+        fields.extend((
+            "duration_seconds",
+            "cleanup_gc_cases",
+            "verified_idle_gc_observed",
+            "pending_responsibilities_after_gc",
+            "stale_journals_after_gc",
+        ))
+    return tuple(dict.fromkeys(fields))
+
+
 def validate_release_bundle_json_artifact_payload(path: Path, source_revision: str, gate: str, field: str,
                                                  run_id: str, window_id: str,
                                                  artifact_kind: str | None = None) -> None:
     payload = read_json(path)
     prefix = f"{gate}.{field}"
+    for marker in ("fixture", "fixture_only", "dry_run", "synthetic"):
+        require(payload.get(marker) is not True, f"{prefix} JSON artifact is {marker}")
     require(payload.get("evidence_type") == "measured", f"{prefix} JSON artifact must be measured")
     require(payload.get("source_revision") == source_revision, f"{prefix} JSON artifact source revision mismatch")
     require(payload.get("run_id") == run_id, f"{prefix} JSON artifact run_id mismatch")
@@ -1530,6 +1582,62 @@ def validate_release_bundle_json_artifact_payload(path: Path, source_revision: s
     require(payload.get("field") == field, f"{prefix} JSON artifact field mismatch")
     if artifact_kind is not None:
         require(payload.get("artifact_kind") == artifact_kind, f"{prefix} JSON artifact kind mismatch")
+    mirror_fields = release_bundle_json_artifact_mirrored_fields(gate, field)
+    for mirror_field in mirror_fields:
+        require(mirror_field in payload, f"{prefix} JSON artifact missing {mirror_field}")
+    if not mirror_fields:
+        return
+    validate_release_bundle_domain_evidence(gate, field, payload)
+    if gate in ("G03", "G09", "R-L"):
+        versions = payload.get("versions")
+        require(isinstance(versions, list) and
+                len(set(versions)) >= 2 and
+                all(isinstance(version, str) and re.fullmatch(r"[0-9a-f]{40}", version) is not None
+                    for version in versions),
+                f"{prefix} JSON artifact requires mixed-version evidence")
+        require(source_revision in versions, f"{prefix} JSON artifact versions omit tested source revision")
+        expected_role = SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES[(gate, field)]
+        require(payload.get("mixed_version_role") == expected_role,
+                f"{prefix} JSON artifact mixed-version role must be {expected_role}")
+    if gate in ("G04", "G07", "R-E", "R-L"):
+        crash_points = payload.get("crash_points")
+        require(isinstance(crash_points, list) and crash_points,
+                f"{prefix} JSON artifact requires crash-boundary evidence")
+    if (gate, field) in SCANNER_HEAL_RELEASE_MRF_DURABLE_REPLAY_FIELDS:
+        evidence_integer(payload.get("replayed_records"), f"{prefix} JSON artifact replayed_records", 1, 2**63 - 1)
+        require(payload.get("responsibility_anchor_retained") is True,
+                f"{prefix} JSON artifact requires retained MRF responsibility anchors")
+        require(payload.get("successor_snapshot_published") is True,
+                f"{prefix} JSON artifact requires successor snapshot publication evidence")
+    if gate == "P4" and field == "mrf_cleanup_gc_soak_evidence":
+        release_bundle_exact_strings(
+            payload.get("cleanup_gc_cases"),
+            SCANNER_HEAL_RELEASE_MRF_CLEANUP_GC_SOAK_CASES,
+            f"{prefix} JSON artifact cleanup_gc_cases",
+        )
+        require(payload.get("verified_idle_gc_observed") is True,
+                f"{prefix} JSON artifact requires verified idle GC evidence")
+        require(payload.get("pending_responsibilities_after_gc") == 0,
+                f"{prefix} JSON artifact requires zero pending responsibilities after GC")
+        require(payload.get("stale_journals_after_gc") == 0,
+                f"{prefix} JSON artifact requires zero stale journals after GC")
+    if gate == "G07":
+        case_field = {
+            "mrf_responsibility_oracle": "mrf_responsibility_cases",
+            "commit_boundary_crash_matrix": "commit_crash_cases",
+        }[field]
+        cases = evidence_string_list(payload.get(case_field), f"{prefix} JSON artifact {case_field}")
+        missing_cases = sorted(set(SCANNER_HEAL_RELEASE_G07_REQUIRED_CASES[field]) - set(cases))
+        require(not missing_cases, f"{prefix} JSON artifact missing cases: {', '.join(missing_cases)}")
+    if gate == "G08":
+        case_field = {
+            "mrf_capacity_evidence": "capacity_cases",
+            "disk_full_matrix": "disk_full_cases",
+            "replica_loss_matrix": "replica_loss_cases",
+        }[field]
+        cases = evidence_string_list(payload.get(case_field), f"{prefix} JSON artifact {case_field}")
+        missing_cases = sorted(set(SCANNER_HEAL_RELEASE_G08_REQUIRED_CASES[field]) - set(cases))
+        require(not missing_cases, f"{prefix} JSON artifact missing cases: {', '.join(missing_cases)}")
 
 
 def release_bundle_bool_true(value: object, name: str) -> None:
@@ -2548,17 +2656,6 @@ class SelfTests(unittest.TestCase):
                     evidence.update({"completed_heal_objects": 1, "duplicate_task_count": 0})
                 if gate == "P3" and field == "recovery_window_measurement":
                     evidence.update({"pressure_recovery_window_seconds": 5, "lock_hold_p95_ms": 0})
-                write_json(artifact, {
-                    "schema": 1,
-                    "evidence_type": "measured",
-                    "source_revision": source_revision,
-                    "run_id": run_id,
-                    "measurement_window_id": window_id,
-                    "gate": gate,
-                    "field": field,
-                    "fixture": True,
-                })
-                evidence["sha256"] = digest(artifact)
                 if gate in ("G03", "G09", "R-L"):
                     evidence["versions"] = ["a" * 40, source_revision]
                     evidence["mixed_version_role"] = SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES[(gate, field)]
@@ -2659,8 +2756,8 @@ class SelfTests(unittest.TestCase):
                     evidence["saved_bytes"] = 2048
                     artifacts = {}
                     for artifact_kind in RELEASE_PROFILE_ARTIFACTS:
-                        artifact = artifact_dir / f"{gate}-{field}-{artifact_kind}.json"
-                        write_json(artifact, {
+                        profile_artifact = artifact_dir / f"{gate}-{field}-{artifact_kind}.json"
+                        write_json(profile_artifact, {
                             "schema": 1,
                             "evidence_type": "measured",
                             "source_revision": source_revision,
@@ -2669,11 +2766,10 @@ class SelfTests(unittest.TestCase):
                             "gate": gate,
                             "field": field,
                             "artifact_kind": artifact_kind,
-                            "fixture": True,
                         })
                         artifacts[artifact_kind] = {
-                            "artifact": artifact.relative_to(bundle_dir).as_posix(),
-                            "sha256": digest(artifact),
+                            "artifact": profile_artifact.relative_to(bundle_dir).as_posix(),
+                            "sha256": digest(profile_artifact),
                             "artifact_format": "json",
                         }
                     evidence["profile_artifacts"] = artifacts
@@ -2703,6 +2799,19 @@ class SelfTests(unittest.TestCase):
                     evidence["fault_modes"] = ["process-restart", "process-crash-restart"]
                     evidence["recovery_p95_ms"] = 500.0
                     evidence["recovery_p99_ms"] = 1000.0
+                artifact_payload = {
+                    "schema": 1,
+                    "evidence_type": "measured",
+                    "source_revision": source_revision,
+                    "run_id": run_id,
+                    "measurement_window_id": window_id,
+                    "gate": gate,
+                    "field": field,
+                }
+                for mirror_field in release_bundle_json_artifact_mirrored_fields(gate, field):
+                    artifact_payload[mirror_field] = evidence[mirror_field]
+                write_json(artifact, artifact_payload)
+                evidence["sha256"] = digest(artifact)
                 fields[field] = evidence
             gates[gate] = {
                 "status": "pass",
@@ -3063,6 +3172,24 @@ class SelfTests(unittest.TestCase):
                 ("P1", "profile_evidence", "rss-samples"),
                 "JSON artifact run_id mismatch",
             ),
+            (
+                "fixture-marker",
+                lambda payload: payload.update({"fixture": True}),
+                ("G08", "disk_full_matrix"),
+                "JSON artifact is fixture",
+            ),
+            (
+                "g08-case-mirror",
+                lambda payload: payload["disk_full_cases"].remove("manifest-write-enospc"),
+                ("G08", "disk_full_matrix"),
+                "JSON artifact missing cases",
+            ),
+            (
+                "p4-gc-mirror",
+                lambda payload: payload.update({"pending_responsibilities_after_gc": 1}),
+                ("P4", "mrf_cleanup_gc_soak_evidence"),
+                "JSON artifact requires zero pending responsibilities",
+            ),
         ):
             with self.subTest(fault=fault), tempfile.TemporaryDirectory() as tmp:
                 root, bundle = self.scanner_heal_release_bundle_fixture(Path(tmp))

scripts/run_scanner_heal_g09_upgrade_evidence.sh

@@ -101,6 +101,16 @@ case_names() {
     esac
 }
 
+validate_test_selection() {
+    case "$TEST_SELECTION" in
+        all|mixed-version|rollback)
+            ;;
+        *)
+            die "unknown test selection: $TEST_SELECTION"
+            ;;
+    esac
+}
+
 artifact_for() {
     case "$1" in
         mixed-version)
@@ -168,7 +178,7 @@ ensure_default_asset_platform() {
 
 verify_sha256() {
     local archive="$1"
-    if command -v sha256sum >/dev/null 2>&1; then
+    if command -v sha256sum >/dev/null 2>&1 && [[ "$(sha256sum --help 2>&1)" == *"--check"* ]]; then
         printf '%s  %s\n' "$SOURCE_SHA256" "$archive" | sha256sum --check --strict
     elif command -v shasum >/dev/null 2>&1; then
         printf '%s  %s\n' "$SOURCE_SHA256" "$archive" | shasum -a 256 --check
@@ -232,7 +242,7 @@ resolve_source_binary() {
     local archive="$SOURCE_DIR/$SOURCE_ASSET"
     curl --fail --location --retry 3 --output "$archive" \
         "https://github.com/$SOURCE_REPOSITORY/releases/download/$SOURCE_VERSION/$SOURCE_ASSET"
-    verify_sha256 "$archive"
+    verify_sha256 "$archive" >&2
     unzip -q "$archive" -d "$SOURCE_DIR"
     chmod +x "$binary"
     test -x "$binary"
@@ -422,6 +432,18 @@ run_self_test() {
     fi
 
     mkdir -p "$tmp/run/mixed-version-upgrade" "$tmp/run/bucket-config-rollback"
+    mkdir -p "$tmp/source"
+    local archive checksum checksum_output
+    archive="$tmp/source/$SOURCE_ASSET"
+    printf 'not a real archive\n' >"$archive"
+    if command -v shasum >/dev/null 2>&1; then
+        checksum="$(shasum -a 256 "$archive" | awk '{print $1}')"
+    else
+        checksum="$(sha256sum "$archive" | awk '{print $1}')"
+    fi
+    checksum_output="$(SOURCE_SHA256="$checksum" verify_sha256 "$archive")"
+    [[ "$checksum_output" == *"OK"* ]]
+
     local current previous
     current="$(git rev-parse HEAD)"
     previous="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
@@ -518,6 +540,7 @@ while [[ $# -gt 0 ]]; do
     esac
 done
 
+validate_test_selection
 CASES=()
 while IFS= read -r case_name; do
     CASES+=("$case_name")

scripts/test_scanner_heal_g09_upgrade_evidence.sh

@@ -25,7 +25,7 @@ fi
 rg -q -- "--sha256 must be a 64-character lowercase hex digest" "$TMP_DIR/bad-sha.err"
 
 VALID_SHA="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
-bash "$RUNNER" \
+env -u CARGO_TARGET_DIR bash "$RUNNER" \
   --dry-run \
   --out-dir "$TMP_DIR/evidence" \
   --source-dir "$TMP_DIR/source" \