chore: merge main into scanner scope entry oracles

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/scanner/src/scanner_io/tests.rs

@@ -39,6 +39,8 @@ use temp_env::with_var;
 use time::OffsetDateTime;
 use uuid::Uuid;
 
+mod scoped_entry_fallback;
+
 #[derive(Clone)]
 struct FixedWorkloadProvider {
     snapshot: WorkloadAdmissionRegistrySnapshot,

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

@@ -0,0 +1,289 @@
+// Copyright 2026 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::*;
+use crate::data_usage_define::{DATA_USAGE_OBJ_NAME_PATH, read_config_with_revision};
+use crate::storage_api::owner::EcstoreDiskAPI;
+
+type DriveIdentities = HashMap<String, (Uuid, DataUsageCacheSource)>;
+type WalkCounts = HashMap<(String, String, String), u64>;
+
+async fn drive_identities(store: &ECStore) -> DriveIdentities {
+    let mut identities = HashMap::new();
+    let mut ids = HashSet::new();
+    for set in store.all_set_disks() {
+        let source = DataUsageCacheSource::new(set.pool_index, set.set_index);
+        for disk in scanner_set_disk_inventory(set.as_ref()).await {
+            let id = EcstoreDiskAPI::get_disk_id(disk.as_ref())
+                .await
+                .expect("fixture disk identity should be readable")
+                .expect("fixture disk must have a durable identity");
+            assert!(!id.is_nil());
+            assert!(ids.insert(id), "fixture disk identities must be unique");
+            let path = crate::ScannerDiskExt::path(disk.as_ref()).to_string_lossy().into_owned();
+            assert!(identities.insert(path, (id, source)).is_none());
+        }
+    }
+    assert_eq!(identities.len(), 8);
+    identities
+}
+
+fn walk_counts(drives: &DriveIdentities) -> WalkCounts {
+    rustfs_scanner_metrics::metrics::global_metrics()
+        .scanner_runtime_details_report()
+        .bucket_drive_results
+        .into_iter()
+        .filter(|result| drives.contains_key(&result.drive))
+        .map(|result| ((result.bucket, result.drive, result.result), result.count))
+        .collect()
+}
+
+async fn put_and_settle(store: &ECStore, bucket: &str, object: &str) {
+    let set = &store.pools[0].disk_set[0];
+    let mut reader = ScannerPutObjReader::from_vec(b"object".to_vec());
+    set.put_object(bucket, object, &mut reader, &ScannerObjectOptions::default())
+        .await
+        .expect("fixture object should persist");
+    let lock = set.new_ns_lock(bucket, object).await.expect("fixture namespace lock");
+    let _settled = lock
+        .get_write_lock(Duration::from_secs(30))
+        .await
+        .expect("quorum-ACK rename tail must settle before taking the activity baseline");
+}
+
+async fn create_bucket(store: &ECStore, bucket: &str) {
+    store
+        .make_bucket(bucket, &MakeBucketOptions::default())
+        .await
+        .expect("fixture bucket should be created");
+    put_and_settle(store, bucket, "initial").await;
+}
+
+async fn persist_baseline(store: &Arc<ECStore>, baseline: &DataUsageInfo) {
+    let mut baseline = baseline.clone();
+    baseline.usage_snapshot_converged = Some(true);
+    crate::save_config(
+        store.clone(),
+        DATA_USAGE_OBJ_NAME_PATH.as_str(),
+        serde_json::to_vec(&baseline).expect("baseline should encode"),
+    )
+    .await
+    .expect("fixture baseline should persist");
+}
+
+// Every invocation uses the production default scope. The expected walker set
+// comes from storage's per-source inventory, not the resolver's selected names.
+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
+        .list_bucket_for_scanner(&BucketOptions::default())
+        .await
+        .expect("fixture inventory should be complete");
+    assert!(inventory.topology_complete);
+    let expected_walks = if expect_walks {
+        inventory
+            .set_buckets
+            .into_iter()
+            .flat_map(|set| {
+                let source = DataUsageCacheSource::new(set.pool_index, set.set_index);
+                set.buckets.into_iter().map(move |bucket| ((source, bucket.name), 1_u64))
+            })
+            .collect::<HashMap<_, _>>()
+    } else {
+        HashMap::new()
+    };
+    let root_before = read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+        .await
+        .expect("root baseline should be readable");
+    let dirty_before = dirty_usage_buckets_for_tests();
+    let generation_before = dirty_usage_generation();
+    let before = walk_counts(&drives);
+    let ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
+    let (updates, mut receiver) = mpsc::channel(1);
+    let (observer, observed) = tokio::sync::oneshot::channel();
+    let result = tokio::time::timeout(
+        Duration::from_secs(30),
+        nsscanner_with_storage_status_scoped(
+            store.as_ref(),
+            ScannerCycleRequest {
+                ctx,
+                budget,
+                updates,
+                want_cycle: cycle,
+                leader_epoch: 11,
+                scan_mode: HealScanMode::Normal,
+                scan_scope: ScannerBucketScanScope::default(),
+                persisted_usage_baseline: root_before.0.clone().map(Bytes::from),
+                requires_full_scan: false,
+                resolved_scope_observer: Some(observer),
+            },
+        ),
+    )
+    .await
+    .expect("entry cycle should finish within the fixture deadline")
+    .expect("entry cycle should succeed");
+    assert_eq!(result.status, ScannerCycleStatus::Complete);
+    let scope = observed.await.expect("production resolver should report its decision");
+    assert_eq!(
+        scope.selected_buckets.as_deref(),
+        selected.map(|name| HashSet::from([name.to_string()])).as_ref()
+    );
+    let usage = receiver.recv().await.expect("one candidate should be delivered");
+    assert!(receiver.recv().await.is_none(), "there must be exactly one terminal candidate");
+    assert!(usage.usage_snapshot_complete);
+    assert!(!usage.usage_snapshot_partial);
+    assert_eq!(usage.scanner_cycle, Some(cycle));
+    assert_eq!(
+        drive_identities(store).await,
+        drives,
+        "drive identities must not change during the oracle"
+    );
+
+    let after = walk_counts(&drives);
+    let mut actual = HashMap::new();
+    for key in before.keys() {
+        assert!(after.contains_key(key), "metrics eviction would invalidate this exact-delta oracle");
+    }
+    for ((bucket, drive, outcome), count) in after {
+        let previous = before
+            .get(&(bucket.clone(), drive.clone(), outcome.clone()))
+            .copied()
+            .unwrap_or(0);
+        let delta = count.checked_sub(previous).expect("fixture counters must not reset");
+        if delta > 0 {
+            assert_eq!(outcome, "success", "no error or partial walker is expected");
+            *actual.entry((drives[&drive].1, bucket)).or_insert(0_u64) += delta;
+        }
+    }
+    assert_eq!(
+        actual, expected_walks,
+        "each listed source/bucket must have exactly the expected real walks"
+    );
+    assert_eq!(
+        read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+            .await
+            .expect("root after scan"),
+        root_before,
+        "producing a candidate must not replace the coordinator-owned root baseline"
+    );
+    assert_eq!(dirty_usage_generation(), generation_before);
+    assert!(
+        dirty_usage_buckets_for_tests() == dirty_before,
+        "candidate delivery must not ACK pending dirty buckets"
+    );
+    usage
+}
+
+#[tokio::test]
+#[serial]
+async fn scoped_entry_fallback_distinguishes_planned_scope_from_real_cold_walks() {
+    let (_dir, store) = setup_two_pool_scanner_store().await;
+    clear_dirty_usage_buckets_for_tests();
+    let hot = format!("hot-{}", Uuid::new_v4().simple());
+    let cold = format!("cold-{}", Uuid::new_v4().simple());
+    create_bucket(&store, &hot).await;
+    create_bucket(&store, &cold).await;
+    record_dirty_usage_bucket(&hot);
+    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
+    // later cycle may reuse unselected buckets without durable incarnation.
+    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);
+    assert_eq!(usage.buckets_usage[&cold].objects_count, 1);
+    assert_eq!(usage.objects_total_count, 2);
+    clear_dirty_usage_buckets_for_tests();
+}
+
+#[tokio::test]
+#[serial]
+async fn scoped_entry_fallback_rejects_invalid_persisted_baseline_at_the_walker() {
+    let (_dir, store) = setup_two_pool_scanner_store().await;
+    clear_dirty_usage_buckets_for_tests();
+    let hot = format!("hot-{}", Uuid::new_v4().simple());
+    let cold = format!("cold-{}", Uuid::new_v4().simple());
+    create_bucket(&store, &hot).await;
+    create_bucket(&store, &cold).await;
+    record_dirty_usage_bucket(&hot);
+    // The first real scan is also the missing persisted-baseline case.
+    let baseline = run_entry(&store, 1, None, true).await;
+    for (index, kind) in [
+        "malformed",
+        "unconverged",
+        "missing-set",
+        "wrong-source",
+        "mixed-plan",
+        "wrong-epoch",
+    ]
+    .into_iter()
+    .enumerate()
+    {
+        let mut candidate = baseline.clone();
+        candidate.usage_snapshot_converged = Some(true);
+        match kind {
+            "unconverged" => candidate.usage_snapshot_converged = Some(false),
+            "missing-set" => {
+                candidate.usage_snapshot_set_states.pop();
+            }
+            "wrong-source" => candidate.usage_snapshot_set_states[0].set_index = 99,
+            "mixed-plan" => candidate.usage_snapshot_set_states[1].scan_plan_digest = Some([0xA5; 32]),
+            "wrong-epoch" => candidate.usage_snapshot_set_states[0].scanner_epoch = Some(10),
+            "malformed" => {}
+            _ => unreachable!(),
+        }
+        let bytes = if kind == "malformed" {
+            b"{broken".to_vec()
+        } else {
+            serde_json::to_vec(&candidate).expect("candidate JSON")
+        };
+        crate::save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), bytes)
+            .await
+            .expect("negative baseline should persist");
+        let usage = run_entry(&store, u64::try_from(index).expect("fixture cycle index should fit") + 2, None, true).await;
+        assert_eq!(usage.objects_total_count, 2, "{kind}");
+        assert_eq!(usage.buckets_usage[&cold].objects_count, 1, "{kind}");
+    }
+    clear_dirty_usage_buckets_for_tests();
+}
+
+#[tokio::test]
+#[serial]
+async fn scoped_entry_fallback_covers_overflow_and_new_bucket_inventory() {
+    let (_dir, store) = setup_two_pool_scanner_store().await;
+    clear_dirty_usage_buckets_for_tests();
+    let hot = format!("hot-{}", Uuid::new_v4().simple());
+    create_bucket(&store, &hot).await;
+    record_dirty_usage_bucket(&hot);
+    let baseline = run_entry(&store, 1, None, true).await;
+    persist_baseline(&store, &baseline).await;
+    for index in 0..=crate::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES {
+        record_dirty_usage_bucket(&format!("overflow-{index}"));
+    }
+    assert!(dirty_usage_buckets_for_tests().len() > crate::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES);
+    let usage = run_entry(&store, 2, None, true).await;
+    assert_eq!(usage.objects_total_count, 1);
+
+    clear_dirty_usage_buckets_for_tests();
+    record_dirty_usage_bucket(&hot);
+    let new_bucket = format!("new-{}", Uuid::new_v4().simple());
+    create_bucket(&store, &new_bucket).await;
+    // Even a previously valid baseline cannot cover the changed inventory.
+    let usage = run_entry(&store, 3, None, true).await;
+    assert_eq!(usage.objects_total_count, 2);
+    assert_eq!(usage.buckets_usage[&new_bucket].objects_count, 1);
+    clear_dirty_usage_buckets_for_tests();
+}