gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-08 13:06:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(scanner): prove restart quantum stages (#7414)

Browse Source 
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-07 23:22:02 +08:00
committed by GitHub
parent 0686277ee4
commit 752d4a81ab
4 changed files with 149 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/diagnose_scanner_enumeration_restart.py
+49 -3
View File
@@ -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)
scripts/test_diagnose_scanner_enumeration_restart.py
+86 -2
View File
@@ -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()