test(scanner): prove segment activation gates (#7388)

Extend the segment observation fixture with durable activation prerequisites and add scanner oracles for cold segment zero-walk and distributed invalidation fallback.

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

crates/scanner/src/scanner_folder/tests.rs

@@ -2451,6 +2451,82 @@ async fn scoped_root_scan_reuses_clean_top_level_entries_and_rescans_dirty_entri
     assert_eq!((bucket.size, bucket.objects), (17, 3));
 }
 
+async fn scan_hot_cold_segment_fixture(scoped: bool) -> (DataUsageEntry, Vec<String>) {
+    let (mut scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
+    write_test_object_metadata(&temp_dir, "bucket", "cold/object").await;
+    write_test_object_metadata(&temp_dir, "bucket", "hot/object").await;
+
+    scanner.old_cache.info.name = "bucket".to_string();
+    scanner.new_cache.info.name = "bucket".to_string();
+    scanner.update_cache.info.name = "bucket".to_string();
+    if scoped {
+        scanner.old_cache.replace("bucket", "", DataUsageEntry::default());
+        scanner.old_cache.replace(
+            "bucket/cold",
+            "bucket",
+            DataUsageEntry {
+                size: 0,
+                objects: 1,
+                ..Default::default()
+            },
+        );
+        scanner.prefix_scan_scope =
+            ScannerBucketPrefixScanScope::from_dirty_top_level_entries(HashSet::from(["hot".to_string()]));
+    }
+
+    let walked = Arc::new(Mutex::new(Vec::<String>::new()));
+    scanner.update_current_path = Arc::new({
+        let walked = walked.clone();
+        move |path: &str| {
+            walked.lock().expect("lock observed scanner paths").push(path.to_string());
+            Box::pin(async {})
+        }
+    });
+
+    let folder = CachedFolder {
+        name: "bucket".to_string(),
+        parent: None,
+        object_heal_prob_div: 1,
+    };
+    let mut root = DataUsageEntry::default();
+    scanner
+        .scan_folder(CancellationToken::new(), folder, &mut root)
+        .await
+        .expect("segment fixture scan should finish");
+    let root = scanner
+        .new_cache
+        .size_recursive("bucket")
+        .expect("segment fixture should produce a bucket cache root");
+    let walked = walked.lock().expect("read observed scanner paths").clone();
+    (root, walked)
+}
+
+fn walked_path_in(paths: &[String], subtree: &str) -> bool {
+    paths
+        .iter()
+        .any(|path| path == subtree || path.strip_prefix(subtree).is_some_and(|rest| rest.starts_with('/')))
+}
+
+#[tokio::test]
+#[serial]
+async fn scoped_root_scan_zero_walks_clean_cold_segment_with_full_oracle_equivalence() {
+    let (full, full_walked) = scan_hot_cold_segment_fixture(false).await;
+    let (scoped, scoped_walked) = scan_hot_cold_segment_fixture(true).await;
+
+    assert_eq!((scoped.size, scoped.objects), (full.size, full.objects));
+    assert_eq!((scoped.size, scoped.objects), (0, 2));
+    assert!(
+        walked_path_in(&full_walked, "bucket/cold"),
+        "the full oracle must prove the cold segment would be walked without scoped reuse"
+    );
+    assert!(walked_path_in(&scoped_walked, "bucket/hot"), "the dirty hot segment must still be walked");
+    assert!(
+        !walked_path_in(&scoped_walked, "bucket/cold"),
+        "a clean cold segment must be copied from the durable baseline without walker callbacks"
+    );
+}
+
 #[tokio::test]
 #[serial]
 async fn scoped_root_scan_preserves_erasure_health_walks() {

crates/scanner/src/scanner_folder/tests/checkpoint_fixture/segment_observation.rs

@@ -39,6 +39,12 @@ impl ProducerKind {
     ];
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum SegmentInvalidationDomain {
+    LocalSingleSet,
+    DistributedEc,
+}
+
 #[derive(Clone, Debug)]
 struct SegmentObservationEnvelope<'a> {
     source: DataUsageCacheSource,
@@ -61,6 +67,10 @@ struct SegmentObservationProof<'a> {
     key_format: u16,
     baseline_scan_plan_digest: DataUsageScanPlanDigest,
     process_epoch: &'a str,
+    durable_producer_identity: bool,
+    invalidation_domain: SegmentInvalidationDomain,
+    distributed_ec_invalidation: bool,
+    cold_zero_walk_oracle: bool,
 }
 
 fn producer_name(kind: ProducerKind) -> &'static str {
@@ -85,6 +95,9 @@ fn trusted_fixture_proposal(
         || envelope.key_format != proof.key_format
         || envelope.baseline_scan_plan_digest != proof.baseline_scan_plan_digest
         || envelope.process_epoch != proof.process_epoch
+        || !proof.durable_producer_identity
+        || !proof.cold_zero_walk_oracle
+        || (proof.invalidation_domain == SegmentInvalidationDomain::DistributedEc && !proof.distributed_ec_invalidation)
         || envelope.generation_start == 0
         || envelope.generation_end < envelope.generation_start
         || envelope.restart_gap
@@ -166,6 +179,10 @@ fn segment_observation_trusted_proposal_requires_identity_and_complete_producer_
         key_format: DATA_USAGE_CACHE_KEY_FORMAT,
         baseline_scan_plan_digest: baseline,
         process_epoch: "epoch-a",
+        durable_producer_identity: true,
+        invalidation_domain: SegmentInvalidationDomain::LocalSingleSet,
+        distributed_ec_invalidation: false,
+        cold_zero_walk_oracle: true,
     };
 
     assert_eq!(
@@ -193,6 +210,10 @@ fn segment_observation_trusted_proposal_requires_identity_and_complete_producer_
     wrong_epoch.process_epoch = "epoch-b";
     assert_eq!(trusted_fixture_proposal(&wrong_epoch, &proof), Err(ProposalError::InvalidKey));
 
+    let mut no_durable_identity = proof.clone();
+    no_durable_identity.durable_producer_identity = false;
+    assert_eq!(trusted_fixture_proposal(&envelope, &no_durable_identity), Err(ProposalError::InvalidKey));
+
     let mut restart_gap = envelope.clone();
     restart_gap.restart_gap = true;
     assert_eq!(trusted_fixture_proposal(&restart_gap, &proof), Err(ProposalError::InvalidKey));
@@ -208,6 +229,27 @@ fn segment_observation_trusted_proposal_requires_identity_and_complete_producer_
     let mut missing_producer = envelope.clone();
     missing_producer.producers.remove(producer_name(ProducerKind::Replication));
     assert_eq!(trusted_fixture_proposal(&missing_producer, &proof), Err(ProposalError::InvalidKey));
+
+    let mut missing_zero_walk_oracle = proof.clone();
+    missing_zero_walk_oracle.cold_zero_walk_oracle = false;
+    assert_eq!(
+        trusted_fixture_proposal(&envelope, &missing_zero_walk_oracle),
+        Err(ProposalError::InvalidKey)
+    );
+
+    let mut distributed_without_invalidation = proof.clone();
+    distributed_without_invalidation.invalidation_domain = SegmentInvalidationDomain::DistributedEc;
+    assert_eq!(
+        trusted_fixture_proposal(&envelope, &distributed_without_invalidation),
+        Err(ProposalError::InvalidKey)
+    );
+
+    let mut distributed_with_invalidation = distributed_without_invalidation;
+    distributed_with_invalidation.distributed_ec_invalidation = true;
+    assert_eq!(
+        trusted_fixture_proposal(&envelope, &distributed_with_invalidation),
+        Ok(BTreeSet::from(["archive".to_string(), "hot".to_string()]))
+    );
 }
 
 fn cache_value(cache: &DataUsageCache) -> serde_json::Value {

crates/scanner/src/scanner_io/tests.rs

@@ -1651,6 +1651,82 @@ fn scoped_scan_uses_only_locally_verified_prefix_hints() {
     assert!(distributed_scope.prefix_scope_for("photos").is_none());
 }
 
+#[test]
+fn remote_dirty_usage_invalidates_local_prefix_hints_until_distributed_proof_exists() {
+    let source = DataUsageCacheSource::new(1, 2);
+    let expected_sources = HashSet::from([source]);
+    let scan_plan_digest = DataUsageScanPlanDigest([6; 32]);
+    let baseline = complete_usage_baseline(source, scan_plan_digest, 7, 11);
+    let expected_peers = HashMap::from([(
+        "node-a:9000".to_string(),
+        ScannerPeerDirtyUsageExpectation {
+            instance_id: "instance-a".to_string(),
+            generation: 7,
+            pending: true,
+        },
+    )]);
+    let remote_dirty_usage = verified_remote_dirty_usage(
+        &expected_peers,
+        vec![(
+            "node-a:9000".to_string(),
+            peer_dirty_usage_snapshot("instance-a", 7, true, &[("photos", 7)]),
+        )],
+    )
+    .expect("fixture remote dirty usage should verify at bucket granularity");
+    let dirty_scopes = HashMap::from([(
+        "photos".to_string(),
+        DirtyUsageBucketScope::TopLevelEntries(HashSet::from(["2026".to_string()])),
+    )]);
+    let locally_scoped = scoped_scan_scope_from_dirty_buckets(
+        ScannerBucketScanScope::default(),
+        HashSet::from(["photos".to_string()]),
+        Some(&dirty_scopes),
+        true,
+        &[bucket_info("photos")],
+        ScannerCacheBaselineProof {
+            authoritative_data: Some(&baseline),
+            observed_candidate_data: None,
+            expected_sources: &expected_sources,
+            leader_epoch: 11,
+            want_cycle: 8,
+            scan_plan_digest,
+        },
+    );
+    assert!(
+        locally_scoped.prefix_scope_for("photos").is_some(),
+        "local-only evidence may narrow to a direct child segment"
+    );
+
+    let distributed = resolve_remote_dirty_usage_scope(
+        ScannerBucketScanScope::default(),
+        HashSet::from(["photos".to_string()]),
+        remote_dirty_usage,
+        &[bucket_info("photos")],
+        ScannerCacheBaselineProof {
+            authoritative_data: Some(&baseline),
+            observed_candidate_data: None,
+            expected_sources: &expected_sources,
+            leader_epoch: 11,
+            want_cycle: 8,
+            scan_plan_digest,
+        },
+    );
+
+    assert_eq!(
+        distributed
+            .scope
+            .selected_buckets
+            .as_deref()
+            .expect("distributed invalidation still selects the dirty bucket"),
+        &HashSet::from(["photos".to_string()])
+    );
+    assert!(
+        distributed.scope.prefix_scope_for("photos").is_none(),
+        "peer dirty state is not a distributed segment invalidation proof"
+    );
+    assert_eq!(distributed.remote_dirty_usage_acknowledgements.len(), 1);
+}
+
 fn peer_dirty_usage_snapshot(
     instance_id: &str,
     generation: u64,