Problem class: python-uring-completion-reorder-crash-recovery-20261009
I diagnosed the problem class and built a runnable reference implementation plus an adversarial verifier. Files are at ~/solution/reference.py and ~/solution/solution.md. Below is the complete solution.
Problem class: python-uring-completion-reorder-crash-recovery-20261009
A userspace storage engine submits linked and multishot SQEs across several io_uring queues. After a crash it either applies an operation twice, loses an operation whose completion it never saw, aborts an operation that completed, or replays a journal that grows with total history. All of these come from one mistake: treating CQE arrival order as the transaction log.
A linked chain's starts are ordered and a multishot op emits CQEs with IORING_CQE_F_MORE set until the final one. But the kernel does not guarantee that CQEs across queues are interleaved in submission order, that a cancellation completion arrives before/after the target CQE, or that a CQE is delivered exactly once. Positional inference ("the 3rd CQE is the 3rd SQE") and "wait for N before N+1" are unsound. Identity and ordering must live in user_data and in a journal.
IORING_OP_ASYNC_CANCEL can return -EALREADY, 0, or -ENOENT, and the target may still deliver a normal CQE already sitting in the CQ ring. Any resolution based on CQE order is wrong under reordering.
There are two durable objects: the effect store (data) and the decision journal (metadata). A crash can land between them. The journal must be the single source of truth and effects must be redone idempotently. Recording "already applied" metadata and skipping redo loses the effect when the data page did not reach disk.
Without a checkpoint, replay is O(total history). A bounded journal anchored by a checksummed checkpoint makes replay O(bounded segment length).
If compaction truncates to empty and the next append recomputes lsn = last + 1, LSNs restart at 1. Records appended after a checkpoint at LSN L then collide with pre-checkpoint LSNs and are misclassified as stale, silently dropping committed work. The LSN namespace must stay monotonic across compaction (a durable BASE marker at the checkpoint LSN solves this).
(op_id, generation, sub_index).BEGIN) and journal every observed CQE and CANCEL with a CRC and monotonic LSN; discard a trailing torn record.done iff all sub-completions succeed, failed on any negative; multishot → MORE means progress, final CQE means done; -ECANCELED → cancelled.CANCEL with no terminal CQE finalizes cancelled. This makes the race order-independent.op_id; recovery redoes every done op whose key is missing.BASE marker, replay at most capacity records.cancelled (at-most-once); durably-committed ops are applied exactly once.reference.py)#!/usr/bin/env python3
"""
io_uring linked/multishot completion-reorder + crash-recovery reference model.
This is a *simulation* of the kernel/userspace boundary. It does not call
io_uring (the sandbox has no ring and CPython has no stdlib io_uring); instead
it models exactly the observable contract that matters for recovery:
* SQEs are tagged with a logical op id + generation, journaled before submit.
* CQEs may be delivered in ANY order, duplicated, dropped, or be stale
(lower generation). They are durable only once journaled.
* linked groups and multishot ops produce several CQEs for one logical op.
* ASYNC_CANCEL races CQE delivery: cancel intent is durable, resolution is
deterministic and order-independent.
* A crash can happen between any two durable writes (data vs metadata).
* The journal is bounded (ring capacity). Recovery = last valid checkpoint +
replay of at most `capacity` records, therefore sublinear in total history.
Run: python3 reference.py # randomized adversarial verification
"""
from __future__ import annotations
import json
import os
import random
import shutil
import tempfile
import zlib
from typing import Any, Optional
# --------------------------------------------------------------------------
# Durable primitives
# --------------------------------------------------------------------------
class TornWrite(Exception):
pass
class DurableJournal:
"""CRC32-checked, append-only, fsync-per-record journal.
A trailing partial/corrupt record is treated as a torn append (crash during
write) and silently dropped -- never trusted.
"""
def __init__(self, path: str, capacity: int = 32) -> None:
self.path = path
self.capacity = capacity
# -- encoding ---------------------------------------------------------
@staticmethod
def _encode(rec: dict) -> str:
body = json.dumps(rec, sort_keys=True, separators=(",", ":"))
return f"{zlib.crc32(body.encode()):08x} {body}\n"
def _read_all(self) -> list[dict]:
recs: list[dict] = []
if not os.path.exists(self.path):
return recs
with open(self.path, "r") as f:
for line in f:
line = line.rstrip("\n")
if not line:
continue
try:
crc_s, body = line.split(" ", 1)
if zlib.crc32(body.encode()) != int(crc_s, 16):
break # corrupt/torn -> stop, ignore rest
recs.append(json.loads(body))
except (ValueError, json.JSONDecodeError):
break
return recs
def records(self) -> list[dict]:
"""Real (non-BASE) records -- what recovery must replay."""
return [r for r in self._read_all() if r["kind"] != "BASE"]
def last_lsn(self) -> int:
recs = self._read_all()
return recs[-1]["lsn"] if recs else 0
# -- api --------------------------------------------------------------
def append(self, kind: str, **kw: Any) -> int:
# LSN is monotonic across truncation because truncate_through leaves a
# durable BASE marker carrying the checkpoint LSN.
lsn = self.last_lsn() + 1
rec = {"lsn": lsn, "kind": kind, **kw}
# O_APPEND + fsync: record is durable before we return.
with open(self.path, "a") as f:
f.write(self._encode(rec))
f.flush()
os.fsync(f.fileno())
return lsn
def records_since(self, lsn: int) -> list[dict]:
return [r for r in self.records() if r["lsn"] > lsn]
def truncate_through(self, lsn: int) -> None:
"""Atomically compact to a single durable BASE marker at `lsn`.
The BASE marker keeps the LSN namespace monotonic after truncation so
that post-checkpoint records can never collide with (and thus be
mistaken for) pre-checkpoint records.
"""
tmp = self.path + ".tmp"
with open(tmp, "w") as f:
f.write(self._encode({"lsn": lsn, "kind": "BASE"}))
f.flush()
os.fsync(f.fileno())
os.replace(tmp, self.path)
class CheckpointStore:
"""Atomic (write-temp + rename) checksummed state snapshot."""
def __init__(self, path: str) -> None:
self.path = path
def save(self, state: dict, lsn: int) -> None:
body = json.dumps(
{"state": state, "lsn": lsn}, sort_keys=True, separators=(",", ":")
)
tmp = self.path + ".tmp"
with open(tmp, "w") as f:
f.write(f"{zlib.crc32(body.encode()):08x} {body}")
f.flush()
os.fsync(f.fileno())
os.replace(tmp, self.path)
def load(self) -> tuple[dict, int]:
if not os.path.exists(self.path):
return {"ops": {}}, 0
with open(self.path) as f:
data = f.read()
crc_s, body = data.split(" ", 1)
if zlib.crc32(body.encode()) != int(crc_s, 16):
raise TornWrite("checkpoint is corrupt")
payload = json.loads(body)
return payload["state"], payload["lsn"]
# --------------------------------------------------------------------------
# Protocol engine
# --------------------------------------------------------------------------
TERMINAL = {"done", "failed", "cancelled"}
# A CQE result encoding. >= 0 == success byte-count, < 0 == -errno.
ECANCELED = -125
EIO = -5
class Storage:
"""Idempotent, op-id-keyed effect store (the 'data' half of the engine).
The value map is persisted atomically (temp + rename), exactly like a data
page write. Applying the same op id twice is a logical no-op: the map has
exactly one entry per committed op, so redo after a crash cannot duplicate
a logical effect. Physical writes may repeat -- that is the point of a
redo log.
"""
def __init__(self, path: str) -> None:
self.path = path
self.value: dict[int, int] = {}
self.physical_writes = 0
self._load()
def _load(self) -> None:
if not os.path.exists(self.path):
return
with open(self.path) as f:
self.value = {int(k): int(v) for k, v in json.load(f).items()}
def apply(self, op_id: int, effect: int) -> None:
self.physical_writes += 1
self.value[op_id] = effect
tmp = self.path + ".tmp"
with open(tmp, "w") as f:
json.dump({str(k): v for k, v in self.value.items()}, f, sort_keys=True)
f.flush()
os.fsync(f.fileno())
os.replace(tmp, self.path) # data becomes durable atomically
def corrupt(self) -> None:
"""Model a crash where the data page never reached disk."""
if os.path.exists(self.path):
os.remove(self.path)
class Engine:
"""Journaled state machine; every state change is derived from records."""
def __init__(self, workdir: str, capacity: int = 32) -> None:
self.workdir = workdir
self.journal = DurableJournal(os.path.join(workdir, "journal.log"), capacity)
self.cp = CheckpointStore(os.path.join(workdir, "checkpoint"))
self.capacity = capacity
# volatile in-memory projection
self.ops: dict[int, dict] = {}
self.applied: set[int] = set()
self.storage = Storage(os.path.join(workdir, "data.json"))
self.replay_count = 0
self.checkpoints = 0
self._load()
# -- recovery ---------------------------------------------------------
def _load(self) -> None:
state, ckpt_lsn = self.cp.load()
# normalize json object keys back to int
self.ops = {int(k): v for k, v in state["ops"].items()}
recs = self.journal.records_since(ckpt_lsn)
self.replay_count = len(recs)
for rec in recs:
self._replay(rec)
# durable cancel intents with no terminal outcome resolve to cancelled
self._resolve_open()
self._redo_effects()
def _replay(self, rec: dict) -> None:
if rec["kind"] == "BASE":
return
self._apply(rec, durable=True)
def _resolve_open(self) -> None:
"""Deterministically finalize every still-open op as aborted.
A logical op with no durably-observed terminal CQE is *not* known to
have committed; choosing 'cancelled' is the safe at-most-once choice
and guarantees a single outcome per op regardless of dropped CQEs.
"""
for op in self.ops.values():
if op["outcome"] is None:
op["outcome"] = "cancelled"
def _redo_effects(self) -> None:
"""Redo every durably-committed ('done') op that is not already in the
durable effect store. The store is keyed by op id, so replaying a
COMMIT can never double-apply a logical effect; if the data page was
lost, the missing key is re-created here."""
for op_id, op in self.ops.items():
if op["outcome"] == "done" and op_id not in self.storage.value:
self.storage.apply(op_id, op["effect"])
self.applied = set(self.storage.value.keys())
# -- record application (idempotent) ----------------------------------
def _op(self, op_id: int) -> dict:
op = self.ops.get(op_id)
if op is None:
op = {
"kind": "simple",
"nsub": 1,
"gen": 0,
"effect": 0,
"progress": {},
"outcome": None,
"cancel_pending": False,
}
self.ops[op_id] = op
return op
def _apply(self, rec: dict, durable: bool) -> Optional[str]:
"""Apply one record. Returns newly-decided outcome (or None)."""
kind = rec["kind"]
if kind == "BEGIN":
op = self._op(rec["op_id"])
# BEGIN is only authoritative for the first generation and only
# while the op is still open.
if rec.get("gen", 0) >= op["gen"] and op["outcome"] is None:
op.update(
kind=rec["kind_name"],
nsub=rec["nsub"],
gen=rec.get("gen", 0),
effect=rec["effect"],
)
return None
if kind == "CQE":
return self._apply_cqe(
rec["op_id"], rec["index"], rec["result"], rec.get("more", False),
rec.get("gen", 0),
)
if kind == "CANCEL":
op = self._op(rec["op_id"])
if rec.get("gen", 0) >= op["gen"] and op["outcome"] is None:
if durable:
op["cancel_pending"] = True
return None
return None
def _apply_cqe(
self, op_id: int, index: int, result: int, more: bool, gen: int
) -> Optional[str]:
op = self._op(op_id)
if gen != op["gen"]:
return None # stale CQE from an older generation
if op["outcome"] is not None:
return None # duplicate / already terminal
op["progress"][str(index)] = result
op["cancel_pending"] = False # a real CQE supersedes pending cancel
outcome: Optional[str] = None
if result == ECANCELED:
outcome = "cancelled"
elif result < 0:
outcome = "failed"
elif op["kind"] == "simple":
outcome = "done"
elif op["kind"] == "multishot":
outcome = None if more else "done"
elif op["kind"] == "linked":
if len(op["progress"]) >= op["nsub"]:
outcome = "done"
if outcome is not None:
op["outcome"] = outcome
return outcome
# -- public operations (journal-first) --------------------------------
def begin(self, op_id: int, kind_name: str, nsub: int, effect: int, gen: int = 0) -> None:
lsn = self.journal.append(
"BEGIN", op_id=op_id, kind_name=kind_name, nsub=nsub, effect=effect, gen=gen
)
self._apply(
{"kind": "BEGIN", "op_id": op_id, "kind_name": kind_name,
"nsub": nsub, "effect": effect, "gen": gen},
durable=True,
)
self._maybe_checkpoint()
def complete(
self, op_id: int, index: int = 0, result: int = 1, more: bool = False, gen: int = 0
) -> Optional[str]:
self.journal.append(
"CQE", op_id=op_id, index=index, result=result, more=more, gen=gen
)
out = self._apply_cqe(op_id, index, result, more, gen)
self._maybe_checkpoint()
return out
def cancel(self, op_id: int, gen: int = 0) -> None:
self.journal.append("CANCEL", op_id=op_id, gen=gen)
self._apply({"kind": "CANCEL", "op_id": op_id, "gen": gen}, durable=True)
self._maybe_checkpoint()
# -- checkpoint / compaction ------------------------------------------
def _maybe_checkpoint(self) -> None:
if len(self.journal.records()) >= self.capacity:
self.checkpoint()
def checkpoint(self) -> None:
lsn = self.journal.last_lsn()
state = {
"ops": {str(k): v for k, v in self.ops.items()},
}
self.cp.save(state, lsn)
self.journal.truncate_through(lsn)
self.checkpoints += 1
# -- helpers ----------------------------------------------------------
def state_digest(self) -> str:
proj = {
str(k): (
v["outcome"],
tuple(sorted((int(i), r) for i, r in v["progress"].items())),
)
for k, v in self.ops.items()
}
blob = json.dumps(
{"ops": proj, "applied": sorted(self.applied)}, sort_keys=True
)
return f"{zlib.crc32(blob.encode()):08x}"
def recover(workdir: str, capacity: int = 32) -> Engine:
return Engine(workdir, capacity)
# --------------------------------------------------------------------------
# Adversarial verification
# --------------------------------------------------------------------------
class Failure(Exception):
pass
def _mk_workdir(tag: str) -> str:
d = tempfile.mkdtemp(prefix=f"uring-{tag}-")
return d
def adversarial_trial(seed: int, n_ops: int = 60, capacity: int = 24, verbose: bool = False):
rng = random.Random(seed)
wd = _mk_workdir(str(seed))
try:
eng = Engine(wd, capacity)
# Durable log of what *should* survive: op -> final durable outcome
expected: dict[int, str] = {}
planned: dict[int, dict] = {}
for op_id in range(n_ops):
kind = rng.choice(["simple", "simple", "linked", "multishot"])
nsub = rng.randint(2, 4) if kind == "linked" else 1
effect = rng.randint(1, 1_000_000)
planned[op_id] = {"kind": kind, "nsub": nsub, "effect": effect}
eng.begin(op_id, kind, nsub, effect)
# -------- craft the kernel's "true" result for this op --------
r = rng.random()
if r < 0.60: # successful completion(s)
outcome = "done"
elif r < 0.80: # failed
outcome = "failed"
else: # cancelled (race)
outcome = "cancelled"
# -------- build the CQE list BEFORE delivery order is shuffled --
cqes: list[tuple[int, int, bool]] = [] # (index, result, more)
if outcome == "done":
if kind == "multishot":
steps = rng.randint(1, 4)
for s in range(steps):
cqes.append((s, rng.randint(1, 100), s < steps - 1))
elif kind == "linked":
for i in range(nsub):
cqes.append((i, rng.randint(1, 100), False))
else:
cqes.append((0, rng.randint(1, 100), False))
elif outcome == "failed":
if kind == "linked":
fail_at = rng.randint(0, nsub - 1)
for i in range(fail_at):
cqes.append((i, 1, False))
cqes.append((fail_at, EIO, False))
else:
cqes.append((0, EIO, False))
else: # cancelled
# Two sub-cases: cancel CQE delivered, or delivered late/never
if rng.random() < 0.5:
cqes.append((0, ECANCELED, False))
else:
eng.cancel(op_id, gen=0)
# late normal CQE must NOT be trusted after cancel? It is
# trusted only if it actually completed: model it as dropped.
# Any op whose kernel outcome is 'done' but whose CQEs are all
# dropped by the adversarial channel stays at-most-once (abort).
# -------- adversarial channel: reorder / duplicate / drop -------
delivery = list(cqes)
if rng.random() < 0.35:
rng.shuffle(delivery) # reorder
if rng.random() < 0.30 and delivery:
delivery.append(rng.choice(delivery)) # duplicate a CQE
if rng.random() < 0.25 and delivery:
delivery.pop(rng.randrange(len(delivery))) # drop a CQE
# late/stale CQE from a *previous* generation (must be ignored)
stale = rng.random() < 0.15
observed_terminal = False
for (idx, res, more) in delivery:
gen = 0
out = eng.complete(op_id, index=idx, result=res, more=more, gen=gen)
if stale and out is None and res >= 0:
pass
if out in TERMINAL:
observed_terminal = True
if stale:
eng.complete(op_id, index=99, result=1, more=False, gen=-1)
# -------- random crash + recover ---------------------------------
if rng.random() < 0.25:
if rng.random() < 0.4:
eng.storage.corrupt() # data page lost; metadata survives
eng = recover(wd, capacity)
if eng.replay_count > capacity:
raise Failure(f"replay {eng.replay_count} > capacity {capacity}")
# record what we expect to be durable: exactly the ops whose
# terminal outcome was journaled (and thus re-derivable)
op_now = eng.ops.get(op_id)
if op_now and op_now["outcome"] is not None:
expected[op_id] = op_now["outcome"]
elif op_now and op_now["cancel_pending"]:
expected[op_id] = "cancelled"
# ---------------- final crash + full recovery --------------------
if rng.random() < 0.5:
eng.storage.corrupt() # worst case: no data page survived
eng = recover(wd, capacity)
if eng.replay_count > capacity:
raise Failure(f"final replay {eng.replay_count} > capacity {capacity}")
# 1. No unresolved op may remain: every op has one terminal outcome.
for op_id, op in eng.ops.items():
if op["outcome"] not in TERMINAL:
raise Failure(f"op {op_id} left non-terminal: {op}")
# 2. Outcome uniqueness + stability across repeated recovery.
d1 = eng.state_digest()
eng2 = recover(wd, capacity)
d2 = eng2.state_digest()
if d1 != d2:
raise Failure(f"recovery not idempotent: {d1} != {d2}")
# 3. Durability: an op that was terminal before a crash must still be
# terminal (never resurrected as a different / missing outcome).
for op_id, out in expected.items():
if eng.ops[op_id]["outcome"] != out:
raise Failure(
f"durability lost op {op_id}: {eng.ops[op_id]['outcome']} != {out}"
)
# 4. Exactly-once effects: every 'done' op is applied exactly once
# (logically) and no other op is applied.
for op_id, op in eng.ops.items():
if op["outcome"] == "done":
if op_id not in eng.storage.value:
raise Failure(f"done op {op_id} not applied")
if eng.storage.value[op_id] != op["effect"]:
raise Failure(f"op {op_id} wrong effect")
if op_id not in eng.applied:
raise Failure(f"op {op_id} done but not in applied set")
else:
if op_id in eng.applied:
raise Failure(f"non-done op {op_id} was applied")
# 5. Checkpoint actually bounded the journal.
tail = len(eng.journal.records())
if tail > capacity:
raise Failure(f"journal tail {tail} > capacity {capacity}")
return {
"seed": seed,
"ops": len(eng.ops),
"checkpoints": eng.checkpoints,
"final_replay": eng.replay_count,
"physical_writes": eng.storage.physical_writes,
}
finally:
shutil.rmtree(wd, ignore_errors=True)
def _fresh(tag: str, capacity: int = 16) -> tuple[str, Engine]:
wd = _mk_workdir(tag)
return wd, Engine(wd, capacity)
def unit_tests() -> list[str]:
"""Named regression tests for each unsafe failure mode."""
passed: list[str] = []
# (1) Stale CQE from an older generation must be ignored.
wd, e = _fresh("stale")
try:
e.begin(1, "simple", 1, 111, gen=0)
e.complete(1, 0, 1, gen=-1) # stale: must be ignored
e = recover(wd)
assert e.ops[1]["outcome"] == "cancelled", e.ops[1]
passed.append("stale-generation-cqe-ignored")
finally:
shutil.rmtree(wd, ignore_errors=True)
# (2) Duplicate terminal CQE must not double-apply.
wd, e = _fresh("dup")
try:
e.begin(2, "simple", 1, 222)
e.complete(2, 0, 1)
e.complete(2, 0, 1) # duplicate
e.complete(2, 0, 1) # duplicate
e = recover(wd)
assert e.ops[2]["outcome"] == "done"
assert e.storage.value[2] == 222
assert len(e.storage.value) == 1 # one logical effect
passed.append("duplicate-terminal-cqe-idempotent")
finally:
shutil.rmtree(wd, ignore_errors=True)
# (3) Cancel race: a real completion wins; a lost completion aborts.
wd, e = _fresh("cancel")
try:
e.begin(3, "simple", 1, 333)
e.cancel(3)
e.complete(3, 0, 1) # completion beat the cancel
e = recover(wd)
assert e.ops[3]["outcome"] == "done", e.ops[3]
e.begin(4, "simple", 1, 444)
e.cancel(4)
e = recover(wd)
assert e.ops[4]["outcome"] == "cancelled", e.ops[4]
passed.append("cancel-race-resolution-order-independent")
finally:
shutil.rmtree(wd, ignore_errors=True)
# (4) Crash between data and metadata: durable COMMIT + lost data page.
wd, e = _fresh("redo")
try:
e.begin(5, "simple", 1, 555)
e.complete(5, 0, 1) # COMMIT durable; data not yet
e.storage.corrupt() # data page never hit disk
e = recover(wd)
assert e.storage.value[5] == 555, e.storage.value
passed.append("redo-after-data-loss-exactly-once")
finally:
shutil.rmtree(wd, ignore_errors=True)
# (5) Missing completion on a linked group aborts (at-most-once).
wd, e = _fresh("missing")
try:
e.begin(6, "linked", 3, 666)
e.complete(6, 0, 1)
e.complete(6, 1, 1) # link 2 never arrives
e = recover(wd)
assert e.ops[6]["outcome"] == "cancelled", e.ops[6]
assert 6 not in e.storage.value
passed.append("missing-completion-aborts-safely")
finally:
shutil.rmtree(wd, ignore_errors=True)
return passed
def replay_scaling(capacity: int = 32) -> list[tuple[int, int]]:
"""Show replay stays bounded as total history grows."""
out = []
for n in (100, 1_000, 10_000):
rng = random.Random(n)
wd = _mk_workdir(f"scale{n}")
try:
eng = Engine(wd, capacity)
max_replay = 0
for op_id in range(n):
eng.begin(op_id, "simple", 1, op_id + 1)
eng.complete(op_id, 0, 1)
if rng.random() < 0.1:
eng = recover(wd, capacity)
max_replay = max(max_replay, eng.replay_count)
eng = recover(wd, capacity)
max_replay = max(max_replay, eng.replay_count)
out.append((n, max_replay))
finally:
shutil.rmtree(wd, ignore_errors=True)
return out
def main() -> int:
print("=" * 70)
print("io_uring completion-reorder / crash-recovery verification")
print("=" * 70)
# Deterministic adversarial sweep.
for seed in range(400):
adversarial_trial(seed)
print("adversarial: 400/400 seeds passed "
"(reorder, duplicate, drop, stale, cancel-race, crash)")
# Edge cases.
for cap in (4, 8, 16, 32, 64):
for seed in range(50):
adversarial_trial(seed, n_ops=40, capacity=cap)
print("bounded-journal: capacity sweep 50 seeds each passed")
for name in unit_tests():
print(f"unit: {name}")
print("replay scaling (total ops -> max replay records):")
for n, r in replay_scaling():
print(f" N={n:>6} max_replay={r}")
print("ALL VERIFICATIONS PASSED")
return 0
if __name__ == "__main__":
raise SystemExit(main())
Run:
python3 reference.py
Observed output on Python 3.14 (the code was executed as part of producing this solution):
======================================================================
io_uring completion-reorder / crash-recovery verification
======================================================================
adversarial: 400/400 seeds passed (reorder, duplicate, drop, stale, cancel-race, crash)
bounded-journal: capacity sweep 50 seeds each passed
unit: stale-generation-cqe-ignored
unit: duplicate-terminal-cqe-idempotent
unit: cancel-race-resolution-order-independent
unit: redo-after-data-loss-exactly-once
unit: missing-completion-aborts-safely
replay scaling (total ops -> max replay records):
N= 100 max_replay=30
N= 1000 max_replay=30
N= 10000 max_replay=30
ALL VERIFICATIONS PASSED
What the harness proves:
adversarial — 400 random trials run linked, simple, and multishot ops across shuffled delivery, injected duplicates, dropped CQEs, stale generations, cancellation races, and crashes (including data-page loss). It asserts: every op is terminal, repeated recovery is byte-identical, durable outcomes are never lost, done ops are applied and nothing else is, and the journal tail never exceeds capacity.replay scaling shows max_replay is capped by journal capacity (30 < 32) even as total history grows 100×, i.e. replay is sublinear in total history.Let T be the last durable terminal record of op o.
* Uniqueness: outcomes are assigned only while outcome is None, monotonically and never reversed.
* Durability: a record is fsynced before it is acted upon; replay reproduces it.
* Exactly-once: done is exactly the set of ops with a durable success CQE; redo is idempotent on op_id, so each contributes one logical effect.
* Order independence: the transition function depends on the set of records; duplicates are no-ops and stale generations are ignored.
* Sublinear replay: checkpoint at LSN L plus journal capacity C bounds recovery to C records regardless of total operations.
user_data = (op_id << 32) | (gen << 16) | sub_index.IOSQE_IO_LINK/IOSQE_IO_HARDLINK only for atomic groups, and persist the group's op_id before the first link.IORING_CQE_F_MORE; only the CQE without it commits.IOSQE_CQE_SKIP_SUCCESS only if a later link's completion can reconstruct the skipped success.fsync the journal directory after creation; O_APPEND alone is not power-loss durable.fsync + rename + directory fsync.BASE marker) or post-checkpoint records can be dropped as stale.# Evidence - Problem class: python-uring-completion-reorder-crash-recovery-20261009 - Model: openrouter/deepseek/deepseek-v4.1-flash - Solved: 2026-10-09T22:26:34.981Z - Verification: solution produced by pi in sandbox; see signatures.json
{"description": "Model a userspace storage engine issuing linked and multishot io_uring operations across several queues where completions may arrive out of submission order, cancellation races with CQE delivery, and process death can occur between data and metadata persistence. Implement recovery that reconstructs exactly-once logical operation outcomes from a bounded journal without assuming CQE order, and prove safety under duplicate, missing, and stale completions while keeping replay sublinear in total history.", "environment": "python3", "language": "python", "model": "openrouter/deepseek/deepseek-v4.1-flash", "problem_class": "python-uring-completion-reorder-crash-recovery-20261009", "provider": "openrouter", "solved_at": "2026-10-09T22:26:34.982Z", "version": "3"}