fix(scanner): invalidate bucket work after namespace completion

crates/scanner/src/scanner/tests.rs

@@ -15,7 +15,7 @@
 use super::heal_info::{classify_background_heal_read_error, decode_background_heal_info};
 use super::*;
 use crate::EcstoreResult;
-use crate::storage_api::ecstore_hold_namespace_commit;
+use crate::storage_api::owner::ecstore_hold_namespace_commit;
 use crate::storage_api::scan::{BucketOperations as _, ObjectIO as _};
 use crate::{
     DATA_USAGE_BLOOM_RECOVERY_PATH, DATA_USAGE_CACHE_KEY_FORMAT, DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT,
@@ -1221,13 +1221,34 @@ async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledgin
         "the pending candidate must not replace the authoritative baseline"
     );
 
+    let committed_body = b"committed-after-walk";
+    let mut reader = PutObjReader::from_vec(committed_body.to_vec());
+    store.pools[0].disk_set[0]
+        .put_object(
+            &bucket,
+            "object",
+            &mut reader,
+            &ObjectOptions {
+                no_lock: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("the pending tail must change the physical object before it drains");
+    assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty_before);
     drop(pending);
+    let retry_budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, ScannerCycleBudgetConfig::default());
     let outcome = tokio::time::timeout(
         Duration::from_secs(30),
-        run_data_scanner_cycle(&ctx, &store, &mut cycle_info, &mut revision, 1),
+        run_data_scanner_cycle_with_budget(&ctx, &store, &mut cycle_info, &mut revision, 1, Arc::clone(&retry_budget)),
     )
     .await
     .expect("the same cycle must converge after the pending PUT drains");
