fix(scanner): bind scoped set reuse to bucket incarnations (#7395)

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/scanner/src/data_usage_define.rs

@@ -536,6 +536,10 @@ pub struct DataUsageEntryInfo {
     pub name: String,
     pub parent: String,
     pub entry: DataUsageEntry,
+    /// Durable bucket incarnation that produced this bucket root. Missing
+    /// values are legacy/unproven and must not authorize cold-bucket reuse.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub bucket_incarnation: Option<uuid::Uuid>,
     /// Registry generation used to classify this root entry. Older remote
     /// workers omit it; callers must reject that result when a frozen cycle
     /// requires generation fencing.
@@ -653,6 +657,11 @@ pub struct DataUsageCacheInfo {
     /// structural plan remains reusable across ordinary bucket writes.
     #[serde(default)]
     pub scan_execution_digest: Option<DataUsageScanPlanDigest>,
+    /// Durable bucket incarnations captured for a complete set aggregate.
+    /// Missing or nil entries are legacy/unproven and cannot authorize
+    /// skipping an unselected bucket in a later scoped set scan.
+    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
+    pub scan_bucket_incarnations: HashMap<String, uuid::Uuid>,
 }
 
 impl Serialize for DataUsageCacheInfo {
@@ -676,7 +685,8 @@ impl Serialize for DataUsageCacheInfo {
             + 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.scan_execution_digest.is_some());
+            + usize::from(self.scan_execution_digest.is_some())
+            + usize::from(!self.scan_bucket_incarnations.is_empty());
         let mut state = serializer.serialize_map(Some(field_count))?;
         state.serialize_entry("name", &self.name)?;
         state.serialize_entry("next_cycle", &self.next_cycle)?;
@@ -736,6 +746,9 @@ impl Serialize for DataUsageCacheInfo {
         if let Some(scan_execution_digest) = self.scan_execution_digest {
             state.serialize_entry("scan_execution_digest", &scan_execution_digest)?;
         }
+        if !self.scan_bucket_incarnations.is_empty() {
+            state.serialize_entry("scan_bucket_incarnations", &self.scan_bucket_incarnations)?;
+        }
         state.end()
     }
 }

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,
+        bucket_incarnation: Some(Uuid::from_u128(7)),
         tier_registry_generation: Some(0),
     }
 }

crates/scanner/src/scanner_io.rs

@@ -741,6 +741,11 @@ pub(crate) fn cache_root_entry_info(cache: &DataUsageCache) -> std::result::Resu
         name: cache.info.name.clone(),
         parent: DATA_USAGE_ROOT.to_string(),
         entry,
+        bucket_incarnation: cache
+            .info
+            .scan_identity
+            .map(|identity| identity.bucket_incarnation)
+            .filter(|incarnation| !incarnation.is_nil()),
         tier_registry_generation: cache.info.tier_registry_generation,
     })
 }
@@ -752,6 +757,14 @@ fn apply_bucket_result_to_cache(cache: &mut DataUsageCache, result: DataUsageEnt
         // forces the caller to re-account it under one frozen registry.
         return false;
     }
+    match result.bucket_incarnation {
+        Some(incarnation) if !incarnation.is_nil() => {
+            cache.info.scan_bucket_incarnations.insert(result.name.clone(), incarnation);
+        }
+        _ => {
+            cache.info.scan_bucket_incarnations.remove(&result.name);
+        }
+    }
     cache.replace(&result.name, &result.parent, result.entry);
     cache.info.last_update = Some(update_time);
     true

crates/scanner/src/scanner_io/io_cache.rs

