From 82ac2dff051bd19c90e226a28784e57c1007b173 Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 9 Sep 2026 08:34:50 +0800 Subject: [PATCH] feat(scanner): admit durable segment proof replay (#7544) * feat(scanner): admit durable segment proof evidence Validate complete set snapshot segment invalidation proof metadata against the current dirty usage generation window and scanner process epoch before clearing durable producer and restart-gap activation blockers. Production segment reuse remains gated by the explicit activation flag. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * feat(scanner): replay segment proof from root snapshots Carry segment invalidation proof metadata into root snapshot set states so a complete published baseline can replay the durable producer evidence recorded by each set cache. Keep the field additive for older readers and leave incomplete or LKG set states unproven. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: zhi22915 --- crates/data-usage/src/data_usage.rs | 52 +++++++++++++++ crates/scanner/src/data_usage_define.rs | 20 ++---- crates/scanner/src/scanner_io.rs | 41 ++++++++++++ crates/scanner/src/scanner_io/cache.rs | 7 +- crates/scanner/src/scanner_io/io_cycle.rs | 2 +- .../src/scanner_io/publish_gate_tests.rs | 24 ++++++- crates/scanner/src/scanner_io/tests.rs | 64 +++++++++++++++++++ 7 files changed, 191 insertions(+), 19 deletions(-) diff --git a/crates/data-usage/src/data_usage.rs b/crates/data-usage/src/data_usage.rs index 68a4d9822..93f325a1c 100644 --- a/crates/data-usage/src/data_usage.rs +++ b/crates/data-usage/src/data_usage.rs @@ -593,6 +593,18 @@ pub struct DataUsageSnapshotIdentity { pub scanner_epoch: Option, } +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct DataUsageSegmentInvalidationProof { + #[serde(default)] + pub process_epoch: String, + #[serde(default)] + pub generation_start: u64, + #[serde(default)] + pub generation_end: u64, + #[serde(default)] + pub producer_identity_coverage_complete: bool, +} + #[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] pub struct DataUsageSnapshotSetState { pub pool_index: u64, @@ -607,6 +619,8 @@ pub struct DataUsageSnapshotSetState { pub complete: bool, #[serde(default)] pub tombstone: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub segment_invalidation_proof: Option, } impl DataUsageInfo { @@ -3073,6 +3087,7 @@ mod tests { scan_plan_digest: Some([1; 32]), complete: false, tombstone: false, + segment_invalidation_proof: None, }]; assert!(observed_data_usage_is_newer(&partial, &authoritative)); } @@ -3095,6 +3110,7 @@ mod tests { scan_plan_digest: Some([1; 32]), complete: true, tombstone: false, + segment_invalidation_proof: None, }, DataUsageSnapshotSetState { pool_index: 1, @@ -3104,6 +3120,7 @@ mod tests { scan_plan_digest: Some([2; 32]), complete: false, tombstone: false, + segment_invalidation_proof: None, }, ], ..Default::default() @@ -3113,6 +3130,41 @@ mod tests { assert!(partial.is_valid_partial_snapshot()); } + #[test] + fn set_state_segment_invalidation_proof_is_additive() { + #[derive(Deserialize)] + struct LegacySetState { + pool_index: u64, + set_index: u64, + complete: bool, + } + + let proof = DataUsageSegmentInvalidationProof { + process_epoch: "scanner-process".to_string(), + generation_start: 3, + generation_end: 5, + producer_identity_coverage_complete: true, + }; + let state = DataUsageSnapshotSetState { + pool_index: 1, + set_index: 2, + scanner_cycle: Some(9), + scanner_epoch: Some(4), + scan_plan_digest: Some([7; 32]), + complete: true, + tombstone: false, + segment_invalidation_proof: Some(proof.clone()), + }; + let encoded = rmp_serde::to_vec_named(&state).expect("set state should encode with additive proof"); + let legacy: LegacySetState = rmp_serde::from_slice(&encoded).expect("legacy readers should ignore proof metadata"); + assert_eq!(legacy.pool_index, 1); + assert_eq!(legacy.set_index, 2); + assert!(legacy.complete); + + let decoded: DataUsageSnapshotSetState = rmp_serde::from_slice(&encoded).expect("new readers should restore proof"); + assert_eq!(decoded.segment_invalidation_proof, Some(proof)); + } + #[test] fn completeness_marker_requires_a_snapshot_timestamp() { let untimestamped = DataUsageInfo { diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 95283b1cd..3634dfd35 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -28,10 +28,10 @@ use metrics::{counter, describe_counter, describe_histogram, histogram}; use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS; pub use rustfs_data_usage::{ AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME, - DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, DataUsageSnapshotSetState, LEGACY_DATA_USAGE_OBJECT_NAME, - PrefixUsageEntry, PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeReconciliationEntry, - SizeReconciliationScope, SizeSummary, TierAccountingProof, TierStats, UNKNOWN_TIER, UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP, - UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP, UnknownTierStats, hash_path, prefix_usage_in_cache, + DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, DataUsageSegmentInvalidationProof, DataUsageSnapshotSetState, + LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry, PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, + SizeReconciliationEntry, SizeReconciliationScope, SizeSummary, TierAccountingProof, TierStats, UNKNOWN_TIER, + UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP, UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP, UnknownTierStats, hash_path, prefix_usage_in_cache, }; use rustfs_heal_contracts::heal_channel::HealScanMode; use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; @@ -564,18 +564,6 @@ impl DataUsageCacheSource { #[serde(transparent)] pub struct DataUsageScanPlanDigest(pub [u8; 32]); -#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] -pub struct DataUsageSegmentInvalidationProof { - #[serde(default)] - pub process_epoch: String, - #[serde(default)] - pub generation_start: u64, - #[serde(default)] - pub generation_end: u64, - #[serde(default)] - pub producer_identity_coverage_complete: bool, -} - #[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum PendingScannerHealKind { diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index c197e3e78..7811c9fb6 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -539,6 +539,47 @@ fn scanner_segment_reuse_activation_preflight_for_cycle( }) } +fn scanner_durable_segment_invalidation_evidence( + dirty_usage_snapshot: &DirtyUsageSnapshot, + results: &[DataUsageCache], + expected_sources: &HashSet, +) -> DirtyUsageProducerEvidence { + let mut evidence = dirty_usage_producer_evidence(dirty_usage_snapshot); + if !evidence.generation_window_bound + || !evidence.producer_identity_coverage_complete + || !scanner_results_form_complete_snapshot(results, expected_sources) + { + return evidence; + } + + let mut covered_sources = HashSet::with_capacity(expected_sources.len()); + let all_sets_proved = results.iter().all(|result| { + let Some(source) = result.info.source else { + return false; + }; + expected_sources.contains(&source) + && covered_sources.insert(source) + && scanner_segment_invalidation_proof_matches(result.info.segment_invalidation_proof.as_ref(), &evidence) + }); + if all_sets_proved && covered_sources.len() == expected_sources.len() { + evidence.durable_producer_identity = true; + evidence.restart_gap_absent = true; + } + evidence +} + +fn scanner_segment_invalidation_proof_matches( + proof: Option<&crate::DataUsageSegmentInvalidationProof>, + evidence: &DirtyUsageProducerEvidence, +) -> bool { + proof.is_some_and(|proof| { + proof.process_epoch == scanner_activity_epoch() + && proof.generation_start == evidence.generation_start + && proof.generation_end == evidence.generation_end + && proof.producer_identity_coverage_complete + }) +} + fn scanner_segment_reuse_activated() -> bool { scanner_segment_reuse_activation_preflight().scanner_segment_reuse_activated } diff --git a/crates/scanner/src/scanner_io/cache.rs b/crates/scanner/src/scanner_io/cache.rs index 328b91ce9..2eb88089f 100644 --- a/crates/scanner/src/scanner_io/cache.rs +++ b/crates/scanner/src/scanner_io/cache.rs @@ -464,6 +464,7 @@ pub(super) fn completed_usage_candidate( scan_plan_digest: Some(result.info.scan_plan_digest?.0), complete: true, tombstone: false, + segment_invalidation_proof: result.info.segment_invalidation_proof.clone(), }) }) .collect::>>()?; @@ -642,13 +643,14 @@ pub(super) fn observational_data_usage_info( let current_snapshot = current.is_some(); let selected = current.or(lkg); if let Some(selected) = selected { - let (cycle, epoch, digest, last_update, complete) = if current_snapshot { + let (cycle, epoch, digest, last_update, complete, segment_invalidation_proof) = if current_snapshot { ( Some(selected.info.next_cycle), Some(selected.info.leader_epoch), selected.info.scan_plan_digest.map(|digest| digest.0), selected.info.last_update, true, + selected.info.segment_invalidation_proof.clone(), ) } else { ( @@ -657,6 +659,7 @@ pub(super) fn observational_data_usage_info( selected.info.lkg_scan_plan_digest.map(|digest| digest.0), selected.info.lkg_last_update, false, + None, ) }; set_states.push(DataUsageSnapshotSetState { @@ -667,6 +670,7 @@ pub(super) fn observational_data_usage_info( scan_plan_digest: digest, complete, tombstone: false, + segment_invalidation_proof, }); usable.push((selected, last_update)); } else { @@ -678,6 +682,7 @@ pub(super) fn observational_data_usage_info( scan_plan_digest: Some(expected_plan_digest.0), complete: false, tombstone: false, + segment_invalidation_proof: None, }); } } diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index 1793bfb50..3b476d8d1 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -715,7 +715,7 @@ where ); let segment_reuse_activation_preflight = scanner_segment_reuse_activation_preflight_for_cycle( &dirty_usage_snapshot, - dirty_usage_producer_evidence(&dirty_usage_snapshot), + scanner_durable_segment_invalidation_evidence(&dirty_usage_snapshot, &results, &expected_sources), distributed, distributed_segment_invalidation_evidence, cold_zero_walk_oracle, diff --git a/crates/scanner/src/scanner_io/publish_gate_tests.rs b/crates/scanner/src/scanner_io/publish_gate_tests.rs index 6274f27fa..8a662fcdf 100644 --- a/crates/scanner/src/scanner_io/publish_gate_tests.rs +++ b/crates/scanner/src/scanner_io/publish_gate_tests.rs @@ -13,7 +13,7 @@ // limitations under the License. use super::*; -use crate::data_usage_define::{UNKNOWN_TIER, UnknownTierStats, hash_path}; +use crate::data_usage_define::{DataUsageSegmentInvalidationProof, UNKNOWN_TIER, UnknownTierStats, hash_path}; use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage, TierAccountingProof}; const TEST_PLAN_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([7; 32]); @@ -176,6 +176,26 @@ fn completed_data_usage_info_rejects_duplicate_bucket_inventory() { assert!(completed_data_usage_info_for_test(&[set], &buckets, false, false).is_none()); } +#[test] +fn completed_data_usage_info_carries_segment_invalidation_proof_to_set_state() { + let source = DataUsageCacheSource::new(0, 0); + let proof = DataUsageSegmentInvalidationProof { + process_epoch: "scanner-process".to_string(), + generation_start: 5, + generation_end: 8, + producer_identity_coverage_complete: true, + }; + let mut set = completed_root_cache("bucket", 2, 10, source); + set.info.segment_invalidation_proof = Some(proof.clone()); + + let (usage, _) = + completed_usage_for_scope(&[set], &HashSet::from([source]), &["bucket".to_string()], &[], true, false, false) + .expect("complete set should publish root usage"); + + assert_eq!(usage.usage_snapshot_set_states.len(), 1); + assert_eq!(usage.usage_snapshot_set_states[0].segment_invalidation_proof, Some(proof)); +} + #[test] fn completed_data_usage_info_rejects_extra_or_detached_bucket_data() { let buckets = vec!["bucket".to_string()]; @@ -350,6 +370,7 @@ fn set_membership_add_remove_uses_generation_and_tombstone() { scan_plan_digest: Some(TEST_PLAN_DIGEST.0), complete: false, tombstone: true, + segment_invalidation_proof: None, }; let encoded = serde_json::to_vec(&state).expect("set state should serialize"); let decoded: DataUsageSnapshotSetState = serde_json::from_slice(&encoded).expect("set state should deserialize"); @@ -371,6 +392,7 @@ fn set_membership_add_remove_uses_generation_and_tombstone() { scan_plan_digest: Some(TEST_PLAN_DIGEST.0), complete: true, tombstone: false, + segment_invalidation_proof: None, }, state, ], diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 7bd1ae71b..ee3d1cc63 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -237,6 +237,49 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_skips_distributed_blocke ); } +#[test] +#[serial] +fn scanner_durable_segment_invalidation_evidence_requires_matching_complete_set_proofs() { + use crate::segment_invalidation::SegmentInvalidationProducerIdentity; + + clear_dirty_usage_buckets_for_tests(); + record_dirty_usage_bucket_from_producers("photos", SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION); + let dirty_usage_snapshot = snapshot_dirty_usage_buckets(&[bucket_info("photos")], dirty_usage_generation()); + let process_proof = dirty_usage_producer_evidence(&dirty_usage_snapshot) + .segment_invalidation_proof() + .expect("complete process-local producer coverage should produce proof metadata"); + let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0), DataUsageCacheSource::new(0, 1)]); + let results = vec![ + complete_set_cache_with_segment_proof(DataUsageCacheSource::new(0, 0), process_proof.clone()), + complete_set_cache_with_segment_proof(DataUsageCacheSource::new(0, 1), process_proof.clone()), + ]; + + let durable_evidence = scanner_durable_segment_invalidation_evidence(&dirty_usage_snapshot, &results, &expected_sources); + + assert!(durable_evidence.producer_identity_coverage_complete); + assert!(durable_evidence.durable_producer_identity); + assert!(durable_evidence.restart_gap_absent); + + let mut stale_epoch = results.clone(); + stale_epoch[0] + .info + .segment_invalidation_proof + .as_mut() + .expect("proof fixture should exist") + .process_epoch = "stale-process".to_string(); + let stale_evidence = scanner_durable_segment_invalidation_evidence(&dirty_usage_snapshot, &stale_epoch, &expected_sources); + assert!(stale_evidence.producer_identity_coverage_complete); + assert!(!stale_evidence.durable_producer_identity); + assert!(!stale_evidence.restart_gap_absent); + + record_dirty_usage_bucket("videos"); + let changed_evidence = scanner_durable_segment_invalidation_evidence(&dirty_usage_snapshot, &results, &expected_sources); + assert!(!changed_evidence.producer_identity_coverage_complete); + assert!(!changed_evidence.durable_producer_identity); + assert!(!changed_evidence.restart_gap_absent); + clear_dirty_usage_buckets_for_tests(); +} + #[test] fn scanner_cycle_result_returns_segment_reuse_activation_preflight() { let proof = ScannerSegmentReuseActivationProof { @@ -275,6 +318,26 @@ fn complete_process_local_producer_evidence() -> DirtyUsageProducerEvidence { } } +fn complete_set_cache_with_segment_proof( + source: DataUsageCacheSource, + proof: crate::DataUsageSegmentInvalidationProof, +) -> DataUsageCache { + DataUsageCache { + info: DataUsageCacheInfo { + name: DATA_USAGE_ROOT.to_string(), + next_cycle: 7, + last_update: Some(SystemTime::UNIX_EPOCH), + leader_epoch: 11, + source: Some(source), + snapshot_complete: true, + scan_plan_digest: Some(DataUsageScanPlanDigest([3; 32])), + segment_invalidation_proof: Some(proof), + ..Default::default() + }, + cache: HashMap::new(), + } +} + async fn setup_two_pool_scanner_store() -> (tempfile::TempDir, Arc) { init_ecstore_config_for_scanner_tests(); let temp_dir = tempfile::tempdir().expect("multi-pool scanner test directory should be created"); @@ -1623,6 +1686,7 @@ fn complete_usage_baseline( scan_plan_digest: Some(scan_plan_digest.0), complete: true, tombstone: false, + segment_invalidation_proof: None, }], ..Default::default() };