fix(scanner): gate segment reuse activation (#7430)

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

crates/scanner/src/scanner_io.rs

@@ -250,7 +250,8 @@ fn resolve_remote_dirty_usage_scope(
     // Peer snapshots contribute bucket names only; the local prefix scopes
     // would narrow a bucket a peer dirtied elsewhere, so the merged scope
     // stays at bucket granularity (same rule as the local fallthrough).
-    let scope = scoped_scan_scope_from_dirty_buckets(requested_scope, dirty_buckets, None, true, all_buckets, baseline_proof);
+    let scope =
+        scoped_scan_scope_from_dirty_buckets(requested_scope, dirty_buckets, None, true, false, all_buckets, baseline_proof);
     if scope.is_default() {
         return default_result(scope);
     }
@@ -362,6 +363,7 @@ fn scoped_scan_scope_from_dirty_buckets(
     dirty_buckets: HashSet<String>,
     dirty_scopes: Option<&DirtyUsageBucketScopes>,
     dirty_snapshot_complete: bool,
+    segment_reuse_activated: bool,
     all_buckets: &[BucketInfo],
     baseline_proof: ScannerCacheBaselineProof<'_>,
 ) -> ScannerBucketScanScope {
@@ -382,24 +384,34 @@ fn scoped_scan_scope_from_dirty_buckets(
         return requested_scope;
     };
 
-    let selected_bucket_prefixes = dirty_scopes
-        .into_iter()
-        .flat_map(|dirty_scopes| {
-            selected_buckets
-                .iter()
-                .filter_map(|bucket| dirty_scopes.get(bucket).map(|scope| (bucket.clone(), scope)))
-        })
-        .filter_map(|(bucket, scope)| match scope {
-            DirtyUsageBucketScope::WholeBucket => None,
-            DirtyUsageBucketScope::TopLevelEntries(entries) => {
-                ScannerBucketPrefixScanScope::from_dirty_top_level_entries(entries.clone()).map(|scope| (bucket, scope))
-            }
-        })
-        .collect();
+    let selected_bucket_prefixes = if segment_reuse_activated {
+        dirty_scopes
+            .into_iter()
+            .flat_map(|dirty_scopes| {
+                selected_buckets
+                    .iter()
+                    .filter_map(|bucket| dirty_scopes.get(bucket).map(|scope| (bucket.clone(), scope)))
+            })
+            .filter_map(|(bucket, scope)| match scope {
+                DirtyUsageBucketScope::WholeBucket => None,
+                DirtyUsageBucketScope::TopLevelEntries(entries) => {
+                    ScannerBucketPrefixScanScope::from_dirty_top_level_entries(entries.clone()).map(|scope| (bucket, scope))
+                }
+            })
+            .collect()
+    } else {
+        HashMap::new()
+    };
 
     ScannerBucketScanScope::from_dirty_buckets(selected_buckets, selected_bucket_prefixes, baseline_scan_plan_digest)
 }
 
+fn scanner_segment_reuse_activated() -> bool {
+    // Production segment reuse stays disabled until a durable mutation-stream
+    // proof satisfies the segment invalidation contract.
+    false
+}
+
 pub(crate) fn is_scanner_metadata_corrupt_error(err: &StorageError) -> bool {
     matches!(err, StorageError::Io(io) if io.to_string().starts_with(SCANNER_METADATA_CORRUPT_ERROR))
 }

crates/scanner/src/scanner_io/io_cycle.rs

@@ -245,6 +245,7 @@ where
         dirty_buckets,
         (!distributed).then_some(resolution.dirty_usage_snapshot.scopes.as_ref()),
         true,
+        scanner_segment_reuse_activated(),
         resolution.all_buckets,
         resolution.baseline_proof,
     ))

crates/scanner/src/scanner_io/tests.rs

@@ -373,9 +373,13 @@ async fn scoped_scan_production_entry_preserves_deep_and_full_maintenance_work()
                 .expect("maintenance object should persist");
             wait_for_namespace_commit_tails(store.as_ref()).await;
             // Only the hot bucket is in the dirty-usage hint. The ordinary