@@ -34,17 +34,12 @@ pub(super) fn prepare_scoped_set_scan(
     all_buckets: &[BucketInfo],
     scope: &ScannerBucketScanScope,
     generation: ScannerSetCacheGeneration,
+    current_bucket_incarnations: Option<&HashMap<String, uuid::Uuid>>,
 ) -> Option<PreparedScopedSetScan> {
     let (Some(selected_buckets), Some(baseline_scan_plan_digest)) = (&scope.selected_buckets, scope.baseline_scan_plan_digest)
     else {
         return None;
     };
-    // The existing cache does not bind each bucket to a durable incarnation.
-    // Listing creation times can come from volume metadata, so even Some(time)
-    // cannot prove that an unselected same-name bucket is the cached bucket.
-    if all_buckets.iter().any(|bucket| !selected_buckets.contains(&bucket.name)) {
-        return None;
-    }
     if selected_buckets.is_empty()
         || !old_cache.info.snapshot_complete
         || old_cache.info.last_update.is_none()
@@ -56,6 +51,7 @@ pub(super) fn prepare_scoped_set_scan(
         || old_cache.info.scan_plan_digest != Some(baseline_scan_plan_digest)
         || old_cache.info.cache_key_format != DATA_USAGE_CACHE_KEY_FORMAT
         || !old_cache.has_complete_root_inventory(&old_cache.find(DATA_USAGE_ROOT)?.children)
+        || !unselected_bucket_incarnations_match(old_cache, all_buckets, selected_buckets, current_bucket_incarnations)
     {
         return None;
     }
@@ -75,6 +71,7 @@ pub(super) fn prepare_scoped_set_scan(
             lkg_last_update: old_cache.info.last_update,
             lkg_leader_epoch: Some(old_cache.info.leader_epoch),
             lkg_scan_plan_digest: old_cache.info.scan_plan_digest,
+            scan_bucket_incarnations: old_cache.info.scan_bucket_incarnations.clone(),
             ..Default::default()
         },
         cache: HashMap::new(),
@@ -85,7 +82,15 @@ pub(super) fn prepare_scoped_set_scan(
         if !current_bucket_names.insert(bucket.name.as_str()) {
             return None;
         }
-        cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
+        if selected_buckets.contains(&bucket.name) {
+            cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
+        } else {
+            cache.copy_with_children(
+                old_cache,
+                &rustfs_data_usage::hash_path(&bucket.name),
+                &Some(rustfs_data_usage::hash_path(DATA_USAGE_ROOT)),
+            );
+        }
     }
 
     Some(PreparedScopedSetScan {
@@ -98,6 +103,47 @@ pub(super) fn prepare_scoped_set_scan(
     })
 }
 
+fn unselected_bucket_incarnations_match(
+    old_cache: &DataUsageCache,
+    all_buckets: &[BucketInfo],
+    selected_buckets: &HashSet<String>,
+    current_bucket_incarnations: Option<&HashMap<String, uuid::Uuid>>,
+) -> bool {
+    let Some(current_bucket_incarnations) = current_bucket_incarnations else {
+        return all_buckets.iter().all(|bucket| selected_buckets.contains(&bucket.name));
+    };
+    all_buckets
+        .iter()
+        .filter(|bucket| !selected_buckets.contains(&bucket.name))
+        .all(|bucket| {
+            let Some(current) = current_bucket_incarnations
+                .get(&bucket.name)
+                .filter(|incarnation| !incarnation.is_nil())
+            else {
+                return false;
+            };
+            old_cache
+                .info
+                .scan_bucket_incarnations
+                .get(&bucket.name)
+                .filter(|cached| !cached.is_nil())
+                == Some(current)
+        })
+}
+
+async fn scanner_current_bucket_incarnations(set: &SetDisks, all_buckets: &[BucketInfo]) -> Option<HashMap<String, uuid::Uuid>> {
+    let mut incarnations = HashMap::with_capacity(all_buckets.len());
+    for bucket in all_buckets {
+        let Ok(incarnation) = set.bucket_incarnation_id_from_disk(&bucket.name).await else {
+            return None;
+        };
+        if incarnation.is_nil() || incarnations.insert(bucket.name.clone(), incarnation).is_some() {
+            return None;
+        }
+    }
+    Some(incarnations)
+}
+
 #[async_trait::async_trait]
 impl ScannerIOCache for SetDisks {
     #[tracing::instrument(skip(self, budget, scan_plan, updates))]
@@ -158,6 +204,7 @@ impl ScannerIOCache for SetDisks {
                 None
             }
         };
+        let current_bucket_incarnations = scanner_current_bucket_incarnations(self.as_ref(), &all_buckets).await;
         let scoped_scan = prepare_scoped_set_scan(
             &old_cache,
             &buckets,
@@ -170,6 +217,7 @@ impl ScannerIOCache for SetDisks {
                 source,
                 scan_plan_digest,
             },
+            current_bucket_incarnations.as_ref(),
         );
         let mut scoped_cache = scoped_scan.map(|mut prepared| {
             buckets = prepared.buckets;
@@ -191,6 +239,7 @@ impl ScannerIOCache for SetDisks {
                             scan_plan_digest: Some(scan_plan_digest),
                             scan_coverage_digest: Some(bucket_coverage_digest),
                             cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
+                            scan_bucket_incarnations: current_bucket_incarnations.clone().unwrap_or_default(),
                             ..Default::default()
                         },
                         cache: HashMap::new(),
@@ -486,6 +535,7 @@ impl ScannerIOCache for SetDisks {
                     lkg_last_update: old_cache.info.lkg_last_update,
                     lkg_leader_epoch: old_cache.info.lkg_leader_epoch,
                     lkg_scan_plan_digest: old_cache.info.lkg_scan_plan_digest,
+                    scan_bucket_incarnations: current_bucket_incarnations.clone().unwrap_or_default(),
                     ..Default::default()
                 },
                 cache: HashMap::new(),

crates/scanner/src/scanner_io/tests.rs

@@ -1232,6 +1232,19 @@ fn complete_set_usage_cache(buckets: &[(&str, usize)], scan_plan_digest: DataUsa
     cache
 }
 
+fn test_bucket_incarnations(buckets: &[&str]) -> HashMap<String, Uuid> {
+    buckets
+        .iter()
+        .enumerate()
+        .map(|(index, bucket)| {
+            (
+                (*bucket).to_string(),
+                Uuid::from_u128(u128::try_from(index).expect("test index should fit") + 1),
+            )
+        })
+        .collect()
+}
+
 #[tokio::test]
 #[serial]
 async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers() {
@@ -2082,6 +2095,7 @@ fn scoped_set_scan_rebuilds_selected_buckets_and_drops_deleted_buckets() {
             source: DataUsageCacheSource::new(1, 2),
             scan_plan_digest: current_digest,
         },
+        None,
     )
     .expect("complete matching set cache should support a scoped scan");
 
@@ -2105,6 +2119,57 @@ fn scoped_set_scan_rebuilds_selected_buckets_and_drops_deleted_buckets() {
     assert_eq!(prepared.cache.info.lkg_scan_plan_digest, Some(baseline_digest));
 }
 
+#[test]
+fn scoped_set_scan_reuses_unselected_buckets_with_matching_incarnations() {
+    let baseline_digest = DataUsageScanPlanDigest([1; 32]);
+    let current_digest = DataUsageScanPlanDigest([2; 32]);
+    let mut old_cache = complete_set_usage_cache(&[("stable", 10), ("dirty", 20)], baseline_digest);
+    old_cache.replace(
+        "stable/prefix",
+        "stable",
+        DataUsageEntry {
+            size: 5,
+            objects: 1,
+            ..Default::default()
+        },
+    );
+    old_cache.info.scan_bucket_incarnations = test_bucket_incarnations(&["stable", "dirty"]);
+    let current_incarnations = old_cache.info.scan_bucket_incarnations.clone();
+    let all_buckets = vec![
+        bucket_info_with_created_time("stable"),
+        bucket_info_with_created_time("dirty"),
+    ];
+
+    let prepared = prepare_scoped_set_scan(
+        &old_cache,
+        &all_buckets,
+        &all_buckets,
+        &ScannerBucketScanScope {
+            selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))),
+            selected_bucket_prefixes: None,
+            baseline_scan_plan_digest: Some(baseline_digest),
+        },
+        ScannerSetCacheGeneration {
+            want_cycle: 8,
+            leader_epoch: 11,
+            tier_registry_generation: 13,
+            source: DataUsageCacheSource::new(1, 2),
+            scan_plan_digest: current_digest,
+        },
+        Some(&current_incarnations),
+    )
+    .expect("matching bucket incarnations should authorize cold bucket reuse");
+
+    assert_eq!(prepared.buckets.iter().map(|bucket| bucket.name.as_str()).collect::<Vec<_>>(), ["dirty"]);
+    let stable = prepared
+        .cache
+        .checked_flatten("stable")
+        .expect("unselected stable bucket should be copied with children");
+    assert_eq!((stable.size, stable.objects), (15, 2));
+    assert_eq!(prepared.cache.find("dirty").map(|entry| (entry.size, entry.objects)), Some((0, 0)));
+    assert_eq!(prepared.cache.info.scan_bucket_incarnations, current_incarnations);
+}
+
 #[test]
 fn scoped_set_scan_rejects_unbound_bucket_incarnations() {
     let baseline_digest = DataUsageScanPlanDigest([1; 32]);
@@ -2130,10 +2195,22 @@ fn scoped_set_scan_rejects_unbound_bucket_incarnations() {
         stable.created = created;
         let buckets = vec![stable, bucket_info_with_created_time("dirty")];
         assert!(
-            prepare_scoped_set_scan(&old_cache, &buckets, &buckets, &scope, generation).is_none(),
+            prepare_scoped_set_scan(&old_cache, &buckets, &buckets, &scope, generation, None).is_none(),
             "missing identity, volume timestamps and same-name recreation must all rebuild"
         );
     }
+    let mut mismatched = test_bucket_incarnations(&["stable", "dirty"]);
+    mismatched.insert("stable".to_string(), Uuid::from_u128(99));
+    let mut old_cache = old_cache;
+    old_cache.info.scan_bucket_incarnations = test_bucket_incarnations(&["stable", "dirty"]);
+    let buckets = vec![
+        bucket_info_with_created_time("stable"),
+        bucket_info_with_created_time("dirty"),
+    ];
+    assert!(
+        prepare_scoped_set_scan(&old_cache, &buckets, &buckets, &scope, generation, Some(&mismatched)).is_none(),
+        "a same-name unselected bucket with a different incarnation must rebuild"
+    );
 }
 
 #[test]
@@ -2159,6 +2236,7 @@ fn scoped_set_scan_falls_back_when_an_unselected_bucket_has_no_baseline() {
                 source: DataUsageCacheSource::new(1, 2),
                 scan_plan_digest: DataUsageScanPlanDigest([4; 32]),
             },
+            Some(&test_bucket_incarnations(&["stable", "new"])),
         )
         .is_none()
     );
