#!/usr/bin/env python3 """Black-box durability checks using an independent OS process and real SIGKILL. Only temporary test worlds are used. This is not a simulation of power loss. Run after cargo build -p shacraft-tools (or use --binary for release). """ import argparse import json from pathlib import Path import subprocess import tempfile import time class Session: def __init__(self, binary, data): self.process = subprocess.Popen( [str(binary), "--data", str(data), "--cache", "2", "session"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", bufsize=1, ) def request(self, request, success=True): self.process.stdin.write(json.dumps(request, separators=(",", ":")) + "\n") self.process.stdin.flush() line = self.process.stdout.readline() if not line: raise AssertionError(f"worker exited: {self.process.stderr.read()}") result = json.loads(line) assert result["ok"] == success, result return result.get("result") if success else result["error"] def close(self, abrupt=False): if abrupt: self.process.kill() else: self.process.stdin.close() self.process.wait(timeout=15) self.process.stdout.close() self.process.stderr.close() if abrupt: self.process.stdin.close() else: assert self.process.returncode == 0, self.process.returncode def check(binary, directory): checks = [] active = Session(binary, directory) try: block = active.request({"op": "register", "state": "shacraft:test_stone"}) active.request({"op": "create", "name": "main"}) acknowledged = [] for index in range(80): operation = {"op": "edit", "world": "main", "expected_revision": index, "operation_id": f"ack-{index}", "changes": [{"pos": [index * 16 - 640 + index % 16, 0, -1], "block": block}]} result = active.request(operation) assert result["revision"] == index + 1 acknowledged.append(operation) for operation in acknowledged: assert active.request({"op": "get", "world": "main", "pos": operation["changes"][0]["pos"]}) == block stats = active.request({"op": "stats"}) assert 0 < stats["storage"]["cache_entries"] <= 2, stats assert stats["storage"]["cache_evictions"] > 0, stats checks.append("bounded resident cache with verified eviction while reading 80 sections") duplicate = subprocess.run([str(binary), "--data", str(directory), "stats"], capture_output=True, text=True, timeout=15) assert duplicate.returncode != 0, "second writer unexpectedly opened live store" checks.append("second writer rejected while owner is running") # Every operation above has returned a flushed success reply. No flush command. active.close(abrupt=True) active = Session(binary, directory) assert active.request({"op": "revision", "world": "main"}) == 80 assert active.request({"op": "register", "state": "shacraft:test_stone"}) == block for operation in acknowledged: pos = operation["changes"][0]["pos"] assert active.request({"op": "get", "world": "main", "pos": pos}) == block replay = active.request(acknowledged[0]) assert replay["replayed"] and replay["revision"] == 1 assert active.request({"op": "revision", "world": "main"}) == 80 checks.append("80 acknowledged edits and registry survive SIGKILL without flush") checks.append("idempotent replay after SIGKILL precedes current revision check") invalid = {"op": "edit", "world": "main", "expected_revision": 80, "operation_id": "invalid-atomic", "changes": [{"pos": [1, 1, 1], "block": block}, {"pos": [2, 1, 1], "block": 4294967295}]} active.request(invalid, success=False) assert active.request({"op": "get", "world": "main", "pos": [1, 1, 1]}) == 0 assert active.request({"op": "revision", "world": "main"}) == 80 checks.append("invalid batch leaves both blocks and revision unchanged") # A large request interrupted before observing the reply must be all-or-none. active.request({"op": "create", "name": "interrupted"}) changes = [{"pos": [i % 64, (i // 64) % 8, i // 512], "block": block} for i in range(32768)] inflight = {"op": "edit", "world": "interrupted", "expected_revision": 0, "operation_id": "in-flight", "changes": changes} active.process.stdin.write(json.dumps(inflight, separators=(",", ":")) + "\n") active.process.stdin.flush() time.sleep(0.005) active.close(abrupt=True) active = Session(binary, directory) revision = active.request({"op": "revision", "world": "interrupted"}) restored = active.request({"op": "read", "world": "interrupted", "min": [0, 0, 0], "max": [63, 7, 63]}) assert (revision, len(restored)) in [(0, 0), (1, 32768)], (revision, len(restored)) assert all(item["block"] == block for item in restored) checks.append(f"interrupted 32768-cell transaction is atomic (recovered revision {revision})") retry = active.request(inflight) assert retry["revision"] == 1 assert active.request({"op": "revision", "world": "interrupted"}) == 1 checks.append("uncertain interrupted request can safely be retried") # Request-level tests cover stale versions, malformed JSON and valid recovery. active.request({"op": "edit", "world": "main", "expected_revision": 0, "operation_id": "stale", "changes": []}, success=False) active.process.stdin.write('{broken json}\n') active.process.stdin.flush() assert not json.loads(active.process.stdout.readline())["ok"] assert active.request({"op": "revision", "world": "main"}) == 80 checks.append("invalid request does not terminate the local test session") active.request({"op": "create", "name": "uniform"}) uniform = [{"pos": [x, y, z], "block": block} for y in range(16) for z in range(16) for x in range(16)] active.request({"op": "edit", "world": "uniform", "expected_revision": 0, "operation_id": "fill", "changes": uniform}) active.close() active = Session(binary, directory) assert active.request({"op": "get", "world": "uniform", "pos": [8, 8, 8]}) == block compact = active.request({"op": "stats"})["storage"] assert compact["cache_entries"] == 1, compact assert compact["cache_payload_bytes"] == 20, compact checks.append("4096-cell uniform section occupies 20 encoded payload bytes in resident cache") return {"passed": True, "checks": checks, "final_metrics": active.request({"op": "stats"}), "limits": ["process crash tested; power loss and faulty hardware are not simulated", "storage-only workload; no Paper or gameplay performance comparison"]} finally: if active.process.poll() is None: active.close() def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--binary", type=Path, default=Path(__file__).resolve().parents[1] / "target/debug/shacraft-tools") args = parser.parse_args() binary = args.binary.resolve() if not binary.is_file(): parser.error("build first: cargo build -p shacraft-tools") with tempfile.TemporaryDirectory(prefix="shacraft-storage-check-") as directory: print(json.dumps(check(binary, Path(directory)), ensure_ascii=False, indent=2)) if __name__ == "__main__": main()