From 752d4a81ab0fb89713045898c36684085549de5a Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 23:22:02 +0800 Subject: [PATCH] test(scanner): prove restart quantum stages (#7414) Co-authored-by: zhi22915 --- .../tests/enumeration_restart.rs | 16 +++- docs/testing/scanner-checkpoint-fixture.md | 4 +- .../diagnose_scanner_enumeration_restart.py | 52 ++++++++++- ...st_diagnose_scanner_enumeration_restart.py | 88 ++++++++++++++++++- 4 files changed, 149 insertions(+), 11 deletions(-) diff --git a/crates/scanner/src/scanner_folder/tests/enumeration_restart.rs b/crates/scanner/src/scanner_folder/tests/enumeration_restart.rs index 48d06f885..2eb78db1e 100644 --- a/crates/scanner/src/scanner_folder/tests/enumeration_restart.rs +++ b/crates/scanner/src/scanner_folder/tests/enumeration_restart.rs @@ -166,16 +166,22 @@ async fn round(request: &Request) -> serde_json::Value { 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"); - let raw_page_index_committed_entries = reloaded - .validated_raw_enumeration_page_index() + let raw_page_index = reloaded.validated_raw_enumeration_page_index(); + let raw_page_index_committed_entries = raw_page_index .and_then(|index| index.committed_entries().ok()) .map(|entries| entries.len()) .unwrap_or(0); - let raw_page_index_indexed_entries = reloaded - .validated_raw_enumeration_page_index() + let raw_page_index_indexed_entries = raw_page_index .and_then(|index| index.indexed_entries().ok()) .map(|entries| entries.len()) .unwrap_or(0); + let (raw_page_index_parent, raw_page_index_complete) = raw_page_index + .map(|index| match index.status() { + crate::raw_page_index::RawEnumerationPageOwnerStatus::Building { parent, .. } => (Some(parent), false), + crate::raw_page_index::RawEnumerationPageOwnerStatus::Ready { parent, complete, .. } => (Some(parent), complete), + crate::raw_page_index::RawEnumerationPageOwnerStatus::Unsupported => (None, false), + }) + .unwrap_or((None, false)); assert_eq!( (retained.objects, retained.versions, retained.size), (scanned.objects, scanned.versions, scanned.size) @@ -188,6 +194,8 @@ async fn round(request: &Request) -> serde_json::Value { "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, + "raw_page_index_parent": raw_page_index_parent, + "raw_page_index_complete": raw_page_index_complete, "raw_page_index_committed_entries": raw_page_index_committed_entries, "raw_page_index_indexed_entries": raw_page_index_indexed_entries, "objects_processed": budget.progress().0, diff --git a/docs/testing/scanner-checkpoint-fixture.md b/docs/testing/scanner-checkpoint-fixture.md index f9d437905..60a32e80d 100644 --- a/docs/testing/scanner-checkpoint-fixture.md +++ b/docs/testing/scanner-checkpoint-fixture.md @@ -19,11 +19,11 @@ python3 scripts/diagnose_scanner_enumeration_restart.py \ --objects 128 --raw-entry-budget 8 --rounds 8 ``` -The output directory must not exist. Each round starts a new OS test-worker process, opens the same synthetic disk, decodes the preceding cache, invokes the real scanner, encodes the returned cache, and decodes it again. When cancellation returns no useful partial cache, it preserves the previous cache. Reports identify the actual child PID, round, raw entries and name bytes observed, processed objects, retained object/version/byte counts, and completeness. No observed-name set, `readdir` offset, or assumed stable ordering is used as durable progress. Namespace creation happens only during fixture setup, before scan accounting. +The output directory must not exist. Each round starts a new OS test-worker process, opens the same synthetic disk, decodes the preceding cache, invokes the real scanner, encodes the returned cache, and decodes it again. When cancellation returns no useful partial cache, it preserves the previous cache. Reports identify the actual child PID, round, raw entries and name bytes observed, raw page-index parent/entries, processed/classified objects, retained object/version/byte counts, and completeness. The driver rejects retained coverage that advances beyond classified object work, root raw page indexes that outrun the fixture namespace, committed page coverage that exceeds indexed coverage, same-parent committed coverage regressions before completion, and any process-restart regression in retained coverage. No observed-name set, `readdir` offset, or assumed stable ordering is used as durable progress. Namespace creation happens only during fixture setup, before scan accounting. The `cfg(test)` hook observes actual entries delivered by `read_dir` and cancels the existing cycle token at the fixed entry limit. This is a deterministic injected **raw-entry work budget**, not a wall-clock performance measurement or a claim that kernel prefetch, probes, allocations, name bytes, or cache I/O are independently budgeted. The watchdog timeout only bounds worker lifetime. The hook does not replace enumeration, classification, or recursion, and does not exist in production builds. In particular, `xl.meta` object-boundary classification is unchanged. -Exit 0 requires exact complete object/version/byte coverage within the same fixed budget on every executed round. Exit 1 means the strict convergence oracle remains unmet, including the current flat-directory enumeration starvation case. Exit 2 means invalid input, worker failure, or invalid evidence; it is not a successful reproduction. There is no final unbudgeted sweep. Small fixtures can pass; that does not establish the general R-E gate from [the scanner review comment](https://github.com/rustfs/backlog/issues/2240#issuecomment-5549222480). Raw entries observed are not a retained enumeration watermark. This is scanner-worker process restart plus codec evidence, **not** whole-daemon restart, EC quorum persistence, crash/fsync durability, remote RPC, or a throughput benchmark. The caller owns the bounded evidence directory and may remove it after inspection. +Exit 0 requires exact complete object/version/byte coverage within the same fixed budget on every executed round and positive evidence for all three stages: raw enumeration/indexing, object classification/processing, and durable cache retention after a fresh worker process reloads the previous report. Exit 1 means the strict convergence oracle remains unmet, including the current flat-directory enumeration starvation case. Exit 2 means invalid input, worker failure, or invalid evidence; it is not a successful reproduction. There is no final unbudgeted sweep. Small fixtures can pass; that does not establish the general R-E gate from [the scanner review comment](https://github.com/rustfs/backlog/issues/2240#issuecomment-5549222480). Raw entries observed are not a retained enumeration watermark. This is scanner-worker process restart plus codec evidence, **not** whole-daemon restart, EC quorum persistence, crash/fsync durability, remote RPC, or a throughput benchmark. The caller owns the bounded evidence directory and may remove it after inspection. ### Missing Storage Capability diff --git a/scripts/diagnose_scanner_enumeration_restart.py b/scripts/diagnose_scanner_enumeration_restart.py index de9810af0..8acbee1a3 100644 --- a/scripts/diagnose_scanner_enumeration_restart.py +++ b/scripts/diagnose_scanner_enumeration_restart.py @@ -47,10 +47,24 @@ def validate_report(report, *, round_number, pid, objects, budget): continue if type(value) is not str or not 0 < len(value.encode("utf-8")) <= 512: raise ValueError(f"invalid raw entry marker: {key}") + if "raw_page_index_parent" not in report: + raise ValueError("missing raw page index parent") + raw_page_index_parent = report.get("raw_page_index_parent") + if raw_page_index_parent is not None and (type(raw_page_index_parent) is not str + or not 0 < len(raw_page_index_parent.encode("utf-8")) <= 512): + raise ValueError("invalid raw page index parent") + if type(report.get("raw_page_index_complete")) is not bool: + raise ValueError("missing raw page index completeness") 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"): raise ValueError("unexpected scanner outcome") + if report["raw_page_index_committed_entries"] > report["raw_page_index_indexed_entries"]: + raise ValueError("raw page index committed entries exceed indexed entries") + if report["raw_page_index_parent"] == "bucket" and report["raw_page_index_indexed_entries"] > objects: + raise ValueError("raw page index exceeds fixture object count") + if report["objects_retained"] > report["objects_before"] + report["objects_processed"]: + raise ValueError("retained coverage advanced beyond classified object work") def converged(report, objects): @@ -66,6 +80,38 @@ def replays_raw_window(previous, current): and current["objects_retained"] == previous["objects_retained"]) +def validate_recoverable_quantum(reports, *, objects, budget, require_converged): + if not reports: + raise ValueError("no scanner restart reports were produced") + previous = None + made_enumeration_progress = False + made_classification_progress = False + made_durable_progress = False + for index, report in enumerate(reports): + validate_report(report, round_number=index, pid=report["pid"], objects=objects, budget=budget) + if previous is not None: + if report["objects_before"] != previous["objects_retained"]: + raise ValueError("durable retained coverage did not survive process restart") + if report["objects_retained"] < previous["objects_retained"]: + raise ValueError("durable retained coverage regressed across restart") + if (report["raw_page_index_parent"] == previous["raw_page_index_parent"] + and report["raw_page_index_committed_entries"] < previous["raw_page_index_committed_entries"] + and not previous["raw_page_index_complete"]): + raise ValueError("committed raw enumeration page coverage regressed before completion") + made_enumeration_progress |= report["raw_entries"] > 0 or report["raw_page_index_indexed_entries"] > 0 + made_classification_progress |= report["objects_processed"] > 0 + made_durable_progress |= report["objects_retained"] > report["objects_before"] + previous = report + if not made_enumeration_progress: + raise ValueError("restart proof did not exercise raw enumeration") + if not made_classification_progress: + raise ValueError("restart proof did not exercise object classification") + if not made_durable_progress: + raise ValueError("restart proof did not persist processed object coverage") + if require_converged and not converged(reports[-1], objects): + raise ValueError("fixed-budget restart convergence was not established") + + def run(args): binary = args.test_binary.resolve(strict=True) listed = subprocess.run([str(binary), WORKER, "--exact", "--list"], @@ -103,15 +149,15 @@ def run(args): report = json.loads(raw) validate_report(report, round_number=round_number, pid=worker.pid, 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") + validate_recoverable_quantum(reports, objects=args.objects, budget=args.raw_entry_budget, require_converged=True) + print("PASS: bounded scanner-worker restart convergence with enumeration/classification/processing evidence") return 0 + validate_recoverable_quantum(reports, objects=args.objects, budget=args.raw_entry_budget, require_converged=False) 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) diff --git a/scripts/test_diagnose_scanner_enumeration_restart.py b/scripts/test_diagnose_scanner_enumeration_restart.py index 718a6cf17..02120f26e 100644 --- a/scripts/test_diagnose_scanner_enumeration_restart.py +++ b/scripts/test_diagnose_scanner_enumeration_restart.py @@ -2,7 +2,12 @@ import unittest -from diagnose_scanner_enumeration_restart import converged, replays_raw_window, validate_report +from diagnose_scanner_enumeration_restart import ( + converged, + replays_raw_window, + validate_recoverable_quantum, + validate_report, +) class ReportTests(unittest.TestCase): @@ -10,6 +15,7 @@ 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_page_index_parent="bucket", raw_page_index_complete=True, raw_page_index_committed_entries=4, raw_page_index_indexed_entries=4, raw_first_entry="bucket/object-0000", @@ -46,6 +52,26 @@ class ReportTests(unittest.TestCase): with self.assertRaises(ValueError): self.validate(report) + def test_raw_page_index_ordering_and_bounds_are_checked(self): + report = self.report() + report["raw_page_index_committed_entries"] = 5 + report["raw_page_index_indexed_entries"] = 4 + with self.assertRaisesRegex(ValueError, "committed entries exceed indexed entries"): + self.validate(report) + + report = self.report() + report["raw_page_index_indexed_entries"] = 5 + with self.assertRaisesRegex(ValueError, "exceeds fixture object count"): + self.validate(report) + + def test_retained_coverage_cannot_advance_without_classified_work(self): + report = self.report() + report["objects_before"] = 1 + report["objects_processed"] = 1 + report["objects_retained"] = 3 + with self.assertRaisesRegex(ValueError, "advanced beyond classified object work"): + self.validate(report) + report = self.report() report["objects_processed"] = 17 with self.assertRaises(ValueError): @@ -73,6 +99,11 @@ class ReportTests(unittest.TestCase): report["raw_first_entry"] = None report["raw_last_entry"] = None report["objects_processed"] = 1 + report["objects_retained"] = 1 + report["versions_retained"] = 1 + report["bytes_retained"] = 1 + report["snapshot_complete"] = False + report["outcome"] = "partial" self.validate(report) def test_missing_wrong_type_and_negative_counter_rejected(self): @@ -84,7 +115,7 @@ class ReportTests(unittest.TestCase): self.validate(report) def test_missing_completeness_or_unknown_outcome_rejected(self): - for key in ("snapshot_complete", "outcome"): + for key in ("raw_page_index_parent", "raw_page_index_complete", "snapshot_complete", "outcome"): report = self.report() del report[key] with self.assertRaises(ValueError): @@ -109,6 +140,59 @@ class ReportTests(unittest.TestCase): advanced = dict(current, objects_retained=1) self.assertFalse(replays_raw_window(previous, advanced)) + def test_recoverable_quantum_requires_three_stage_progress_and_convergence(self): + first = self.report() + first.update(round=0, pid=123, raw_entries=2, raw_page_index_committed_entries=2, + raw_page_index_indexed_entries=2, objects_processed=2, objects_before=0, + objects_retained=2, versions_retained=2, bytes_retained=2, + snapshot_complete=False, outcome="partial") + second = self.report() + second.update(round=1, pid=124, raw_entries=2, raw_page_index_committed_entries=4, + raw_page_index_indexed_entries=4, objects_processed=2, objects_before=2, + objects_retained=4, snapshot_complete=True, outcome="complete") + validate_recoverable_quantum([first, second], objects=4, budget=16, require_converged=True) + + def test_recoverable_quantum_rejects_restart_regression(self): + first = self.report() + first.update(snapshot_complete=False, outcome="partial", objects_retained=2, + versions_retained=2, bytes_retained=2) + second = self.report() + second.update(round=1, pid=124, objects_before=1, objects_retained=1, + versions_retained=1, bytes_retained=1, snapshot_complete=False, + outcome="partial") + with self.assertRaisesRegex(ValueError, "did not survive process restart"): + validate_recoverable_quantum([first, second], objects=4, budget=16, require_converged=False) + + def test_recoverable_quantum_allows_new_raw_page_parent_after_processing(self): + first = self.report() + first.update(snapshot_complete=False, outcome="partial", objects_before=0, + objects_processed=0, objects_retained=0, versions_retained=0, + bytes_retained=0, raw_page_index_parent="bucket", + raw_page_index_complete=True, raw_page_index_committed_entries=4, + raw_page_index_indexed_entries=4) + second = self.report() + second.update(round=1, pid=124, raw_entries=0, raw_first_entry=None, + raw_last_entry=None, raw_name_bytes=0, objects_before=0, + objects_processed=2, objects_retained=2, + versions_retained=2, bytes_retained=2, + snapshot_complete=False, outcome="partial", + raw_page_index_parent="bucket/object-0000", + raw_page_index_complete=False, + raw_page_index_committed_entries=1, + raw_page_index_indexed_entries=1) + validate_recoverable_quantum([first, second], objects=4, budget=16, require_converged=False) + + def test_recoverable_quantum_rejects_missing_processing_stage(self): + report = self.report() + report["objects_processed"] = 0 + report["objects_retained"] = 0 + report["versions_retained"] = 0 + report["bytes_retained"] = 0 + report["snapshot_complete"] = False + report["outcome"] = "partial" + with self.assertRaisesRegex(ValueError, "object classification"): + validate_recoverable_quantum([report], objects=4, budget=16, require_converged=False) + if __name__ == "__main__": unittest.main()