From 47f640f23e354368fde6dee72b3ac4b3a45f1c05 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 19:30:43 +0800 Subject: [PATCH] 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 --- crates/scanner/src/scanner_folder/tests.rs | 76 +++++++++++++++++++ .../checkpoint_fixture/segment_observation.rs | 42 ++++++++++ crates/scanner/src/scanner_io/tests.rs | 76 +++++++++++++++++++ 3 files changed, 194 insertions(+) diff --git a/crates/scanner/src/scanner_folder/tests.rs b/crates/scanner/src/scanner_folder/tests.rs index 772c11d8b..385de36fd 100644 --- a/crates/scanner/src/scanner_folder/tests.rs +++ b/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) { + 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::::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() { diff --git a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/segment_observation.rs b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/segment_observation.rs index a141b412b..9e086b8dd 100644 --- a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/segment_observation.rs +++ b/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 { diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index c45ca1f7e..5189af0d0 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/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,