feat(scanner): bind producer evidence to dirty generations (#7538)

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/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"),
@@ -516,17 +519,20 @@ 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,
-        durable_producer_identity: false,
-        restart_gap_absent: 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_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(),

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,23 @@ 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,
+}
+
 /// A point-in-time view of the local dirty bucket generations.
 ///
 /// `complete == false` is an all-or-nothing overflow signal: `buckets` is
@@ -264,8 +280,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 +301,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 +312,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 +323,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 +355,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 +367,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 +386,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 +402,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 +410,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 +707,29 @@ 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,
+    }
+}
+
 #[cfg(test)]
 pub(super) fn dirty_usage_bucket_count() -> usize {
     dirty_usage_buckets().len()

crates/scanner/src/scanner_io/io_cycle.rs

@@ -467,8 +467,13 @@ where
         } else {
             Vec::new()
         };
-        let segment_reuse_activation_preflight =
-            scanner_segment_reuse_activation_preflight_for_cycle(&dirty_usage_snapshot, distributed, None, false);
+        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)
@@ -708,6 +713,7 @@ where
     );
     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,

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");
@@ -166,8 +171,13 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_reports_cycle_inputs_wit
         all_peers_bound_to_generation_window: true,
     };
 
-    let preflight =
-        scanner_segment_reuse_activation_preflight_for_cycle(&dirty_usage_snapshot, true, Some(distributed_evidence), 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);
@@ -186,7 +196,13 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_blocks_unbounded_inputs(
         covers_all_pending: false,
     };
 
-    let preflight = scanner_segment_reuse_activation_preflight_for_cycle(&dirty_usage_snapshot, true, None, 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);
@@ -205,7 +221,13 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_skips_distributed_blocke
         covers_all_pending: true,
     };
 
-    let preflight = scanner_segment_reuse_activation_preflight_for_cycle(&dirty_usage_snapshot, false, None, 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);
@@ -219,6 +241,7 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_skips_distributed_blocke
 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,
@@ -241,6 +264,15 @@ 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,
+    }
+}
+
 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");
@@ -1195,6 +1227,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);