+    assert_eq!(
+        retry_budget.progress().0,
+        1,
+        "the same-cycle retry must not reuse the pre-tail bucket cache"
+    );
     assert!(matches!(
         outcome,
         ScannerCycleOutcome::Completed | ScannerCycleOutcome::CompletedWithPendingMaintenance
@@ -1241,13 +1262,16 @@ async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledgin
     assert_eq!(usage.usage_snapshot_converged, Some(true));
     assert_eq!(usage.scanner_cycle, Some(1));
     assert_eq!(usage.objects_total_count, 1);
-    assert_eq!(usage.objects_total_size, 5);
+    assert_eq!(
+        usage.objects_total_size,
+        u64::try_from(committed_body.len()).expect("fixture body length")
+    );
     let bucket_usage = usage
         .buckets_usage
         .get(&bucket)
         .expect("the scanned bucket should be published");
     assert_eq!(bucket_usage.objects_count, 1);
-    assert_eq!(bucket_usage.size, 5);
+    assert_eq!(bucket_usage.size, u64::try_from(committed_body.len()).expect("fixture body length"));
     global_metrics().set_cycle(None).await;
     crate::scanner_io::clear_dirty_usage_buckets_for_tests();
 }

crates/scanner/src/scanner_io.rs

@@ -271,6 +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,
     leader_epoch: u64,
     tier_registry_generation: u64,
     /// Epoch captured once for the whole scanner cycle.  `None` is retained

crates/scanner/src/scanner_io/io_cache.rs

@@ -118,6 +118,7 @@ impl ScannerIOCache for SetDisks {
             all_buckets,
             scope,
             digest: scan_plan_digest,
+            bucket_cache_digest,
             leader_epoch,
             tier_registry_generation,
             publication_epoch,
@@ -637,7 +638,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(scan_plan_digest, dirty_usage_buckets_clone.get(&bucket.name).copied());
+                        scanner_bucket_cache_digest(bucket_cache_digest, dirty_usage_buckets_clone.get(&bucket.name).copied());
 
                     if let Some(server_epoch) = remote_server_epoch {
                         let request_sequence = remote_session_sequence;

crates/scanner/src/scanner_io/io_cycle.rs

@@ -263,6 +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 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,
@@ -412,6 +416,7 @@ where
             all_buckets: Arc::clone(&all_buckets),
             scope: scan_scope.clone(),
             digest: scan_plan_digest,
+            bucket_cache_digest,
             leader_epoch,
             tier_registry_generation,
             publication_epoch,

crates/scanner/src/scanner_io/tests.rs

@@ -17,11 +17,12 @@ use super::io_disk::tier_stats_template;
 use super::*;
 use crate::scanner_budget::ScannerCycleBudgetConfig;
 use crate::scanner_folder::ScannerItem;
+use crate::storage_api::EcstoreScannerPeerDirtyUsageSnapshot;
 use crate::storage_api::owner::{
     EcstorePoolDecommissionInfo, EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats,
+    ecstore_hold_namespace_commit,
 };
 use crate::storage_api::scan::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions, ObjectIO as _};
-use crate::storage_api::{EcstoreScannerPeerDirtyUsageSnapshot, ecstore_hold_namespace_commit};
 use crate::{
     DiskOption, ECStore, Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, ScannerObjectOptions,
     ScannerPutObjReader, UNKNOWN_TIER, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests,
@@ -411,12 +412,21 @@ async fn pending_put_commit_keeps_scanner_walk_live_without_authoritative_usage(
     }
 
     let mut pending = Some(ecstore_hold_namespace_commit(store.as_ref()));
+    let mut previous_activity_digest = None;
+    let mut structural_plan_digest = None;
     for (cycle, converged) in [(1, false), (2, true)] {
         if converged {
             drop(pending.take());
         }
         assert_eq!(store.scanner_data_usage_publication_blocked().await, !converged);
         assert!(!store.scanner_data_movement_pause_status().await.paused);
+        let activity = crate::scanner::probe_scanner_activity(store.as_ref(), false)
+            .await
+            .expect("the fixture activity should be observable");
+        let activity_digest = crate::scanner::scanner_activity_snapshot_digest(&activity);
+        if let Some(previous) = previous_activity_digest.replace(activity_digest) {
+            assert_ne!(previous, activity_digest, "draining a namespace commit must change the publication proof");
+        }
         let ctx = CancellationToken::new();
         let budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, ScannerCycleBudgetConfig::default());
         let (updates, mut receiver) = mpsc::channel(1);
@@ -435,6 +445,7 @@ async fn pending_put_commit_keeps_scanner_walk_live_without_authoritative_usage(
         .await
         .expect("namespace scanning must finish while a PUT commit is pending")
         .expect("namespace scanning must remain available during a pending PUT commit");
+        assert_eq!(result.activity_digest(), Some(activity_digest));
         if !converged {
             assert_eq!(budget.progress().0, 2, "the pending commit must not suppress actual object traversal");
         }
@@ -454,6 +465,13 @@ async fn pending_put_commit_keeps_scanner_walk_live_without_authoritative_usage(
         assert_eq!(usage.scanner_cycle, Some(cycle));
         assert_eq!(usage.objects_total_count, 2);
         assert_eq!(usage.objects_total_size, 11);
+        assert_eq!(usage.usage_snapshot_set_states.len(), 2);
+        for state in &usage.usage_snapshot_set_states {
+            let digest = state
+                .scan_plan_digest
+                .expect("each set must retain its structural cache identity");
+            assert_eq!(*structural_plan_digest.get_or_insert(digest), digest);
+        }
         let bucket_usage = usage.buckets_usage.get(&bucket).expect("the walked bucket must be present");
         assert_eq!(bucket_usage.objects_count, 2);
         assert_eq!(bucket_usage.size, 11);

crates/scanner/src/storage_api.rs

@@ -114,8 +114,6 @@ pub(crate) use rustfs_ecstore::api::runtime::{
 };
 pub(crate) use rustfs_ecstore::api::set_disk::SetDisks as EcstoreSetDisks;
 #[cfg(test)]
-pub(crate) use rustfs_ecstore::api::set_disk::test_util::hold_namespace_commit as ecstore_hold_namespace_commit;
-#[cfg(test)]
 pub(crate) use rustfs_ecstore::api::storage::SCANNER_PUBLICATION_LEASE_TTL_MS as ECSTORE_SCANNER_PUBLICATION_LEASE_TTL_MS;
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::storage::init_local_disks_with_instance_ctx as ecstore_init_local_disks_with_instance_ctx;
@@ -129,6 +127,9 @@ pub(crate) use rustfs_lifecycle::{
 use rustfs_storage_api as storage_contracts;
 
 pub(crate) mod owner {
+    #[cfg(test)]
+    pub(crate) use rustfs_ecstore::api::set_disk::test_util::hold_namespace_commit as ecstore_hold_namespace_commit;
+
     pub(crate) use super::storage_contracts::{
         HTTPPreconditions, HTTPRangeSpec, NS_SCANNER_PROTOCOL_VERSION, ObjectIO, ObjectOperations, ObjectToDelete,
     };