fix(scanner): fence unknown tier accounting (#6396)

crates/scanner/src/remote_scanner/stream.rs

@@ -16,8 +16,8 @@
 use crate::RUSTFS_META_BUCKET;
 use crate::scanner_budget::{ScannerCycleBudget, ScannerCycleBudgetConfig};
 use crate::scanner_io::{
-    DataUsageCacheScanState, ScannerDiskScanOutcome, ScannerIODisk, acquire_scanner_cache_locks, cache_root_entry_info,
-    current_cache_root_or_prepare, scanner_set_disk_inventory,
+    DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerDiskScanOutcome, ScannerIODisk, acquire_scanner_cache_locks,
+    cache_root_entry_info, current_cache_root_or_prepare_with_generation, scanner_set_disk_inventory,
 };
 use crate::storage_api::owner::NS_SCANNER_PROTOCOL_VERSION;
 use crate::{
@@ -213,6 +213,7 @@ pub(crate) struct RemoteScannerScanSpec<'a> {
     pub(crate) session_id: Uuid,
     pub(crate) session_sequence: u64,
     pub(crate) scan_plan_digest: DataUsageScanPlanDigest,
+    pub(crate) tier_registry_generation: u64,
     pub(crate) skip_healing: bool,
     pub(crate) scan_mode: HealScanMode,
 }
@@ -223,6 +224,7 @@ struct RemoteScannerResponseExpectation<'a> {
     source: DataUsageCacheSource,
     next_cycle: u64,
     scan_plan_digest: DataUsageScanPlanDigest,
+    tier_registry_generation: u64,
 }
 
 #[derive(Debug)]
@@ -669,6 +671,11 @@ async fn scan_and_persist_local_bucket(
         scan_mode,
         ..
     } = request;
+    // Keep the worker's cycle snapshot alive through cache reuse, scanning,
+    // and persistence. Without the guard, a later cycle can prune this key
+    // while this request is still running and allow a second registry to be
+    // selected for the same cycle.
+    let _tier_cycle_guard = crate::begin_tier_registry_cycle(next_cycle, leader_epoch);
     let store = resolve_scanner_object_store_handle()
         .ok_or_else(|| RemoteScannerServerError::worker("remote namespace scanner object layer is unavailable"))?;
     validate_remote_scanner_request_fence_with_store(next_cycle, leader_epoch, store.clone())
@@ -704,7 +711,25 @@ async fn scan_and_persist_local_bucket(
     let revisions = cache.load_with_revisions(set.clone(), &cache_name).await.map_err(|err| {
         RemoteScannerServerError::worker(format!("remote namespace scanner cache load or revision lookup failed: {err}"))
     })?;
-    let scan_state = current_cache_root_or_prepare(&mut cache, &bucket, source, next_cycle, leader_epoch, scan_plan_digest, true);
+    // Remote workers use the same cycle-frozen registry as `scan_data_folder`.
+    // Requiring its generation here prevents a cache snapshot classified by an
+    // older registry from being reused before the folder scan gets a chance to
+    // refresh it.
+    let tier_registry_generation = crate::runtime_tier_registry_for_cycle(next_cycle, leader_epoch)
+        .await
+        .generation;
+    let scan_state = current_cache_root_or_prepare_with_generation(
+        &mut cache,
+        &bucket,
+        source,
+        next_cycle,
+        leader_epoch,
+        scan_plan_digest,
+        DataUsageCacheReuseOptions {
+            require_source: true,
+            tier_registry_generation: Some(tier_registry_generation),
+        },
+    );
     match scan_state {
         DataUsageCacheScanState::Current(usage) => {
             if guard.is_lock_lost() {
@@ -869,6 +894,7 @@ pub(crate) async fn scan_remote_bucket(
         session_id,
         session_sequence,
         scan_plan_digest,
+        tier_registry_generation,
         skip_healing,
         scan_mode,
     } = spec;
@@ -957,6 +983,7 @@ pub(crate) async fn scan_remote_bucket(
             source: expected_source,
             next_cycle,
             scan_plan_digest,
+            tier_registry_generation,
         },
         authenticator,
         rpc_deadline,
@@ -1012,6 +1039,7 @@ where
             source: expected_source,
             next_cycle: TEST_NEXT_CYCLE,
             scan_plan_digest: expected_scan_plan_digest,
+            tier_registry_generation: 0,
         },
         authenticator,
         Instant::now() + NS_SCANNER_MAX_RPC_LIFETIME,
@@ -1111,6 +1139,11 @@ where
                         "remote namespace scanner returned usage for a different bucket plan",
                     )));
                 }
+                if complete.usage.tier_registry_generation != Some(expected.tier_registry_generation) {
+                    return Err(RemoteScannerStreamError::reconciled(StorageError::other(
+                        "remote namespace scanner returned usage for a different tier registry generation",
+                    )));
+                }
                 if !complete.usage.entry.children.is_empty() {
                     return Err(RemoteScannerStreamError::reconciled(StorageError::other(
                         "remote namespace scanner returned non-flattened bucket usage",

crates/scanner/src/remote_scanner/stream/tests.rs

@@ -195,6 +195,7 @@ fn test_usage(bucket: &str, objects: usize) -> DataUsageEntryInfo {
         name: bucket.to_string(),
         parent: crate::DATA_USAGE_ROOT.to_string(),
         entry,
+        tier_registry_generation: Some(0),
     }
 }
 
@@ -751,6 +752,43 @@ async fn complete_terminal_frame_reconciles_progress_and_usage() {
     assert_eq!(budget.progress(), (3, 2));
 }
 
+#[tokio::test]
+async fn terminal_usage_from_a_different_tier_generation_is_rejected() {
+    let request_id = Uuid::new_v4();
+    let writer_auth = FrameAuthenticator::for_test(request_id);
+    let reader_auth = FrameAuthenticator::for_test(request_id);
+    let (mut writer, reader) = tokio::io::duplex(4096);
+    tokio::spawn(async move {
+        let mut usage = test_usage("bucket", 1);
+        usage.tier_registry_generation = Some(1);
+        let mut sequence = 0;
+        write_frame(
+            &mut writer,
+            &writer_auth,
+            &mut sequence,
+            &RemoteScannerFrame::terminal(
+                RemoteScannerProgress::default(),
+                RemoteScannerFrameResult::Complete(Box::new(RemoteScannerComplete {
+                    source: TEST_SOURCE,
+                    scan_plan_digest: TEST_PLAN_DIGEST,
+                    usage,
+                    pending_maintenance_work: false,
+                })),
+            ),
+        )
+        .await
+        .expect("terminal frame should write");
+    });
+
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new(&parent, ScannerCycleBudgetConfig::default());
+    let error = consume_remote_scanner_stream(reader, parent, budget, "bucket", TEST_SOURCE, TEST_PLAN_DIGEST, reader_auth)
+        .await
+        .expect_err("generation mismatch must fail closed");
+
+    assert!(error.to_string().contains("tier registry generation"));
+}
+
 #[tokio::test]
 async fn complete_terminal_frame_after_budget_expiry_is_partial() {
     let request_id = Uuid::new_v4();