From a26025d2db46ea0e3291ea4abe5300d3b5a1b8f3 Mon Sep 17 00:00:00 2001 From: overtrue Date: Sat, 5 Sep 2026 19:40:15 +0800 Subject: [PATCH] fix(scanner): fence cached snapshots by scan execution --- crates/scanner/src/data_usage_define.rs | 12 +- crates/scanner/src/data_usage_define/tests.rs | 4 + crates/scanner/src/scanner_io.rs | 6 +- crates/scanner/src/scanner_io/cache.rs | 42 +++++-- crates/scanner/src/scanner_io/io_cache.rs | 40 +++--- crates/scanner/src/scanner_io/io_cycle.rs | 10 +- crates/scanner/src/scanner_io/tests.rs | 118 ++++++++++++++++++ .../architecture/scanner-usage-publication.md | 24 +++- 8 files changed, 217 insertions(+), 39 deletions(-) diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 35229ba6b..dbc803839 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -196,7 +196,7 @@ pub(crate) async fn read_config_revision(store: Arc, path } } -#[derive(Clone, Debug)] +#[derive(Clone, Debug, PartialEq, Eq)] pub(crate) struct DataUsageCacheRevisions { main: DataUsageCacheRevision, backup: Option, @@ -503,6 +503,10 @@ pub struct DataUsageCacheInfo { pub lkg_leader_epoch: Option, #[serde(default)] pub lkg_scan_plan_digest: Option, + /// Activity-sensitive identity for same-cycle set snapshot reuse. The + /// structural plan remains reusable across ordinary bucket writes. + #[serde(default)] + pub scan_execution_digest: Option, } impl Serialize for DataUsageCacheInfo { @@ -519,7 +523,8 @@ impl Serialize for DataUsageCacheInfo { + usize::from(self.lkg_next_cycle.is_some()) + usize::from(self.lkg_last_update.is_some()) + usize::from(self.lkg_leader_epoch.is_some()) - + usize::from(self.lkg_scan_plan_digest.is_some()); + + usize::from(self.lkg_scan_plan_digest.is_some()) + + usize::from(self.scan_execution_digest.is_some()); let mut state = serializer.serialize_map(Some(field_count))?; state.serialize_entry("name", &self.name)?; state.serialize_entry("next_cycle", &self.next_cycle)?; @@ -558,6 +563,9 @@ impl Serialize for DataUsageCacheInfo { if let Some(scan_plan_digest) = self.lkg_scan_plan_digest { state.serialize_entry("lkg_scan_plan_digest", &scan_plan_digest)?; } + if let Some(scan_execution_digest) = self.scan_execution_digest { + state.serialize_entry("scan_execution_digest", &scan_execution_digest)?; + } state.end() } } diff --git a/crates/scanner/src/data_usage_define/tests.rs b/crates/scanner/src/data_usage_define/tests.rs index 624ba13a3..19f4f778c 100644 --- a/crates/scanner/src/data_usage_define/tests.rs +++ b/crates/scanner/src/data_usage_define/tests.rs @@ -1067,6 +1067,7 @@ fn test_data_usage_cache_info_deserialize_defaults_scan_resume_after() { assert!(decoded.source.is_none()); assert!(!decoded.snapshot_complete); assert!(decoded.scan_plan_digest.is_none()); + assert!(decoded.scan_execution_digest.is_none()); assert_eq!(decoded.cache_key_format, 0); } @@ -1109,6 +1110,7 @@ fn test_data_usage_cache_info_unmarshal_old_msgpack_defaults_scan_resume_after() assert!(decoded.source.is_none()); assert!(!decoded.snapshot_complete); assert!(decoded.scan_plan_digest.is_none()); + assert!(decoded.scan_execution_digest.is_none()); assert_eq!(decoded.cache_key_format, 0); } @@ -1145,6 +1147,7 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() { source: Some(DataUsageCacheSource::new(1, 2)), snapshot_complete: true, scan_plan_digest: Some(TEST_PLAN_DIGEST), + scan_execution_digest: Some(DataUsageScanPlanDigest([42; 32])), cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, ..Default::default() }, @@ -1164,6 +1167,7 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() { assert_eq!(current.info.source, Some(DataUsageCacheSource::new(1, 2))); assert!(current.info.snapshot_complete); assert_eq!(current.info.scan_plan_digest, Some(TEST_PLAN_DIGEST)); + assert_eq!(current.info.scan_execution_digest, Some(DataUsageScanPlanDigest([42; 32]))); assert_eq!(current.info.cache_key_format, DATA_USAGE_CACHE_KEY_FORMAT); assert_eq!(current.find("bucket").map(|entry| entry.objects), Some(3)); diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index b9a9a997e..f57df0eec 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::data_usage_define::DATA_USAGE_CACHE_KEY_FORMAT; +use crate::data_usage_define::{DATA_USAGE_CACHE_KEY_FORMAT, DataUsageCacheRevisions}; use crate::scanner_budget::ScannerCycleBudget; use crate::scanner_folder::{ScannerItem, scan_data_folder}; use crate::sleeper::SCANNER_SLEEPER; @@ -271,8 +271,8 @@ pub struct ScannerBucketScanPlan { all_buckets: Arc>, scope: ScannerBucketScanScope, digest: DataUsageScanPlanDigest, - // Bucket work must invalidate on namespace completion even when its scoped baseline remains reusable. - bucket_cache_digest: DataUsageScanPlanDigest, + // Cache work must invalidate on namespace completion even when its scoped baseline remains reusable. + execution_digest: DataUsageScanPlanDigest, leader_epoch: u64, tier_registry_generation: u64, /// Epoch captured once for the whole scanner cycle. `None` is retained diff --git a/crates/scanner/src/scanner_io/cache.rs b/crates/scanner/src/scanner_io/cache.rs index 7ff684c99..ba5203f20 100644 --- a/crates/scanner/src/scanner_io/cache.rs +++ b/crates/scanner/src/scanner_io/cache.rs @@ -604,10 +604,12 @@ pub(super) async fn persist_and_publish_cache_snapshot( store: Arc, updates: &mpsc::Sender, mut cache_snapshot: DataUsageCache, + initial_revisions: Option<&DataUsageCacheRevisions>, cache_cycle_floor: &AtomicU64, expected_publication_epoch: u64, ) -> Option { let source = cache_snapshot.info.source?; + let execution_digest = cache_snapshot.info.scan_execution_digest?; let guard = match acquire_scanner_cache_locks(store.as_ref(), DATA_USAGE_CACHE_NAME, source).await { Ok(guard) => guard, Err(err) => { @@ -672,20 +674,36 @@ pub(super) async fn persist_and_publish_cache_snapshot( ); return None; } - if matches!( - current_cache_root_entry_with_generation( - &persisted, - DATA_USAGE_ROOT, - source, - cache_snapshot.info.next_cycle, - cache_snapshot.info.leader_epoch, - scan_plan_digest, - cache_snapshot.info.tier_registry_generation, - ), - Ok(Some(_)) - ) { + if persisted.info.scan_execution_digest == Some(execution_digest) + && matches!( + current_cache_root_entry_with_generation( + &persisted, + DATA_USAGE_ROOT, + source, + cache_snapshot.info.next_cycle, + cache_snapshot.info.leader_epoch, + scan_plan_digest, + cache_snapshot.info.tier_registry_generation, + ), + Ok(Some(_)) + ) + { cache_snapshot = persisted; } else { + // A later execution may have completed while this scan was walking. + // Only replace the cache revision from which this scan started. + if initial_revisions != Some(&revisions) { + warn!( + target: "rustfs::scanner::io", + event = EVENT_SCANNER_CACHE_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_IO, + state = "scan_baseline_revision_changed", + cache_name = DATA_USAGE_CACHE_NAME, + "Scanner skipped set snapshot without an unchanged baseline revision" + ); + return None; + } if guard.is_lock_lost() { error!( target: "rustfs::scanner::io", diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index e24b1471e..b4f04481b 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -118,7 +118,7 @@ impl ScannerIOCache for SetDisks { all_buckets, scope, digest: scan_plan_digest, - bucket_cache_digest, + execution_digest, leader_epoch, tier_registry_generation, publication_epoch, @@ -138,20 +138,24 @@ impl ScannerIOCache for SetDisks { .ok_or_else(|| StorageError::other("scanner cache publication is blocked by data movement"))?, }; let mut old_cache = DataUsageCache::default(); - if let Err(e) = old_cache.load(self.clone(), DATA_USAGE_CACHE_NAME).await { - warn!( - target: "rustfs::scanner::io", - event = EVENT_SCANNER_CACHE_PERSIST_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_IO, - pool = self.pool_index, - set = self.set_index, - cache_name = DATA_USAGE_CACHE_NAME, - state = "old_cache_load_failed", - error = %e, - "Scanner old data usage cache load failed; rebuilding from bucket caches" - ); - } + let initial_revisions = match old_cache.load_with_revisions(self.clone(), DATA_USAGE_CACHE_NAME).await { + Ok(revisions) => Some(revisions), + Err(e) => { + warn!( + target: "rustfs::scanner::io", + event = EVENT_SCANNER_CACHE_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_IO, + pool = self.pool_index, + set = self.set_index, + cache_name = DATA_USAGE_CACHE_NAME, + state = "old_cache_load_failed", + error = %e, + "Scanner old data usage cache load failed; rebuilding from bucket caches" + ); + None + } + }; let scoped_scan = prepare_scoped_set_scan( &old_cache, &buckets, @@ -196,6 +200,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.lkg_snapshot_complete = false; cache.info.lkg_next_cycle = None; cache.info.lkg_last_update = None; @@ -209,6 +214,7 @@ impl ScannerIOCache for SetDisks { self, &updates, cache, + initial_revisions.as_ref(), cache_cycle_floor.as_ref(), expected_publication_epoch, ) @@ -638,7 +644,7 @@ impl ScannerIOCache for SetDisks { let cache_name = path_join_buf(&[&bucket.name, DATA_USAGE_CACHE_NAME]); let bucket_scan_plan_digest = - scanner_bucket_cache_digest(bucket_cache_digest, dirty_usage_buckets_clone.get(&bucket.name).copied()); + scanner_bucket_cache_digest(execution_digest, dirty_usage_buckets_clone.get(&bucket.name).copied()); if let Some(server_epoch) = remote_server_epoch { let request_sequence = remote_session_sequence; @@ -1361,6 +1367,7 @@ impl ScannerIOCache for SetDisks { 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.lkg_snapshot_complete = false; cache.info.lkg_next_cycle = None; cache.info.lkg_last_update = None; @@ -1372,6 +1379,7 @@ impl ScannerIOCache for SetDisks { self.clone(), &updates, cache_snapshot, + initial_revisions.as_ref(), cache_cycle_floor.as_ref(), expected_publication_epoch, ) diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index d1b0af3f9..ad0afd579 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -263,10 +263,10 @@ where let activity_digest = crate::scanner::scanner_activity_snapshot_digest(&activity_before); let scan_plan_digest = scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_structural_digest(&activity_before)); - let mut bucket_cache_hasher = Sha256::new(); - bucket_cache_hasher.update(scan_plan_digest.0); - bucket_cache_hasher.update(activity_digest); - let bucket_cache_digest = DataUsageScanPlanDigest(bucket_cache_hasher.finalize().into()); + let mut execution_hasher = Sha256::new(); + execution_hasher.update(scan_plan_digest.0); + execution_hasher.update(activity_digest); + let execution_digest = DataUsageScanPlanDigest(execution_hasher.finalize().into()); let dirty_usage_snapshot = Arc::new(snapshot_dirty_usage_buckets(&all_buckets, dirty_generation_before_bucket_list)); let scan_scope = resolve_scanner_bucket_scan_scope( store, @@ -416,7 +416,7 @@ where all_buckets: Arc::clone(&all_buckets), scope: scan_scope.clone(), digest: scan_plan_digest, - bucket_cache_digest, + execution_digest, leader_epoch, tier_registry_generation, publication_epoch, diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 480709388..89f81ff23 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -914,6 +914,124 @@ fn complete_set_usage_cache(buckets: &[(&str, usize)], scan_plan_digest: DataUsa cache } +#[tokio::test] +#[serial] +async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers() { + let (_temp_dir, store) = setup_two_pool_scanner_store().await; + let set = Arc::clone(&store.pools[0].disk_set[0]); + let epoch = scanner_publication_epoch(Arc::clone(&set)).await.expect("idle set admission"); + let mut legacy = complete_set_usage_cache(&[("photos", 5)], DataUsageScanPlanDigest([1; 32])); + legacy.info.source = Some(DataUsageCacheSource::new(0, 0)); + legacy + .save(Arc::clone(&set), DATA_USAGE_CACHE_NAME) + .await + .expect("seed legacy set cache"); + let mut persisted = DataUsageCache::default(); + let initial = persisted + .load_with_revisions(Arc::clone(&set), DATA_USAGE_CACHE_NAME) + .await + .expect("capture the shared starting revision"); + let mut fresh = legacy.clone(); + fresh.info.scan_execution_digest = Some(DataUsageScanPlanDigest([2; 32])); + fresh.replace( + "photos", + DATA_USAGE_ROOT, + DataUsageEntry { + size: 20, + objects: 1, + ..Default::default() + }, + ); + let cycle_floor = AtomicU64::new(fresh.info.next_cycle); + let (tx, mut rx) = mpsc::channel(1); + assert!( + persist_and_publish_cache_snapshot(Arc::clone(&set), &tx, fresh.clone(), Some(&initial), &cycle_floor, epoch) + .await + .is_some(), + "a legacy cache without execution identity must be refreshed" + ); + let published = rx.try_recv().expect("fresh snapshot should be forwarded"); + assert_eq!(published.find("photos").expect("published bucket").size, 20); + assert_eq!(published.info.scan_execution_digest, fresh.info.scan_execution_digest); + let current = persisted + .load_with_revisions(Arc::clone(&set), DATA_USAGE_CACHE_NAME) + .await + .expect("capture the current revision for the unidentified execution"); + + let mut stale = legacy.clone(); + stale.info.scan_execution_digest = Some(DataUsageScanPlanDigest([3; 32])); + for (candidate, revisions) in [(stale, &initial), (legacy, ¤t)] { + assert!( + persist_and_publish_cache_snapshot(Arc::clone(&set), &tx, candidate, Some(revisions), &cycle_floor, epoch) + .await + .is_none(), + "a stale or unidentified execution must not replace the newer snapshot" + ); + assert!(matches!(rx.try_recv(), Err(mpsc::error::TryRecvError::Empty))); + } + fresh.info.scan_execution_digest = Some(DataUsageScanPlanDigest([4; 32])); + assert!( + persist_and_publish_cache_snapshot(Arc::clone(&set), &tx, fresh.clone(), None, &cycle_floor, epoch) + .await + .is_none(), + "an unreadable starting revision must not authorize an overwrite" + ); + + fresh.info.scan_execution_digest = published.info.scan_execution_digest; + fresh.replace("photos", DATA_USAGE_ROOT, DataUsageEntry::default()); + assert!( + persist_and_publish_cache_snapshot(Arc::clone(&set), &tx, fresh, Some(&initial), &cycle_floor, epoch) + .await + .is_some(), + "an overlapping identical execution must reuse the completed snapshot" + ); + assert_eq!( + rx.try_recv() + .expect("reused snapshot") + .find("photos") + .expect("reused bucket") + .size, + 20 + ); + persisted + .load(Arc::clone(&set), DATA_USAGE_CACHE_NAME) + .await + .expect("read the final durable set cache"); + assert_eq!(persisted.find("photos").expect("durable bucket").size, 20); + assert_eq!(persisted.info.scan_execution_digest, published.info.scan_execution_digest); + + let ctx = CancellationToken::new(); + let empty_execution = DataUsageScanPlanDigest([5; 32]); + set.nsscanner_cache( + ctx.clone(), + ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default()), + ScannerBucketScanPlan { + buckets: Vec::new(), + all_buckets: Arc::new(Vec::new()), + scope: ScannerBucketScanScope::default(), + digest: DataUsageScanPlanDigest([6; 32]), + execution_digest: empty_execution, + leader_epoch: 11, + tier_registry_generation: 13, + publication_epoch: Some(epoch), + dirty_usage_buckets: Arc::new(HashMap::new()), + bucket_failures: ScannerBucketFailureState::default(), + pending_maintenance_work: Arc::new(AtomicBool::new(false)), + cache_cycle_floor: Arc::new(AtomicU64::new(8)), + }, + tx, + 8, + HealScanMode::Normal, + ) + .await + .expect("empty set scope should replace its prior nonempty cache"); + let empty = rx.try_recv().expect("empty set snapshot should be published"); + assert_eq!(empty.info.scan_execution_digest, Some(empty_execution)); + assert!(empty.info.snapshot_complete); + let root = empty.checked_flatten(DATA_USAGE_ROOT).expect("complete empty root"); + assert_eq!((root.size, root.objects), (0, 0)); +} + fn complete_usage_baseline( source: DataUsageCacheSource, scan_plan_digest: DataUsageScanPlanDigest, diff --git a/docs/architecture/scanner-usage-publication.md b/docs/architecture/scanner-usage-publication.md index 7f0a57719..0665fc473 100644 --- a/docs/architecture/scanner-usage-publication.md +++ b/docs/architecture/scanner-usage-publication.md @@ -57,7 +57,7 @@ They must not be collapsed unless the replacement proves the same exclusions. | Scanner leadership claim | scanner | competing scanner leaders and stale cycle writers | | Storage publication epoch | ECStore | usage computed across rebalance, decommission, or other data-movement generations | | Publication lease | scanner peers through ECStore-facing activity probes | remote dirty-usage or maintenance state that has not acknowledged the candidate | -| CAS revision | backing config object store | lost updates to `.usage.v2.json`, `.usage.json`, or cycle-state objects | +| CAS revision | backing config object store | lost updates to usage snapshots, scanner caches, or cycle-state objects | | Per-set freshness | scanner aggregation | a merged usage snapshot that combines stale and current set results | | Tier registry generation | scanner tier accounting | bytes classified against a different warm-tier registry | | Usage floor identity | scanner publication and ECStore quota fallback | empty or legacy values becoming plausible authoritative quota input | @@ -66,6 +66,28 @@ A reader that cannot prove the required fence for its surface must fail closed or use the documented observed path below. It must not synthesize an empty usage snapshot for a missing or corrupt authoritative object. +## Cache Execution Identity + +The structural scan-plan digest can remain stable across ordinary bucket writes +so a scoped scan can retain unaffected baseline buckets. It is not sufficient +proof for reusing a completed result within the same cycle. Bucket work uses an +execution digest combining the structural plan and the full activity snapshot, +with the bucket's dirty generation included in its cache identity. Completed set +caches carry the same execution digest separately from their structural plan. +The persisted set-root fast path requires equal execution identities as well as +the existing source, cycle, leader, tier, and cache-structure checks. + +A set scan also captures its starting cache revisions. When the persisted +execution differs, replacement requires those revisions to remain unchanged; +otherwise a slow scan could overwrite a newer completed result. The existing +cache lock, conditional save, and movement admission still fence the commit. + +The optional `scan_execution_digest` field is appended to the map-encoded cache +metadata. Legacy caches remain readable but cannot satisfy same-cycle set-root +reuse without this identity. Older readers can ignore the added map key, but +older writers do not enforce its fence; readability is not a mixed-version +publication-safety guarantee. + ## Persisted Objects The persisted objects are part of the compatibility contract. Removing one