feat(scanner): surface cold segment reuse oracle

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

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

crates/scanner/src/scanner_io.rs

@@ -514,6 +514,25 @@ fn scanner_segment_reuse_activation_preflight_from_proof(
     }
 }
 
+fn scanner_segment_reuse_activation_preflight_for_cycle(
+    dirty_usage_snapshot: &DirtyUsageSnapshot,
+    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,
+        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 || distributed_segment_invalidation_evidence.is_some(),
+    })
+}
+
 fn scanner_segment_reuse_activated() -> bool {
     scanner_segment_reuse_activation_preflight().scanner_segment_reuse_activated
 }
@@ -577,6 +596,7 @@ pub struct ScannerBucketScanPlan {
     bucket_failures: ScannerBucketFailureState,
     pending_maintenance_work: Arc<AtomicBool>,
     cache_cycle_floor: Arc<AtomicU64>,
+    cold_zero_walk_reuse_observed: Arc<AtomicBool>,
 }
 
 #[derive(Clone, Default)]
@@ -709,6 +729,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 +1202,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 +1220,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 +1295,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_cache.rs

@@ -172,6 +172,7 @@ impl ScannerIOCache for SetDisks {
             bucket_failures,
             pending_maintenance_work,
             cache_cycle_floor,
+            cold_zero_walk_reuse_observed,
         } = 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 +220,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);
@@ -1466,6 +1470,9 @@ impl ScannerIOCache for SetDisks {
                 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(

crates/scanner/src/scanner_io/io_cycle.rs

@@ -467,12 +467,15 @@ where
         } else {
             Vec::new()
         };
+        let segment_reuse_activation_preflight =
+            scanner_segment_reuse_activation_preflight_for_cycle(&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 +500,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 +594,7 @@ 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(),
         };
         // Spawn task to run the scanner
         let scanner_fut = tokio::spawn(async move {
@@ -693,6 +698,20 @@ 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,
+        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 +805,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,90 @@ 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, 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, 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, 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,
+        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);
 
@@ -1437,6 +1521,7 @@ 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)),
         },
         tx,
         8,

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