feat(scanner): carry segment activation preflight evidence (#7535)

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

crates/scanner/src/scanner_io.rs

@@ -514,6 +514,24 @@ fn scanner_segment_reuse_activation_preflight_from_proof(
     }
 }
 
+fn scanner_segment_reuse_activation_preflight_for_cycle(
+    dirty_usage_snapshot: &DirtyUsageSnapshot,
+    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,
+        generation_window_bound: dirty_usage_snapshot.covers_all_pending
+            && dirty_usage_snapshot.generation != 0
+            && dirty_usage_snapshot.generation != u64::MAX,
+        overflow_absent: dirty_usage_snapshot.covers_all_pending,
+        cold_zero_walk_oracle,
+        distributed_peer_invalidation: distributed_segment_invalidation_evidence.is_some(),
+    })
+}
+
 fn scanner_segment_reuse_activated() -> bool {
     scanner_segment_reuse_activation_preflight().scanner_segment_reuse_activated
 }
@@ -1163,6 +1181,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 +1199,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 +1274,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/io_cycle.rs

@@ -410,6 +410,11 @@ where
     .await;
     let remote_dirty_usage_acknowledgements = scope_resolution.remote_dirty_usage_acknowledgements;
     let distributed_segment_invalidation_evidence = scope_resolution.distributed_segment_invalidation_evidence;
+    let segment_reuse_activation_preflight = scanner_segment_reuse_activation_preflight_for_cycle(
+        &dirty_usage_snapshot,
+        distributed_segment_invalidation_evidence,
+        false,
+    );
     let scan_scope = scope_resolution.scope;
     #[cfg(test)]
     if let Some(observer) = resolved_scope_observer {
@@ -473,6 +478,7 @@ where
             .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));
     }
 
@@ -786,6 +792,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

@@ -149,6 +149,70 @@ 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, 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, 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_cycle_result_returns_segment_reuse_activation_preflight() {
+    let proof = ScannerSegmentReuseActivationProof {
+        production_activation: 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);