From 7da2fe02f01213e2531f8ee9dcd397992ccb11ec Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 10:58:49 +0800 Subject: [PATCH] test(scanner): classify raw enumeration replay Record bounded first/last raw entry markers in the restart diagnostic worker and have the driver classify repeated raw windows when retained coverage does not advance. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- .../tests/enumeration_restart.rs | 11 ++++++++ .../diagnose_scanner_enumeration_restart.py | 17 +++++++++++- ...st_diagnose_scanner_enumeration_restart.py | 26 ++++++++++++++++++- 3 files changed, 52 insertions(+), 2 deletions(-) diff --git a/crates/scanner/src/scanner_folder/tests/enumeration_restart.rs b/crates/scanner/src/scanner_folder/tests/enumeration_restart.rs index 90a1b30eb..ccb37882b 100644 --- a/crates/scanner/src/scanner_folder/tests/enumeration_restart.rs +++ b/crates/scanner/src/scanner_folder/tests/enumeration_restart.rs @@ -13,6 +13,8 @@ struct Observation { limit: u64, entries: u64, name_bytes: u64, + first_entry: Option, + last_entry: Option, } static OBSERVATION: Mutex> = Mutex::new(None); @@ -24,6 +26,12 @@ pub(in crate::scanner_folder) fn observe_raw_entry(dir: &str, name: &std::ffi::O if let Some(observation) = guard.as_mut() && Path::new(dir).starts_with(&observation.root) { + let relative_dir = Path::new(dir) + .strip_prefix(&observation.root) + .unwrap_or_else(|_| Path::new("")); + let entry_marker = relative_dir.join(name).to_string_lossy().to_string(); + observation.first_entry.get_or_insert_with(|| entry_marker.clone()); + observation.last_entry = Some(entry_marker); observation.entries += 1; observation.name_bytes += u64::try_from(name.as_encoded_bytes().len()).expect("bounded entry name"); if observation.entries >= observation.limit { @@ -103,6 +111,8 @@ async fn round(request: &Request) -> serde_json::Value { limit: request.raw_entry_budget, entries: 0, name_bytes: 0, + first_entry: None, + last_entry: None, }); let _observation_guard = ObservationGuard; let result = scan_data_folder( @@ -141,6 +151,7 @@ async fn round(request: &Request) -> serde_json::Value { "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, + "raw_first_entry": observation.first_entry, "raw_last_entry": observation.last_entry, "objects_processed": budget.progress().0, "objects_before": before, "objects_retained": retained.objects, "versions_retained": retained.versions, "bytes_retained": retained.size, diff --git a/scripts/diagnose_scanner_enumeration_restart.py b/scripts/diagnose_scanner_enumeration_restart.py index 951c164a7..88f39a485 100644 --- a/scripts/diagnose_scanner_enumeration_restart.py +++ b/scripts/diagnose_scanner_enumeration_restart.py @@ -37,6 +37,10 @@ def validate_report(report, *, round_number, pid, objects, budget): raise ValueError("nonempty fixture must observe raw entries; budget hook may not have run") if report["raw_entries"] > budget: raise ValueError("raw-entry budget exceeded; no unbudgeted tail is permitted") + for key in ("raw_first_entry", "raw_last_entry"): + value = report.get(key) + if type(value) is not str or not 0 < len(value.encode("utf-8")) <= 512: + raise ValueError(f"invalid raw entry marker: {key}") if type(report.get("snapshot_complete")) is not bool: raise ValueError("missing explicit completeness") if report.get("outcome") not in ("complete", "partial", "cancelled_without_cache"): @@ -49,6 +53,13 @@ def converged(report, objects): ("objects_retained", "versions_retained", "bytes_retained"))) +def replays_raw_window(previous, current): + return (previous["raw_first_entry"] == current["raw_first_entry"] + and previous["raw_last_entry"] == current["raw_last_entry"] + and previous["objects_retained"] == current["objects_before"] + and current["objects_retained"] == previous["objects_retained"]) + + def run(args): binary = args.test_binary.resolve(strict=True) listed = subprocess.run([str(binary), WORKER, "--exact", "--list"], @@ -58,6 +69,7 @@ def run(args): workspace = args.output.resolve() workspace.mkdir() # Refuse reuse/overwrite of previous evidence or customer data. reports = [] + replayed_raw_window = False for round_number in range(args.rounds): request = {"workspace": str(workspace), "objects": args.objects, "raw_entry_budget": args.raw_entry_budget, "round": round_number} @@ -87,12 +99,15 @@ def run(args): objects=args.objects, budget=args.raw_entry_budget) if reports and report["objects_before"] != reports[-1]["objects_retained"]: raise ValueError("cache coverage did not survive the process boundary") + if reports and replays_raw_window(reports[-1], report): + replayed_raw_window = True reports.append(report) print(json.dumps(report, sort_keys=True), flush=True) if converged(report, args.objects): print("PASS: bounded scanner-worker restart convergence for this fixture only") return 0 - print("FAIL: fixed-budget restart convergence not established; R-E gate remains unmet", + reason = "replayed raw enumeration window" if replayed_raw_window else "no bounded restart convergence" + print(f"FAIL: fixed-budget restart convergence not established ({reason}); R-E gate remains unmet", file=sys.stderr) return 1 diff --git a/scripts/test_diagnose_scanner_enumeration_restart.py b/scripts/test_diagnose_scanner_enumeration_restart.py index efe5f2ddd..208b1d040 100644 --- a/scripts/test_diagnose_scanner_enumeration_restart.py +++ b/scripts/test_diagnose_scanner_enumeration_restart.py @@ -2,7 +2,7 @@ import unittest -from diagnose_scanner_enumeration_restart import converged, validate_report +from diagnose_scanner_enumeration_restart import converged, replays_raw_window, validate_report class ReportTests(unittest.TestCase): @@ -10,6 +10,8 @@ class ReportTests(unittest.TestCase): return dict(schema=1, round=0, pid=123, objects_expected=4, raw_entry_budget=16, raw_entries=8, raw_name_bytes=64, objects_before=0, objects_retained=4, versions_retained=4, bytes_retained=4, objects_processed=4, + raw_first_entry="bucket/object-0000", + raw_last_entry="bucket/object-0003/xl.meta", snapshot_complete=True, outcome="complete") def validate(self, report): @@ -42,6 +44,14 @@ class ReportTests(unittest.TestCase): with self.assertRaises(ValueError): self.validate(report) + def test_missing_or_oversized_raw_marker_rejected(self): + for value in (None, True, "", "x" * 513): + with self.subTest(value=value): + report = self.report() + report["raw_first_entry"] = value + with self.assertRaises(ValueError): + self.validate(report) + def test_complete_coverage_without_entry_observation_rejected(self): report = self.report() report["raw_entries"] = 0 @@ -68,6 +78,20 @@ class ReportTests(unittest.TestCase): with self.assertRaises(ValueError): self.validate(report) + def test_repeated_raw_window_without_retained_progress_is_diagnosed(self): + previous = self.report() + previous["objects_retained"] = 0 + previous["versions_retained"] = 0 + previous["bytes_retained"] = 0 + previous["objects_processed"] = 0 + previous["snapshot_complete"] = False + previous["outcome"] = "cancelled_without_cache" + current = dict(previous, round=1, pid=124, objects_before=0) + self.assertTrue(replays_raw_window(previous, current)) + + advanced = dict(current, objects_retained=1) + self.assertFalse(replays_raw_window(previous, advanced)) + if __name__ == "__main__": unittest.main()