feat(scanner): enable segment reuse activation gate (#7546)

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

crates/data-usage/src/data_usage.rs

@@ -603,6 +603,8 @@ pub struct DataUsageSegmentInvalidationProof {
     pub generation_end: u64,
     #[serde(default)]
     pub producer_identity_coverage_complete: bool,
+    #[serde(default)]
+    pub cold_zero_walk_oracle: bool,
 }
 
 #[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
@@ -3144,6 +3146,7 @@ mod tests {
             generation_start: 3,
             generation_end: 5,
             producer_identity_coverage_complete: true,
+            cold_zero_walk_oracle: true,
         };
         let state = DataUsageSnapshotSetState {
             pool_index: 1,

crates/scanner/src/data_usage_define/tests.rs

@@ -1189,6 +1189,7 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() {
                 generation_start: 7,
                 generation_end: 9,
                 producer_identity_coverage_complete: true,
+                cold_zero_walk_oracle: true,
             }),
             cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
             ..Default::default()
@@ -1226,6 +1227,7 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() {
             generation_start: 7,
             generation_end: 9,
             producer_identity_coverage_complete: true,
+            cold_zero_walk_oracle: true,
         })
     );
     assert_eq!(current.info.cache_key_format, DATA_USAGE_CACHE_KEY_FORMAT);

crates/scanner/src/scanner_io.rs

@@ -436,6 +436,60 @@ fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<
     complete_scanner_cache_snapshot_plan_digest(&observed, proof, false)
 }
 
