test(scanner): diagnose raw enumeration across restarts (#7228)

* test(scanner): diagnose raw enumeration across process restarts Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(scanner): observe the canonical synthetic disk path Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(scanner): reject unobserved enumeration budget evidence Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(scanner): remove redundant disk path clone * fix(app): keep list-through header import test-only Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com> Co-authored-by: overtrue <anzhengchao@gmail.com>
2026-09-07 20:46:11 +00:00 · 2026-09-06 10:51:14 +08:00
parent 55fd73ed48
commit fddecf0afe
7 changed files with 411 additions and 0 deletions
@@ -1294,6 +1294,8 @@ impl FolderScanner {
                    }
                    Err(e) => return Err(ScannerError::Io(e)),
                };
+                #[cfg(test)]
+                tests::enumeration_restart::observe_raw_entry(&dir_path, &entry.file_name(), &self.budget);
                pending_entry_progress = pending_entry_progress.saturating_add(1);
                if pending_entry_progress >= SCANNER_ENTRY_PROGRESS_BATCH
                    || last_entry_progress.elapsed() >= SCANNER_ENTRY_PROGRESS_INTERVAL
@@ -25,6 +25,7 @@ use std::os::unix::fs::{PermissionsExt, symlink};
 use std::sync::Mutex;

 mod checkpoint_fixture;
