fix(scanner): fence cached snapshots by scan execution

crates/scanner/src/data_usage_define.rs

@@ -196,7 +196,7 @@ pub(crate) async fn read_config_revision<S: ScannerObjectIO>(store: Arc<S>, path
     }
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub(crate) struct DataUsageCacheRevisions {
     main: DataUsageCacheRevision,
     backup: Option<DataUsageCacheRevision>,
@@ -503,6 +503,10 @@ pub struct DataUsageCacheInfo {
     pub lkg_leader_epoch: Option<u64>,
     #[serde(default)]
     pub lkg_scan_plan_digest: Option<DataUsageScanPlanDigest>,
+    /// 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<DataUsageScanPlanDigest>,
 }
 
 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()
     }
 }

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));
 

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<Vec<BucketInfo>>,
     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

crates/scanner/src/scanner_io/cache.rs

@@ -604,10 +604,12 @@ pub(super) async fn persist_and_publish_cache_snapshot(
     store: Arc<SetDisks>,
     updates: &mpsc::Sender<DataUsageCache>,
     mut cache_snapshot: DataUsageCache,
+    initial_revisions: Option<&DataUsageCacheRevisions>,
     cache_cycle_floor: &AtomicU64,
     expected_publication_epoch: u64,
 ) -> Option<SystemTime> {
     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",

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,
             )

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,

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, &current)] {
+        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,