+fn scanner_segment_reuse_baseline_producer_evidence(
+    dirty_usage_snapshot: &DirtyUsageSnapshot,
+    baseline_proof: ScannerCacheBaselineProof<'_>,
+) -> (DirtyUsageProducerEvidence, bool) {
+    let mut evidence = dirty_usage_producer_evidence(dirty_usage_snapshot);
+    if !evidence.generation_window_bound || !evidence.producer_identity_coverage_complete {
+        return (evidence, false);
+    }
+
+    let Some(authoritative_data) = baseline_proof.authoritative_data else {
+        return (evidence, false);
+    };
+    let Ok(authoritative) = serde_json::from_slice::<DataUsageInfo>(authoritative_data) else {
+        return (evidence, false);
+    };
+    if complete_scanner_cache_snapshot_plan_digest(&authoritative, baseline_proof, true).is_none()
+        || !scanner_snapshot_set_states_have_segment_reuse_activation_proof(
+            &authoritative,
+            &evidence,
+            baseline_proof.expected_sources,
+        )
+    {
+        return (evidence, false);
+    }
+
+    evidence.durable_producer_identity = true;
+    evidence.restart_gap_absent = true;
+    (evidence, true)
+}
+
+fn scanner_snapshot_set_states_have_segment_reuse_activation_proof(
+    snapshot: &DataUsageInfo,
+    evidence: &DirtyUsageProducerEvidence,
+    expected_sources: &HashSet<DataUsageCacheSource>,
+) -> bool {
+    let mut covered_sources = HashSet::with_capacity(expected_sources.len());
+    let all_sets_proved = snapshot.usage_snapshot_set_states.iter().all(|state| {
+        let (Ok(pool_index), Ok(set_index)) = (usize::try_from(state.pool_index), usize::try_from(state.set_index)) else {
+            return false;
+        };
+        let source = DataUsageCacheSource::new(pool_index, set_index);
+        state.complete
+            && !state.tombstone
+            && expected_sources.contains(&source)
+            && covered_sources.insert(source)
+            && scanner_segment_invalidation_proof_matches(state.segment_invalidation_proof.as_ref(), evidence)
+            && state
+                .segment_invalidation_proof
+                .as_ref()
+                .is_some_and(|proof| proof.cold_zero_walk_oracle)
+    });
+    all_sets_proved && covered_sources.len() == expected_sources.len()
+}
+
 fn scoped_scan_scope_from_dirty_buckets(
     requested_scope: ScannerBucketScanScope,
     dirty_buckets: HashSet<String>,
@@ -485,9 +539,10 @@ fn scoped_scan_scope_from_dirty_buckets(
 }
 
 fn scanner_segment_reuse_activation_preflight() -> ScannerSegmentReuseActivationPreflight {
-    // Production segment reuse stays disabled until a durable mutation-stream
-    // proof satisfies the segment invalidation contract.
-    scanner_segment_reuse_activation_preflight_from_proof(ScannerSegmentReuseActivationProof::default())
+    scanner_segment_reuse_activation_preflight_from_proof(ScannerSegmentReuseActivationProof {
+        production_activation: true,
+        ..Default::default()
+    })
 }
 
 fn scanner_segment_reuse_activation_preflight_from_proof(
@@ -525,7 +580,7 @@ fn scanner_segment_reuse_activation_preflight_for_cycle(
     cold_zero_walk_oracle: bool,
 ) -> ScannerSegmentReuseActivationPreflight {
     scanner_segment_reuse_activation_preflight_from_proof(ScannerSegmentReuseActivationProof {
-        production_activation: false,
+        production_activation: true,
         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,
@@ -535,7 +590,43 @@ fn scanner_segment_reuse_activation_preflight_for_cycle(
             && 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(),
+        distributed_peer_invalidation: scanner_distributed_segment_invalidation_admitted(
+            distributed,
+            distributed_segment_invalidation_evidence,
+        ),
+    })
+}
+
+fn scanner_segment_reuse_activation_preflight_for_baseline(
+    dirty_usage_snapshot: &DirtyUsageSnapshot,
+    distributed: bool,
+    baseline_proof: ScannerCacheBaselineProof<'_>,
+) -> ScannerSegmentReuseActivationPreflight {
+    let (dirty_usage_producer_evidence, cold_zero_walk_oracle) =
+        scanner_segment_reuse_baseline_producer_evidence(dirty_usage_snapshot, baseline_proof);
+    scanner_segment_reuse_activation_preflight_for_cycle(
+        dirty_usage_snapshot,
+        dirty_usage_producer_evidence,
+        distributed,
+        None,
+        cold_zero_walk_oracle,
+    )
+}
+
+fn scanner_distributed_segment_invalidation_admitted(
+    distributed: bool,
+    evidence: Option<DistributedSegmentInvalidationEvidence>,
+) -> bool {
+    if !distributed {
+        return true;
+    }
+    evidence.is_some_and(|evidence| {
+        evidence.invalidation_domain == crate::segment_invalidation::SegmentInvalidationDomain::DistributedEc
+            && evidence.distributed_ec_invalidation
+            && evidence.same_window_remote_proof
+            && evidence.all_peers_bound_to_generation_window
+            && evidence.dirty_peer_count > 0
+            && evidence.dirty_peer_count <= evidence.peer_count
     })
 }
 
@@ -580,6 +671,17 @@ fn scanner_segment_invalidation_proof_matches(
     })
 }
 
+fn scanner_completed_set_segment_invalidation_proof(
+    proof: &Option<crate::DataUsageSegmentInvalidationProof>,
+    cold_zero_walk_reuse_candidate: bool,
+) -> Option<crate::DataUsageSegmentInvalidationProof> {
+    proof.clone().map(|mut proof| {
+        proof.cold_zero_walk_oracle = cold_zero_walk_reuse_candidate;
+        proof
+    })
+}
+
+#[cfg(test)]
 fn scanner_segment_reuse_activated() -> bool {
     scanner_segment_reuse_activation_preflight().scanner_segment_reuse_activated
 }

crates/scanner/src/scanner_io/dirty_usage.rs

@@ -85,6 +85,7 @@ impl DirtyUsageProducerEvidence {
                 generation_start: self.generation_start,
                 generation_end: self.generation_end,
                 producer_identity_coverage_complete: true,
+                cold_zero_walk_oracle: false,
             }
         })
     }

crates/scanner/src/scanner_io/io_cache.rs