+pub(super) mod enumeration_restart;

 /// Reset the process-global alert cooldown map; test-only.
 fn reset_alert_cooldowns() {
@@ -0,0 +1,181 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use super::*;
+use std::path::{Path, PathBuf};
+use tokio::io::AsyncReadExt;
+
+const MAX_CACHE_BYTES: u64 = 1024 * 1024;
+const REQUEST_ENV: &str = "RUSTFS_ENUMERATION_REQUEST";
+
+struct Observation {
+    root: PathBuf,
+    limit: u64,
+    entries: u64,
+    name_bytes: u64,
+}
+
+static OBSERVATION: Mutex<Option<Observation>> = Mutex::new(None);
+
+// Only the selected synthetic disk is observed; concurrent unrelated scanners
+// do not consume its budget. This hook is absent from non-test builds.
+pub(in crate::scanner_folder) fn observe_raw_entry(dir: &str, name: &std::ffi::OsStr, budget: &ScannerCycleBudget) {
+    let mut guard = OBSERVATION.lock().expect("enumeration observation lock");
+    if let Some(observation) = guard.as_mut()
+        && Path::new(dir).starts_with(&observation.root)
+    {
+        observation.entries += 1;
+        observation.name_bytes += u64::try_from(name.as_encoded_bytes().len()).expect("bounded entry name");
+        if observation.entries >= observation.limit {
+            budget.cancel_for_runtime();
+        }
+    }
+}
+
+struct ObservationGuard;
+
+impl Drop for ObservationGuard {
+    fn drop(&mut self) {
+        *OBSERVATION.lock().expect("enumeration observation cleanup") = None;
+    }
+}
+
+#[derive(serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+struct Request {
+    workspace: PathBuf,
+    objects: usize,
+    raw_entry_budget: u64,
+    round: u32,
+}
+
+async fn read_bounded(path: &Path) -> Vec<u8> {
+    let file = tokio::fs::File::open(path).await.expect("open fixture artifact");
+    let mut bytes = Vec::new();
+    file.take(MAX_CACHE_BYTES + 1)
+        .read_to_end(&mut bytes)
+        .await
+        .expect("read fixture artifact");
+    assert!(u64::try_from(bytes.len()).expect("artifact size") <= MAX_CACHE_BYTES);
+    bytes
+}
+
+async fn round(request: &Request) -> serde_json::Value {
+    assert!((1..=1024).contains(&request.objects));
+    assert!((1..=4096).contains(&request.raw_entry_budget));
+    assert!(request.round < 64);
+    let disk_root = request.workspace.join("disk");
+    let cache_path = request.workspace.join("cache.bin");
+    if request.round == 0 {
+        tokio::fs::create_dir(&disk_root).await.expect("create fresh synthetic disk");
+        for index in 0..request.objects {
+            let object = format!("object-{index:04}");
+            let version = Uuid::from_u128(u128::try_from(index).expect("fixture index") + 1);
+            let bytes = metadata_for_object_version("bucket", &object, Some(version));
+            write_test_object_metadata_bytes(&disk_root, "bucket", &object, &bytes).await;
+        }
+        let mut initial = DataUsageCache::default();
+        initial.info.name = "bucket".to_string();
+        initial.info.skip_healing = true;
+        initial.info.snapshot_complete = false;
+        initial.replace("bucket", "", DataUsageEntry::default());
+        tokio::fs::write(&cache_path, initial.marshal_msg().expect("initial cache codec"))
+            .await
+            .expect("persist initial cache");
+    }
+    let cache = DataUsageCache::unmarshal(&read_bounded(&cache_path).await).expect("reload cache codec before scan");
+    assert_eq!(cache.info.name, "bucket");
+    let before = cache.checked_flatten("bucket").expect("persisted bucket root").objects;
+    let endpoint = Endpoint::try_from(disk_root.to_string_lossy().as_ref()).expect("fixture endpoint");
+    let disk = new_disk(
+        &endpoint,
+        &DiskOption {
+            cleanup: false,
+            health_check: false,
+        },
+    )
+    .await
+    .expect("open synthetic disk in this process");
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(&parent, Default::default());
+    *OBSERVATION.lock().expect("install observation") = Some(Observation {
+        root: disk.path(),
+        limit: request.raw_entry_budget,
+        entries: 0,
+        name_bytes: 0,
+    });
+    let _observation_guard = ObservationGuard;
+    let result = scan_data_folder(
+        budget.token(),
+        budget.clone(),
+        vec![disk.clone()],
+        disk,
+        cache.clone(),
+        None,
+        HealScanMode::Normal,
+        SCANNER_SLEEPER.clone(),
+    )
+    .await;
+    let (returned, outcome) = match result {
+        Ok(cache) => (cache, "complete"),
+        Err(ScannerError::PartialCache(cache)) => (*cache, "partial"),
+        Err(ScannerError::Other(message)) if budget.token().is_cancelled() && message == "Operation cancelled" => {
+            (cache, "cancelled_without_cache")
+        }
+        Err(error) => panic!("unexpected real scanner failure: {error}"),
+    };
+    let encoded = returned.marshal_msg().expect("returned cache codec");
+    assert!(u64::try_from(encoded.len()).expect("encoded length") <= MAX_CACHE_BYTES);
+    tokio::fs::write(&cache_path, encoded).await.expect("persist returned cache");
+    let reloaded = DataUsageCache::unmarshal(&read_bounded(&cache_path).await).expect("reload returned cache codec");
+    let retained = reloaded.checked_flatten("bucket").expect("reloaded bucket root");
+    let scanned = returned.checked_flatten("bucket").expect("returned bucket root");
+    assert_eq!(
+        (retained.objects, retained.versions, retained.size),
+        (scanned.objects, scanned.versions, scanned.size)
+    );
+    assert_eq!(reloaded.info.snapshot_complete, returned.info.snapshot_complete);
+    let guard = OBSERVATION.lock().expect("read observation");
+    let observation = guard.as_ref().expect("installed observation");
+    serde_json::json!({
+        "schema": 1, "pid": std::process::id(), "round": request.round,
+        "objects_expected": request.objects, "raw_entry_budget": request.raw_entry_budget,
+        "raw_entries": observation.entries, "raw_name_bytes": observation.name_bytes,
+        "objects_processed": budget.progress().0,
+        "objects_before": before, "objects_retained": retained.objects,
+        "versions_retained": retained.versions, "bytes_retained": retained.size,
+        "snapshot_complete": reloaded.info.snapshot_complete, "outcome": outcome,
+    })
+}
+
+/// Default CI is a positive healthy control. The external driver selects the
+/// same worker in a fresh OS process per round and applies its strict oracle.
+#[tokio::test]
+#[serial]
+async fn enumeration_restart_worker() {
+    if let Some(path) = std::env::var_os(REQUEST_ENV) {
+        let request: Request = serde_json::from_slice(&read_bounded(Path::new(&path)).await).expect("bounded worker request");
+        let report = round(&request).await;
+        tokio::fs::write(
+            request.workspace.join(format!("round-{}.json", request.round)),
+            serde_json::to_vec(&report).expect("report JSON"),
+        )
+        .await
+        .expect("write worker report");
+    } else {
+        let temp = tempfile::tempdir().expect("healthy fixture directory");
+        let report = round(&Request {
+            workspace: temp.path().to_path_buf(),
+            objects: 4,
+            raw_entry_budget: 16,
+            round: 0,
+        })
+        .await;
+        assert_eq!(report["outcome"], "complete");
+        assert_eq!(report["snapshot_complete"], true);
+        assert_eq!(report["objects_retained"], 4);
+        assert_eq!(report["versions_retained"], 4);
+        assert_eq!(report["bytes_retained"], 4);
+        assert!(report["raw_entries"].as_u64().expect("observed entries") >= 8, "{report}");
+    }
+}