-            // dirty cycle exercises scoped reuse; the following maintenance
-            // cycles mutate cold storage and must still walk it.
-            record_dirty_usage_bucket("hot-bucket");
+            // dirty cycle exercises bucket-scoped reuse; object-level segment
+            // hints remain activation-gated.
+            if index == 1 {
+                record_dirty_usage_object("hot-bucket", &format!("added-{index}"));
+            } else {
+                record_dirty_usage_bucket("hot-bucket");
+            }
         }
         let requested_scope = if explicit_scope {
             ScannerBucketScanScope::from_dirty_buckets(
@@ -422,6 +426,10 @@ async fn scoped_scan_production_entry_preserves_deep_and_full_maintenance_work()
                 Some(&HashSet::from(["hot-bucket".to_string()])),
                 "ordinary dirty work must retain the existing planner"
             );
+            assert!(
+                resolved.prefix_scope_for("hot-bucket").is_none(),
+                "production segment reuse must remain disabled before activation"
+            );
         } else {
             assert!(resolved.is_default(), "cycle {cycle} must visit the full maintenance scope");
         }
@@ -1566,6 +1574,7 @@ fn scoped_scan_selects_only_current_dirty_buckets_after_baseline_validation() {
         HashSet::from(["photos".to_string(), "deleted".to_string()]),
         None,
         true,
+        false,
         &[bucket_info("photos")],
         ScannerCacheBaselineProof {
             authoritative_data: Some(&baseline),
@@ -1620,7 +1629,7 @@ fn scoped_scan_baseline_work_proof_requires_uniform_known_set_identity() {
 }
 
 #[test]
-fn scoped_scan_uses_only_locally_verified_prefix_hints() {
+fn scoped_scan_prefix_hints_require_segment_reuse_activation() {
     let source = DataUsageCacheSource::new(1, 2);
     let expected_sources = HashSet::from([source]);
     let scan_plan_digest = DataUsageScanPlanDigest([6; 32]);
@@ -1638,6 +1647,7 @@ fn scoped_scan_uses_only_locally_verified_prefix_hints() {
         HashSet::from(["photos".to_string(), "videos".to_string()]),
         Some(&dirty_scopes),
         true,
+        false,
         &[bucket_info("photos"), bucket_info("videos")],
         ScannerCacheBaselineProof {
             authoritative_data: Some(&baseline),
@@ -1648,14 +1658,41 @@ fn scoped_scan_uses_only_locally_verified_prefix_hints() {
             scan_plan_digest,
         },
     );
-    assert!(locally_scoped.prefix_scope_for("photos").is_some());
+    assert_eq!(
+        locally_scoped.selected_buckets.as_deref(),
+        Some(&HashSet::from(["photos".to_string(), "videos".to_string()]))
+    );
+    assert!(
+        locally_scoped.prefix_scope_for("photos").is_none(),
+        "production must not consume segment hints before activation"
+    );
     assert!(locally_scoped.prefix_scope_for("videos").is_none());
 
+    let activated = scoped_scan_scope_from_dirty_buckets(
+        ScannerBucketScanScope::default(),
+        HashSet::from(["photos".to_string(), "videos".to_string()]),
+        Some(&dirty_scopes),
+        true,
+        true,
+        &[bucket_info("photos"), bucket_info("videos")],
+        ScannerCacheBaselineProof {
+            authoritative_data: Some(&baseline),
+            observed_candidate_data: None,
+            expected_sources: &expected_sources,
+            leader_epoch: 11,
+            want_cycle: 8,
+            scan_plan_digest,
+        },
+    );
+    assert!(activated.prefix_scope_for("photos").is_some());
+    assert!(activated.prefix_scope_for("videos").is_none());
+
     let distributed_scope = scoped_scan_scope_from_dirty_buckets(
         ScannerBucketScanScope::default(),
         HashSet::from(["photos".to_string(), "videos".to_string()]),
         None,
         true,
+        true,
         &[bucket_info("photos"), bucket_info("videos")],
         ScannerCacheBaselineProof {
             authoritative_data: Some(&baseline),
@@ -1700,6 +1737,7 @@ fn remote_dirty_usage_invalidates_local_prefix_hints_until_distributed_proof_exi
         HashSet::from(["photos".to_string()]),
         Some(&dirty_scopes),
         true,
+        true,
         &[bucket_info("photos")],
         ScannerCacheBaselineProof {
             authoritative_data: Some(&baseline),