@@ -2183,19 +2261,19 @@ fn scoped_set_scan_requires_an_exact_complete_baseline() {
 
     let mut incomplete = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
     incomplete.info.snapshot_complete = false;
-    assert!(prepare_scoped_set_scan(&incomplete, &all_buckets, &all_buckets, &scope, generation).is_none());
+    assert!(prepare_scoped_set_scan(&incomplete, &all_buckets, &all_buckets, &scope, generation, None).is_none());
 
     let mut not_durable = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
     not_durable.info.last_update = None;
-    assert!(prepare_scoped_set_scan(&not_durable, &all_buckets, &all_buckets, &scope, generation).is_none());
+    assert!(prepare_scoped_set_scan(&not_durable, &all_buckets, &all_buckets, &scope, generation, None).is_none());
 
     let mut unscoped_usage = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
     unscoped_usage.cache.get_mut(DATA_USAGE_ROOT).expect("set root").objects = 1;
-    assert!(prepare_scoped_set_scan(&unscoped_usage, &all_buckets, &all_buckets, &scope, generation).is_none());
+    assert!(prepare_scoped_set_scan(&unscoped_usage, &all_buckets, &all_buckets, &scope, generation, None).is_none());
 
     let mut wrong_digest = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
     wrong_digest.info.scan_plan_digest = Some(DataUsageScanPlanDigest([7; 32]));
-    assert!(prepare_scoped_set_scan(&wrong_digest, &all_buckets, &all_buckets, &scope, generation).is_none());
+    assert!(prepare_scoped_set_scan(&wrong_digest, &all_buckets, &all_buckets, &scope, generation, None).is_none());
 
     let empty_scope = ScannerBucketScanScope {
         selected_buckets: Some(Arc::new(HashSet::new())),
@@ -2203,18 +2281,18 @@ fn scoped_set_scan_requires_an_exact_complete_baseline() {
         baseline_scan_plan_digest: Some(baseline_digest),
     };
     let complete = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
-    assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &empty_scope, generation).is_none());
+    assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &empty_scope, generation, None).is_none());
-    assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &scope, generation).is_some());
+    assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &scope, generation, None).is_some());
 
     let unidentified_buckets = vec![bucket_info("dirty")];
     assert!(
-        prepare_scoped_set_scan(&complete, &unidentified_buckets, &unidentified_buckets, &scope, generation).is_some(),
+        prepare_scoped_set_scan(&complete, &unidentified_buckets, &unidentified_buckets, &scope, generation, None).is_some(),
         "fully selected buckets are rebuilt without reusing an unproven incarnation"
     );
 
     let mut future_cache = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
     future_cache.info.next_cycle = generation.want_cycle.saturating_add(1);
