diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 19e0fc457..a008dc81e 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/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, diff --git a/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs b/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs new file mode 100644 index 000000000..88a7d91e2 --- /dev/null +++ b/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; +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, 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, 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::>() + } 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(); +}