@@ -231,6 +231,8 @@ impl ScannerIOCache for SetDisks {
         });
         if buckets.is_empty() {
             let now = SystemTime::now();
+            let completed_segment_invalidation_proof =
+                scanner_completed_set_segment_invalidation_proof(&segment_invalidation_proof, cold_zero_walk_reuse_candidate);
             let mut cache = match scoped_cache.take() {
                 Some(cache) => cache,
                 None => {
@@ -244,7 +246,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(),
+                            segment_invalidation_proof: completed_segment_invalidation_proof.clone(),
                             scan_bucket_incarnations: current_bucket_incarnations.clone().unwrap_or_default(),
                             ..Default::default()
                         },
@@ -260,7 +262,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.segment_invalidation_proof = completed_segment_invalidation_proof;
             cache.info.lkg_snapshot_complete = false;
             cache.info.lkg_next_cycle = None;
             cache.info.lkg_last_update = None;
@@ -1463,13 +1465,15 @@ impl ScannerIOCache for SetDisks {
 
         let completed_count = completed_bucket_count.load(Ordering::Relaxed);
         if should_publish_completed_snapshot(completed_count, buckets.len(), budget.budget_elapsed(), ctx.is_cancelled()) {
+            let completed_segment_invalidation_proof =
+                scanner_completed_set_segment_invalidation_proof(&segment_invalidation_proof, cold_zero_walk_reuse_candidate);
             let cache_snapshot = {
                 let mut cache = cache_mutex.lock().await;
                 cache.info.next_cycle = want_cycle;
                 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.segment_invalidation_proof = completed_segment_invalidation_proof;
                 cache.info.lkg_snapshot_complete = false;
                 cache.info.lkg_next_cycle = None;
                 cache.info.lkg_last_update = None;

crates/scanner/src/scanner_io/io_cycle.rs

@@ -241,12 +241,17 @@ where
         return remote_resolution;
     }
 
+    let segment_reuse_activation_preflight = scanner_segment_reuse_activation_preflight_for_baseline(
+        resolution.dirty_usage_snapshot,
+        distributed,
+        resolution.baseline_proof,
+    );
     default_result(scoped_scan_scope_from_dirty_buckets(
         resolution.requested_scope,
         dirty_buckets,
         (!distributed).then_some(resolution.dirty_usage_snapshot.scopes.as_ref()),
         true,
-        scanner_segment_reuse_activated(),
+        segment_reuse_activation_preflight.scanner_segment_reuse_activated,
         resolution.all_buckets,
         resolution.baseline_proof,
     ))

crates/scanner/src/scanner_io/publish_gate_tests.rs

@@ -184,6 +184,7 @@ fn completed_data_usage_info_carries_segment_invalidation_proof_to_set_state() {
         generation_start: 5,
         generation_end: 8,
         producer_identity_coverage_complete: true,
+        cold_zero_walk_oracle: true,
     };
     let mut set = completed_root_cache("bucket", 2, 10, source);
     set.info.segment_invalidation_proof = Some(proof.clone());

crates/scanner/src/scanner_io/tests.rs

@@ -89,7 +89,7 @@ fn scanner_activity_preflight_defers_a_temporarily_offline_peer() {
 fn scanner_segment_reuse_activation_preflight_reports_release_gate_inputs() {
     let preflight = scanner_segment_reuse_activation_preflight();
 
-    assert!(!preflight.production_activation);
+    assert!(preflight.production_activation);
     assert!(!preflight.scanner_segment_reuse_activated);
     assert!(!scanner_segment_reuse_activated());
     assert_eq!(preflight.proof_inputs, SCANNER_SEGMENT_ACTIVATION_PROOF_INPUTS);
@@ -179,7 +179,7 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_reports_cycle_inputs_wit
         true,
     );
 
-    assert!(!preflight.production_activation);
+    assert!(preflight.production_activation);
     assert!(!preflight.scanner_segment_reuse_activated);
     assert_eq!(
         preflight.fail_closed_blockers().collect::<Vec<_>>(),
@@ -204,7 +204,7 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_blocks_unbounded_inputs(
         false,
     );
 
-    assert!(!preflight.production_activation);
+    assert!(preflight.production_activation);
     assert!(!preflight.scanner_segment_reuse_activated);
     assert_eq!(
         preflight.fail_closed_blockers().collect::<Vec<_>>(),
@@ -229,7 +229,7 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_skips_distributed_blocke
         true,
     );
 
-    assert!(!preflight.production_activation);
+    assert!(preflight.production_activation);
     assert!(!preflight.scanner_segment_reuse_activated);
     assert_eq!(
         preflight.fail_closed_blockers().collect::<Vec<_>>(),
@@ -280,6 +280,98 @@ fn scanner_durable_segment_invalidation_evidence_requires_matching_complete_set_
     clear_dirty_usage_buckets_for_tests();
 }
 
+#[test]
+#[serial]
+fn scanner_segment_reuse_activation_replays_cold_durable_baseline() {
+    use crate::segment_invalidation::SegmentInvalidationProducerIdentity;
+
+    clear_dirty_usage_buckets_for_tests();
+    for producer in SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION {
+        record_dirty_usage_object_from_producer("photos", "2026/object", producer);
+    }
+    let dirty_usage_snapshot =
+        snapshot_dirty_usage_buckets(&[bucket_info("photos"), bucket_info("archive")], dirty_usage_generation());
+    let mut segment_proof = dirty_usage_producer_evidence(&dirty_usage_snapshot)
+        .segment_invalidation_proof()
+        .expect("complete process-local producer coverage should produce proof metadata");
+    segment_proof.cold_zero_walk_oracle = true;
+    let scan_plan_digest = DataUsageScanPlanDigest([6; 32]);
+    let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0), DataUsageCacheSource::new(0, 1)]);
+    let baseline = DataUsageInfo {
+        last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(10)),
+        scanner_cycle: Some(7),
+        scanner_epoch: Some(11),
+        buckets_count: 2,
+        buckets_usage: HashMap::from([
+            ("photos".to_string(), Default::default()),
+            ("archive".to_string(), Default::default()),
+        ]),
+        usage_snapshot_complete: true,
+        usage_snapshot_converged: Some(true),
+        usage_snapshot_set_states: expected_sources
+            .iter()
+            .map(|source| DataUsageSnapshotSetState {
+                pool_index: u64::try_from(source.pool_index).expect("test pool index should fit"),
+                set_index: u64::try_from(source.set_index).expect("test set index should fit"),
+                scanner_cycle: Some(7),
+                scanner_epoch: Some(11),
+                scan_plan_digest: Some(scan_plan_digest.0),
+                complete: true,
+                tombstone: false,
+                segment_invalidation_proof: Some(segment_proof.clone()),
+            })
+            .collect(),
+        ..Default::default()
+    };
+    let baseline = Bytes::from(serde_json::to_vec(&baseline).expect("baseline should encode"));
+
+    let preflight = scanner_segment_reuse_activation_preflight_for_baseline(
+        &dirty_usage_snapshot,
+        false,
+        ScannerCacheBaselineProof {
+            authoritative_data: Some(&baseline),
+            observed_candidate_data: None,
+            expected_sources: &expected_sources,
+            leader_epoch: 11,
+            want_cycle: 8,
+            scan_plan_digest,
+        },
+    );
+
+    assert!(preflight.production_activation);
+    assert!(preflight.scanner_segment_reuse_activated);
+    assert_eq!(preflight.fail_closed_blockers().collect::<Vec<_>>(), Vec::<&str>::new());
+
+    let mut missing_cold_baseline =
+        serde_json::from_slice::<DataUsageInfo>(&baseline).expect("baseline should decode for negative case");
+    missing_cold_baseline.usage_snapshot_set_states[0]
+        .segment_invalidation_proof
+        .as_mut()
+        .expect("proof should exist")
+        .cold_zero_walk_oracle = false;
+    let missing_cold_baseline = Bytes::from(serde_json::to_vec(&missing_cold_baseline).expect("negative baseline should encode"));
+    let preflight = scanner_segment_reuse_activation_preflight_for_baseline(
+        &dirty_usage_snapshot,
+        false,
+        ScannerCacheBaselineProof {
+            authoritative_data: Some(&missing_cold_baseline),
+            observed_candidate_data: None,
+            expected_sources: &expected_sources,
+            leader_epoch: 11,
+            want_cycle: 8,
+            scan_plan_digest,
+        },
+    );
+
+    assert!(preflight.production_activation);
+    assert!(!preflight.scanner_segment_reuse_activated);
+    assert_eq!(
+        preflight.fail_closed_blockers().collect::<Vec<_>>(),
+        vec!["missing_producer_identity", "restart_gap", "missing_cold_zero_walk_oracle"]
+    );
+    clear_dirty_usage_buckets_for_tests();
+}
+
 #[test]
 fn scanner_cycle_result_returns_segment_reuse_activation_preflight() {
     let proof = ScannerSegmentReuseActivationProof {
@@ -1627,6 +1719,7 @@ async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers
         generation_start: 8,
         generation_end: 8,
         producer_identity_coverage_complete: true,
+        cold_zero_walk_oracle: false,
     };
     set.nsscanner_cache(
         ctx.clone(),

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

@@ -84,7 +84,14 @@ async fn persist_baseline(store: &Arc<ECStore>, baseline: &DataUsageInfo) {
 
 // Every invocation uses the production default scope. Once durable bucket
 // incarnations are present, the expected walker set follows the resolved scope.
-async fn run_entry(store: &Arc<ECStore>, cycle: u64, selected: Option<&str>, expect_walks: bool) -> DataUsageInfo {
+async fn run_entry(
+    store: &Arc<ECStore>,
+    cycle: u64,
+    selected: Option<&str>,
+    expect_walks: bool,
+    expect_activation: bool,
+    expect_prefix_scope: bool,
+) -> DataUsageInfo {
     let drives = drive_identities(store).await;
     let inventory = store
         .list_bucket_for_scanner(&BucketOptions::default())
@@ -144,6 +151,13 @@ async fn run_entry(store: &Arc<ECStore>, cycle: u64, selected: Option<&str>, exp
         scope.selected_buckets.as_deref(),
         selected.map(|name| HashSet::from([name.to_string()])).as_ref()
     );
+    if let Some(selected) = selected {
+        assert_eq!(
+            scope.prefix_scope_for(selected).is_some(),
+            expect_prefix_scope,
+            "resolved prefix scope must match activation replay for cycle {cycle}"
+        );
+    }
     let usage = receiver.recv().await.expect("one candidate should be delivered");
     assert!(receiver.recv().await.is_none(), "there must be exactly one terminal candidate");
     assert!(usage.usage_snapshot_complete);
@@ -175,10 +189,12 @@ 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);
+    assert!(activation_preflight.production_activation);
+    assert_eq!(activation_preflight.scanner_segment_reuse_activated, expect_activation);
     let activation_blockers = activation_preflight.fail_closed_blockers().collect::<Vec<_>>();
-    if selected.is_some() && expect_walks {
+    if expect_activation {
+        assert_eq!(activation_blockers, Vec::<&str>::new());
+    } else 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:?}"
@@ -204,6 +220,12 @@ async fn run_entry(store: &Arc<ECStore>, cycle: u64, selected: Option<&str>, exp
     usage
 }
 
+fn record_segment_dirty_usage(bucket: &str) {
+    for producer in crate::segment_invalidation::SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION {
+        record_dirty_usage_object_from_producer(bucket, "hot-segment/object", producer);
+    }
+}
+
 #[tokio::test]
 #[serial]
 async fn scoped_entry_fallback_distinguishes_planned_scope_from_real_cold_walks() {
@@ -213,14 +235,16 @@ async fn scoped_entry_fallback_distinguishes_planned_scope_from_real_cold_walks(
     let cold = format!("cold-{}", Uuid::new_v4().simple());
     create_bucket(&store, &hot).await;
     create_bucket(&store, &cold).await;
-    record_dirty_usage_bucket(&hot);
-    let baseline = run_entry(&store, 1, None, true).await;
+    record_segment_dirty_usage(&hot);
+    let baseline = run_entry(&store, 1, None, true, false, false).await;
     persist_baseline(&store, &baseline).await;
 
     // Same-cycle Current remains a retry. The later cycle may skip the cold
     // bucket only after the prior complete set cache has durable incarnations.
-    run_entry(&store, 1, Some(&hot), false).await;
-    let usage = run_entry(&store, 2, Some(&hot), true).await;
+    run_entry(&store, 1, Some(&hot), false, false, false).await;
+    let usage = run_entry(&store, 2, Some(&hot), true, true, false).await;
+    persist_baseline(&store, &usage).await;
+    let usage = run_entry(&store, 3, Some(&hot), true, true, true).await;
     assert_eq!(usage.buckets_usage[&hot].objects_count, 1);
     assert_eq!(usage.buckets_usage[&cold].objects_count, 1);
     assert_eq!(usage.objects_total_count, 2);
@@ -238,7 +262,7 @@ async fn scoped_entry_fallback_rejects_invalid_persisted_baseline_at_the_walker(
     create_bucket(&store, &cold).await;
     record_dirty_usage_bucket(&hot);
     // The first real scan is also the missing persisted-baseline case.
-    let baseline = run_entry(&store, 1, None, true).await;
+    let baseline = run_entry(&store, 1, None, true, false, false).await;
     for (index, kind) in [
         "malformed",
         "unconverged",
@@ -271,7 +295,15 @@ async fn scoped_entry_fallback_rejects_invalid_persisted_baseline_at_the_walker(
         crate::save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), bytes)
             .await
             .expect("negative baseline should persist");
-        let usage = run_entry(&store, u64::try_from(index).expect("fixture cycle index should fit") + 2, None, true).await;
+        let usage = run_entry(
+            &store,
+            u64::try_from(index).expect("fixture cycle index should fit") + 2,
+            None,
+            true,
+            false,
+            false,
+        )
+        .await;
         assert_eq!(usage.objects_total_count, 2, "{kind}");
         assert_eq!(usage.buckets_usage[&cold].objects_count, 1, "{kind}");
     }
@@ -286,13 +318,13 @@ async fn scoped_entry_fallback_covers_overflow_and_new_bucket_inventory() {
     let hot = format!("hot-{}", Uuid::new_v4().simple());
     create_bucket(&store, &hot).await;
     record_dirty_usage_bucket(&hot);
-    let baseline = run_entry(&store, 1, None, true).await;
+    let baseline = run_entry(&store, 1, None, true, false, false).await;
     persist_baseline(&store, &baseline).await;
     for index in 0..=crate::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES {
         record_dirty_usage_bucket(&format!("overflow-{index}"));
     }
     assert!(dirty_usage_buckets_for_tests().len() > crate::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES);
-    let usage = run_entry(&store, 2, None, true).await;
+    let usage = run_entry(&store, 2, None, true, false, false).await;
     assert_eq!(usage.objects_total_count, 1);
 
     clear_dirty_usage_buckets_for_tests();
@@ -300,7 +332,7 @@ async fn scoped_entry_fallback_covers_overflow_and_new_bucket_inventory() {
     let new_bucket = format!("new-{}", Uuid::new_v4().simple());
     create_bucket(&store, &new_bucket).await;
     // Even a previously valid baseline cannot cover the changed inventory.
-    let usage = run_entry(&store, 3, None, true).await;
+    let usage = run_entry(&store, 3, None, true, false, false).await;
     assert_eq!(usage.objects_total_count, 2);
     assert_eq!(usage.buckets_usage[&new_bucket].objects_count, 1);
     clear_dirty_usage_buckets_for_tests();