-    assert!(prepare_scoped_set_scan(&future_cache, &all_buckets, &all_buckets, &scope, generation).is_none());
+    assert!(prepare_scoped_set_scan(&future_cache, &all_buckets, &all_buckets, &scope, generation, None).is_none());
 }
 
 #[test]
@@ -3083,6 +3161,7 @@ fn apply_bucket_result_to_cache_updates_bucket_entry() {
                 objects: 2,
                 ..Default::default()
             },
+            bucket_incarnation: Some(Uuid::from_u128(7)),
             tier_registry_generation: None,
         },
         update_time,
@@ -3092,6 +3171,7 @@ fn apply_bucket_result_to_cache_updates_bucket_entry() {
     let entry = cache.find("bucket").expect("bucket entry should remain present");
     assert_eq!(entry.size, 10);
     assert_eq!(entry.objects, 2);
+    assert_eq!(cache.info.scan_bucket_incarnations.get("bucket"), Some(&Uuid::from_u128(7)));
 }
 
 #[test]
@@ -3122,6 +3202,7 @@ fn apply_bucket_result_to_cache_rejects_a_different_tier_generation() {
                 size: 11,
                 ..Default::default()
             },
+            bucket_incarnation: Some(Uuid::from_u128(7)),
             tier_registry_generation: Some(8),
         },
         SystemTime::now(),

crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs

@@ -82,8 +82,8 @@ async fn persist_baseline(store: &Arc<ECStore>, baseline: &DataUsageInfo) {
     .expect("fixture baseline should persist");
 }
 
-// Every invocation uses the production default scope. The expected walker set
+// Every invocation uses the production default scope. Once durable bucket
-// comes from storage's per-source inventory, not the resolver's selected names.
+// incarnations are present, the expected walker set follows the resolved scope.
 async fn run_entry(store: &Arc<ECStore>, cycle: u64, selected: Option<&str>, expect_walks: bool) -> DataUsageInfo {
     let drives = drive_identities(store).await;
     let inventory = store
@@ -99,6 +99,7 @@ async fn run_entry(store: &Arc<ECStore>, cycle: u64, selected: Option<&str>, exp
                 let source = DataUsageCacheSource::new(set.pool_index, set.set_index);
                 set.buckets.into_iter().map(move |bucket| ((source, bucket.name), 1_u64))
             })
+            .filter(|((_, bucket), _)| selected.is_none_or(|selected| bucket == selected))
             .collect::<HashMap<_, _>>()
     } else {
         HashMap::new()
@@ -201,8 +202,8 @@ async fn scoped_entry_fallback_distinguishes_planned_scope_from_real_cold_walks(
     let baseline = run_entry(&store, 1, None, true).await;
     persist_baseline(&store, &baseline).await;
 
-    // A same-intent, same-cycle Current cache is a retry, not proof that a
+    // Same-cycle Current remains a retry. The later cycle may skip the cold
-    // later cycle may reuse unselected buckets without durable incarnation.
+    // bucket only after the prior complete set cache has durable incarnations.
     run_entry(&store, 1, Some(&hot), false).await;
     let usage = run_entry(&store, 2, Some(&hot), true).await;
     assert_eq!(usage.buckets_usage[&hot].objects_count, 1);