Problem class: python-serializable-mvcc-range-predicate-recovery
I've built and verified a complete, self-contained recovery verifier. Here is the solution.
Problem class: python-serializable-mvcc-range-predicate-recovery
Runtime: Python 3.11+ · standard library only
Artifacts: /workspace/mvcc_recovery.py · /workspace/test_mvcc_recovery.py · /workspace/SOLUTION.md
The verifier must decide, from damaged evidence, which committed transactions can be kept so the history is serializable, and if none, what is the smallest witness that explains the violation. Five failure modes make a naive implementation wrong:
A truncated log is not a corrupted database. A transaction with no commit/abort record is in flight and must be rolled back. But a version already in a per-shard chain is durable proof its writer committed, even when the commit record was lost to the torn WAL tail. Treating "no commit record" uniformly as committed or aborted is wrong: the chain is the durability source of truth, while an explicit abort overrides it and forces the torn version to be rolled back.
MVCC visibility is not v.ts < snapshot under clock skew. With bounded skew ε, the true instant of stamp t lies in [t−ε, t+ε]. If a version stamp and a snapshot are within ε, both orders are possible. Collapsing them fabricates anti-dependencies (false positives) or hides real ones (false negatives). Edges must be split into definite (constrain serializability) and possible (only constrain it under an adversarial clock resolution).
Range predicates create phantom (rw) edges that point reads miss. For a read of [lo, hi] at snapshot s, every matching key whose version is definitely after s is a conflict the reader did not see — including keys that did not exist at s. Following only point reads drops the phantoms that make predicate anomalies appear.
Write skew is not a retryable transient. A cycle made purely of committed rw anti-dependencies is genuine: both transactions saw the same pre-state and each wrote what the other read; re-running either reproduces it. A conflict caused by an in-flight/torn transaction disappears once that transaction is rolled back, so its retry is safe. A verifier that reports only "cycle" conflates the two and advises an unsafe retry.
An arbitrary cycle is not actionable. Operators need (a) the shortest witness cycle and (b) the minimum set of transactions to abort so the retained history is maximal (minimum feedback vertex set / maximum induced acyclic subgraph).
Single-file library. Pipeline: parse/reconcile → build MVSG → decide on the definite committed subgraph → find shortest witness + min abort set → classify.
Edge kinds: ww (version order per key), wr (writer→reader), rw (reader→writer of an unseen version, covering range phantoms), rt (real-time commit-before-begin). The bounded-skew relation is a definitely-before b ⇔ a.ts + ε < b.ts; visibility is v.ts + ε < snap (visible), snap + ε < v.ts (invisible), else a possible edge. Classification is one of serializable, safe_retry, write_skew, unsafe_conflict, clock_uncertain.
mvcc_recovery.py#!/usr/bin/env python3
"""
mvcc_recovery.py -- Recovery verifier for a sharded MVCC store.
Problem class: python-serializable-mvcc-range-predicate-recovery
Given (a) a crash-truncated write-ahead log and (b) per-shard MVCC version
chains, this module reconstructs the maximal serializable committed history
or produces a minimal witness cycle. It also separates:
* SAFE_RETRY -- a cycle/abort caused by a transaction that never
durably committed (or a version being rolled back),
so retrying it after recovery is safe.
* WRITE_SKEW -- a cycle made only of committed read-write
anti-dependencies; retrying reproduces it.
* UNSAFE_CONFLICT -- a committed cycle involving ww/wr/real-time edges.
* CLOCK_UNCERTAIN -- no definite cycle, but a cycle appears once
timestamps that are within the bounded skew are
allowed to reorder; requires re-sync / conservative
abort rather than a blind retry.
Timestamp model
---------------
Every event carries a physical timestamp. The timestamp service is only
partially ordered with bounded skew ``epsilon``: the true instant of an event
with stamp ``t`` lies in ``[t - eps, t + eps]``. Hence
a definitely-before b <=> a.ts + eps < b.ts
a is concurrent / unknown otherwise.
Visibility of a version ``v`` at snapshot ``snap`` therefore is:
definitely visible <=> v.ts + eps < snap
definitely invisible <=> snap + eps < v.ts
otherwise => *possible* edge (tracked separately).
The committed history is serializable iff the definite multiversion
serialization graph (MVSG) over committed transactions is acyclic.
Only the Python standard library is used. Python >= 3.8.
"""
from __future__ import annotations
import argparse
import json
import sys
from collections import deque
from dataclasses import dataclass, field
from enum import Enum
from typing import (Any, Dict, FrozenSet, Iterable, List, Optional, Sequence,
Set, Tuple)
__all__ = [
"EdgeType", "Edge", "TxnStatus", "Txn", "Version", "History",
"Graph", "Diagnosis", "parse_history", "build_graph",
"minimal_witness_cycle", "minimum_feedback_vertex_set",
"diagnose", "diagnose_spec",
]
# --------------------------------------------------------------------------- #
# Total order over heterogeneous JSON keys.
# --------------------------------------------------------------------------- #
def _sort_key(k: Any) -> Tuple[int, Any]:
"""A total order that is stable across ints, floats and strings.
Numbers sort numerically and before strings; strings sort lexically.
This lets range predicates mix the two without raising TypeError.
"""
if isinstance(k, bool): # bool is an int subclass -> keep separate
return (2, int(k))
if isinstance(k, (int, float)):
return (0, float(k))
return (1, str(k))
# --------------------------------------------------------------------------- #
# Core data model
# --------------------------------------------------------------------------- #
class EdgeType(str, Enum):
WR = "wr" # write-read : writer(v) -> reader of v (happens-before)
WW = "ww" # write-write : writer(v_prev) -> writer(v_next) (order on key)
RW = "rw" # read-write : reader -> writer of a version it cannot see
RT = "rt" # real-time : committed a definitely-before begin/commit of b
#: Edge kinds that are pure anti-dependencies (used for write-skew detection).
ANTI_DEPENDENCIES = {EdgeType.RW.value}
@dataclass(frozen=True)
class Edge:
src: str
dst: str
kind: str
definite: bool = True
detail: str = ""
def __repr__(self) -> str: # pragma: no cover - cosmetic
tag = "" if self.definite else "?"
return f"{self.src}-[{tag}{self.kind}]->{self.dst}"
class TxnStatus(str, Enum):
COMMITTED = "committed"
ABORTED = "aborted"
IN_FLIGHT = "in_flight" # crash truncated before commit/abort -- retry
@dataclass
class Txn:
tid: str
status: str = TxnStatus.IN_FLIGHT.value
begin_ts: Optional[float] = None
commit_ts: Optional[float] = None
shard: Optional[str] = None
# ops are kept verbatim for auditing
ops: List[Dict[str, Any]] = field(default_factory=list)
@property
def committed(self) -> bool:
return self.status == TxnStatus.COMMITTED.value
@property
def snapshot(self) -> Optional[float]:
return self.begin_ts
def to_dict(self) -> Dict[str, Any]:
return {
"tid": self.tid,
"status": self.status,
"begin_ts": self.begin_ts,
"commit_ts": self.commit_ts,
"shard": self.shard,
}
@dataclass
class Version:
shard: str
key: Any
value: Any
writer: str
ts: float
ver: int = 0
rolled_back: bool = False
def to_dict(self) -> Dict[str, Any]:
return {
"shard": self.shard,
"key": self.key,
"value": self.value,
"writer": self.writer,
"ts": self.ts,
"ver": self.ver,
"rolled_back": self.rolled_back,
}
@dataclass
class ReadOp:
txn: str
keys: List[Any] # explicitly read point keys
ranges: List[Tuple[Any, Any, bool, bool]] # (lo, hi, lo_incl, hi_incl)
ts: Optional[float] = None # snapshot used for the read (defaults to begin)
@dataclass
class History:
epsilon: float
txns: Dict[str, Txn]
chain: Dict[str, Dict[Any, List[Version]]]
reads: List[ReadOp]
truncation_index: Optional[int] = None
warnings: List[str] = field(default_factory=list)
# ---------------------------------------------------------------- helpers
def committed_txns(self) -> Set[str]:
return {t for t, x in self.txns.items() if x.committed}
def all_versions(self) -> List[Version]:
out: List[Version] = []
for shard in self.chain.values():
for versions in shard.values():
out.extend(versions)
return out
def versions_for(self, key: Any) -> List[Version]:
out: List[Version] = []
for shard in self.chain.values():
out.extend(shard.get(key, ()))
# live (non rolled back) committed versions only
return sorted(
(v for v in out if not v.rolled_back),
key=lambda v: (_sort_key(v.ts), v.ver),
)
# --------------------------------------------------------------------------- #
# Graph
# --------------------------------------------------------------------------- #
@dataclass
class Graph:
nodes: Set[str] = field(default_factory=set)
edges: List[Edge] = field(default_factory=list)
def add(self, edge: Edge) -> None:
if edge.src == edge.dst:
return
self.nodes.add(edge.src)
self.nodes.add(edge.dst)
if edge not in self.edges:
self.edges.append(edge)
def adjacency(self, edge_filter=None) -> Dict[str, List[Edge]]:
adj: Dict[str, List[Edge]] = {n: [] for n in self.nodes}
for e in self.edges:
if edge_filter is not None and not edge_filter(e):
continue
adj.setdefault(e.src, []).append(e)
adj.setdefault(e.dst, [])
return adj
def subgraph(self, nodes: Iterable[str]) -> "Graph":
keep = set(nodes)
g = Graph(nodes=set(keep))
for e in self.edges:
if e.src in keep and e.dst in keep:
g.edges.append(e)
return g
def to_dict(self) -> Dict[str, Any]:
return {
"nodes": sorted(self.nodes),
"edges": [
{"src": e.src, "dst": e.dst, "kind": e.kind,
"definite": e.definite, "detail": e.detail}
for e in self.edges
],
}
# --------------------------------------------------------------------------- #
# Parsing / recovery
# --------------------------------------------------------------------------- #
def _is_number(x: Any) -> bool:
return isinstance(x, (int, float)) and not isinstance(x, bool)
def parse_history(spec: Dict[str, Any]) -> History:
"""Parse an input specification into a :class:`History`.
Expected schema (all timestamps are numbers)::
{
"epsilon": 5,
"log": [
{"txn": "T1", "op": "begin", "ts": 100},
{"txn": "T1", "op": "read", "key": "a"},
{"txn": "T1", "op": "read_range", "lo": "a", "hi": "c",
"lo_incl": true, "hi_incl": false},
{"txn": "T1", "op": "write", "key": "b", "value": 1, "ts": 104},
{"txn": "T1", "op": "commit", "ts": 105}
],
"chains": {
"s0": {
"a": [{"value": 0, "writer": "T0", "ts": 50, "ver": 1}]
}
}
}
A record missing ``txn`` or ``op`` terminates the log: everything from
that point on is discarded (crash truncation). The index is reported so
the operator knows where the WAL was torn.
"""
epsilon = float(spec.get("epsilon", 0.0))
txns: Dict[str, Txn] = {}
reads: List[ReadOp] = []
read_idx: Dict[Tuple[str, float], ReadOp] = {}
truncation_index: Optional[int] = None
warnings: List[str] = []
log = spec.get("log", [])
for i, rec in enumerate(log):
if not isinstance(rec, dict) or "txn" not in rec or "op" not in rec:
truncation_index = i
warnings.append(f"torn log record at index {i}: {rec!r}")
break
tid = str(rec["txn"])
op = str(rec["op"]).lower()
t = txns.setdefault(tid, Txn(tid=tid))
t.ops.append(rec)
if rec.get("shard") is not None:
t.shard = rec.get("shard")
if op == "begin":
t.begin_ts = float(rec["ts"]) if _is_number(rec.get("ts")) else t.begin_ts
elif op == "commit":
t.commit_ts = float(rec["ts"]) if _is_number(rec.get("ts")) else t.commit_ts
t.status = TxnStatus.COMMITTED.value
elif op == "abort":
t.status = TxnStatus.ABORTED.value
if _is_number(rec.get("ts")):
t.commit_ts = float(rec["ts"])
elif op in ("read", "read_range", "scan"):
ts = float(rec["ts"]) if _is_number(rec.get("ts")) else None
# group reads by (txn, ts) so one ReadOp holds a whole predicate
key = (tid, ts if ts is not None else -1.0)
ro = read_idx.get(key)
if ro is None:
ro = ReadOp(txn=tid, keys=[], ranges=[], ts=ts)
read_idx[key] = ro
reads.append(ro)
if op in ("read_range", "scan"):
lo = rec.get("lo", rec.get("from"))
hi = rec.get("hi", rec.get("to"))
lo_incl = bool(rec.get("lo_incl", True))
hi_incl = bool(rec.get("hi_incl", False))
ro.ranges.append((lo, hi, lo_incl, hi_incl))
else:
if "key" in rec:
ro.keys.append(rec["key"])
elif op == "write":
if _is_number(rec.get("ts")):
t.commit_ts = max(t.commit_ts or float("-inf"), float(rec["ts"]))
else:
warnings.append(f"unknown op {op!r} for txn {tid} at index {i}")
# ---- version chains ----------------------------------------------------
chain: Dict[str, Dict[Any, List[Version]]] = {}
for shard, keys in (spec.get("chains") or {}).items():
chain.setdefault(shard, {})
for key, versions in keys.items():
bucket = chain[shard].setdefault(key, [])
for v in versions:
writer = str(v.get("writer", v.get("txn", "?")))
ts = float(v.get("ts", v.get("commit_ts", 0.0)))
ver = Version(
shard=shard, key=key, value=v.get("value"),
writer=writer, ts=ts, ver=int(v.get("ver", 0)),
)
bucket.append(ver)
# A version that is present in a durable chain proves its
# writer committed, even if the commit record was lost to the
# crash truncation. An explicit abort is stronger evidence
# and forces a rollback.
w = txns.get(writer)
if w is None:
w = txns.setdefault(writer, Txn(tid=writer))
w.status = TxnStatus.COMMITTED.value
w.commit_ts = ts
warnings.append(
f"recovered commit of {writer} from chain "
f"{shard}/{key}@{ts}")
elif w.status == TxnStatus.ABORTED.value:
ver.rolled_back = True
warnings.append(
f"rolled back torn version {shard}/{key}@{ts} "
f"from aborted txn {writer}")
elif w.status == TxnStatus.IN_FLIGHT.value:
w.status = TxnStatus.COMMITTED.value
w.commit_ts = max(w.commit_ts or float("-inf"), ts)
warnings.append(
f"recovered commit of {writer} from chain "
f"{shard}/{key}@{ts}")
# ensure every txn that only appears as a reader is registered
for ro in reads:
txns.setdefault(ro.txn, Txn(tid=ro.txn))
# fill missing snapshots with the transaction's begin/commit ts
for ro in reads:
if ro.ts is None:
t = txns.get(ro.txn)
if t is not None:
ro.ts = t.begin_ts if t.begin_ts is not None else t.commit_ts
# a transaction that has written versions but no begin_ts is still usable
hist = History(epsilon=epsilon, txns=txns, chain=chain, reads=reads,
truncation_index=truncation_index, warnings=warnings)
return hist
# --------------------------------------------------------------------------- #
# Predicate matching
# --------------------------------------------------------------------------- #
def _matches(key: Any, rng: Tuple[Any, Any, bool, bool]) -> bool:
lo, hi, lo_incl, hi_incl = rng
kk = _sort_key(key)
if lo is not None:
c = (kk > _sort_key(lo)) or (lo_incl and kk == _sort_key(lo))
if not c:
return False
if hi is not None:
c = (kk < _sort_key(hi)) or (hi_incl and kk == _sort_key(hi))
if not c:
return False
return True
def _all_keys(hist: History) -> List[Any]:
keys: List[Any] = []
seen = set()
for shard in hist.chain.values():
for k in shard.keys():
sk = _sort_key(k)
if sk not in seen:
seen.add(sk)
keys.append(k)
return keys
# --------------------------------------------------------------------------- #
# MVSG construction
# --------------------------------------------------------------------------- #
class Relation(str, Enum):
DEFINITE = "definite"
POSSIBLE = "possible" # only under some bounded-skew resolution
def _relation(a_ts: float, b_ts: float, eps: float) -> Relation:
"""Temporal relation of event a to event b under skew eps."""
if a_ts + eps < b_ts:
return Relation.DEFINITE # a before b
return Relation.POSSIBLE # concurrent / unknown
def build_graph(hist: History) -> Graph:
"""Construct the multiversion serialization graph (MVSG).
Definite edges form the MVSG used for the serializability decision.
*Possible* edges capture conflicts that exist only if clock skew is
resolved unfavourably; they never make a history non-serializable on
their own but are surfaced as ``CLOCK_UNCERTAIN``.
"""
eps = hist.epsilon
g = Graph()
for tid in hist.txns:
g.nodes.add(tid)
committed = {t for t in hist.txns if hist.txns[t].committed}
# ---- ww edges: order versions of the same key -------------------------
for shard in hist.chain.values():
for key, versions in shard.items():
live = sorted((v for v in versions if not v.rolled_back),
key=lambda v: (_sort_key(v.ts), v.ver))
for a, b in zip(live, live[1:]):
rel = _relation(a.ts, b.ts, eps)
g.add(Edge(a.writer, b.writer, EdgeType.WW.value,
definite=(rel is Relation.DEFINITE),
detail=f"{key}: {a.ts}->{b.ts}"))
# ---- reads: wr and rw edges ------------------------------------------
for ro in hist.reads:
reader = ro.txn
rt = hist.txns.get(reader)
if rt is None:
continue
snap = ro.ts
if snap is None:
snap = rt.begin_ts if rt.begin_ts is not None else rt.commit_ts
if snap is None:
continue
# collect the key set touched by this read
read_keys: List[Any] = list(ro.keys)
if ro.ranges:
for k in _all_keys(hist):
if any(_matches(k, r) for r in ro.ranges):
read_keys.append(k)
per_key: Dict[Any, List[Version]] = {}
for k in read_keys:
per_key.setdefault(_sort_key(k), hist.versions_for(k))
for kk, versions in per_key.items():
# exclude the reader's own versions: a txn sees its own writes
others = [v for v in versions if v.writer != reader]
if not others:
continue
visible = [v for v in others if v.ts + eps < snap]
invisible = [v for v in others if snap + eps < v.ts]
uncertain = [
v for v in others
if not (v.ts + eps < snap) and not (snap + eps < v.ts)
]
# the version actually read: latest visible; if none, "empty"
if visible:
vr = max(visible, key=lambda v: (_sort_key(v.ts), v.ver))
if vr.writer in g.nodes:
g.add(Edge(vr.writer, reader, EdgeType.WR.value,
detail=f"read {vr.key}@{vr.ts}"))
# anti-dependency: reader -> first writer it must not have seen
if invisible:
nxt = min(invisible, key=lambda v: (_sort_key(v.ts), v.ver))
g.add(Edge(reader, nxt.writer, EdgeType.RW.value,
detail=f"missed {nxt.key}@{nxt.ts}"))
# skew-uncertain conflicts -> possible edges only
for v in uncertain:
if v.ts <= snap:
# could have been visible -> possible wr
g.add(Edge(v.writer, reader, EdgeType.WR.value,
definite=False,
detail=f"skew vis {v.key}@{v.ts} vs {snap}"))
else:
# could have been invisible -> possible rw
g.add(Edge(reader, v.writer, EdgeType.RW.value,
definite=False,
detail=f"skew inv {v.key}@{v.ts} vs {snap}"))
# ---- real-time edges among committed transactions ---------------------
committed_sorted = sorted(
(hist.txns[t] for t in committed),
key=lambda t: (t.begin_ts if t.begin_ts is not None else float("-inf")),
)
for a in committed_sorted:
for b in committed_sorted:
if a.tid == b.tid:
continue
a_commit = a.commit_ts
b_begin = b.begin_ts
if a_commit is None or b_begin is None:
continue
if a_commit + eps < b_begin:
g.add(Edge(a.tid, b.tid, EdgeType.RT.value,
detail=f"{a_commit}<{b_begin}"))
# never emit edges pointing at/from rolled-back or aborted writers as
# definite serialization constraints
cleaned: List[Edge] = []
for e in g.edges:
if e.definite and (e.src not in committed or e.dst not in committed):
# a committed transaction can depend on an uncommitted writer only
# through a torn version, which we already roll back; downgrade
e = Edge(e.src, e.dst, e.kind, definite=False, detail=e.detail)
cleaned.append(e)
g.edges = []
for e in cleaned:
g.add(e)
return g
# --------------------------------------------------------------------------- #
# Cycle machinery
# --------------------------------------------------------------------------- #
def _adj(g: Graph, nodes: Set[str], definite_only: bool = True
) -> Dict[str, List[str]]:
adj: Dict[str, List[str]] = {n: [] for n in nodes}
for e in g.edges:
if definite_only and not e.definite:
continue
if e.src in nodes and e.dst in nodes:
adj.setdefault(e.src, []).append(e.dst)
return adj
def _find_cycle_nodes(g: Graph, nodes: Set[str],
definite_only: bool = True) -> Optional[List[str]]:
"""Return *some* elementary cycle (as a node list) or None."""
adj = _adj(g, nodes, definite_only)
WHITE, GREY, BLACK = 0, 1, 2
color = {n: WHITE for n in nodes}
stack: List[str] = []
for start in nodes:
if color[start] != WHITE:
continue
# iterative DFS with explicit frames
frames: List[Tuple[str, int]] = [(start, 0)]
color[start] = GREY
stack = [start]
while frames:
u, i = frames[-1]
if i < len(adj.get(u, ())):
frames[-1] = (u, i + 1)
v = adj[u][i]
if color.get(v, WHITE) == WHITE:
color[v] = GREY
stack.append(v)
frames.append((v, 0))
elif color.get(v) == GREY:
idx = stack.index(v)
return stack[idx:]
else:
color[u] = BLACK
stack.pop()
frames.pop()
return None
def is_acyclic(g: Graph, nodes: Optional[Set[str]] = None,
definite_only: bool = True) -> bool:
nodes = set(g.nodes) if nodes is None else set(nodes)
return _find_cycle_nodes(g, nodes, definite_only) is None
def minimal_witness_cycle(g: Graph, nodes: Optional[Set[str]] = None,
definite_only: bool = True) -> List[str]:
"""Minimum-length elementary cycle, or [] when acyclic.
For every edge u->v, the shortest path v->u (BFS) closes a cycle; take
the global shortest. Polynomial and exact for a minimal witness.
"""
nodes = set(g.nodes) if nodes is None else set(nodes)
adj = _adj(g, nodes, definite_only)
best: List[str] = []
def shortest_path(s: str, t: str) -> Optional[List[str]]:
if s == t:
return [s]
prev: Dict[str, str] = {s: s}
dq = deque([s])
while dq:
u = dq.popleft()
for v in adj.get(u, ()):
if v in prev:
continue
prev[v] = u
if v == t:
path = [t]
while path[-1] != s:
path.append(prev[path[-1]])
path.reverse()
return path
dq.append(v)
return None
for e in g.edges:
if definite_only and not e.definite:
continue
if e.src not in nodes or e.dst not in nodes:
continue
path = shortest_path(e.dst, e.src)
if path is None:
continue
cycle = [e.src] + path # e.src -> e.dst -> ... -> e.src
if not best or len(cycle) < len(best):
best = cycle
return best
def _cycle_edges(g: Graph, cycle: List[str],
definite_only: bool = True) -> List[Edge]:
if not cycle:
return []
out: List[Edge] = []
for a, b in zip(cycle, cycle[1:] + cycle[:1]):
for e in g.edges:
if e.src == a and e.dst == b:
if definite_only and not e.definite:
continue
out.append(e)
break
return out
def minimum_feedback_vertex_set(g: Graph,
nodes: Optional[Set[str]] = None,
definite_only: bool = True,
max_exact: int = 24) -> Set[str]:
"""Smallest set of committed transactions to abort to make the graph
acyclic (a.k.a. maximum induced acyclic subgraph).
Exact branch-and-bound for up to ``max_exact`` nodes; a greedy
degree heuristic beyond that. Every cycle must lose at least one vertex,
so branching on the vertices of a discovered cycle is complete.
"""
nodes = set(g.nodes) if nodes is None else set(nodes)
if is_acyclic(g, nodes, definite_only):
return set()
if len(nodes) > max_exact:
return _greedy_fvs(g, nodes, definite_only)
from functools import lru_cache
@lru_cache(maxsize=None)
def solve(active: FrozenSet[str]) -> FrozenSet[str]:
cyc = _find_cycle_nodes(g, set(active), definite_only)
if cyc is None:
return frozenset()
best: Optional[FrozenSet[str]] = None
for v in cyc:
rem = solve(active - {v})
cand = rem | {v}
if best is None or len(cand) < len(best):
best = cand
assert best is not None
return best
return set(solve(frozenset(nodes)))
def _greedy_fvs(g: Graph, nodes: Set[str], definite_only: bool) -> Set[str]:
active = set(nodes)
removed: Set[str] = set()
adj = _adj(g, active, definite_only)
while True:
cyc = _find_cycle_nodes(g, active, definite_only)
if cyc is None:
return removed
# pick the cycle vertex with the highest total degree
def deg(v: str) -> int:
out = len([w for w in adj.get(v, ()) if w in active])
inn = len([w for w, ws in adj.items() if v in ws and w in active])
return out + inn
victim = max(cyc, key=deg)
active.discard(victim)
removed.add(victim)
# --------------------------------------------------------------------------- #
# Diagnosis
# --------------------------------------------------------------------------- #
class Classification(str, Enum):
SERIALIZABLE = "serializable"
SAFE_RETRY = "safe_retry"
WRITE_SKEW = "write_skew"
UNSAFE_CONFLICT = "unsafe_conflict"
CLOCK_UNCERTAIN = "clock_uncertain"
@dataclass
class Diagnosis:
classification: str
serializable: bool
committed: List[str]
kept: List[str]
aborted: List[str] # transactions to roll back / retry
witness_cycle: List[str]
witness_edges: List[Dict[str, Any]]
possible_cycle: List[str]
in_flight: List[str]
truncation_index: Optional[int]
warnings: List[str]
def to_dict(self) -> Dict[str, Any]:
return {
"classification": self.classification,
"serializable": self.serializable,
"committed": self.committed,
"kept": self.kept,
"aborted": self.aborted,
"witness_cycle": self.witness_cycle,
"witness_edges": self.witness_edges,
"possible_cycle": self.possible_cycle,
"in_flight": self.in_flight,
"truncation_index": self.truncation_index,
"warnings": self.warnings,
}
def _classify(hist: History, witness: List[str], witness_edges: List[Edge],
possible_cycle: List[str]) -> str:
if not witness:
if possible_cycle:
return Classification.CLOCK_UNCERTAIN.value
return Classification.SERIALIZABLE.value
# A cycle that involves an in-flight / aborted participant, or whose edge
# is only possible, is recoverable by rollback or re-sync.
if any(hist.txns.get(t) is None or not hist.txns[t].committed
for t in witness):
return Classification.SAFE_RETRY.value
if any(not e.definite for e in witness_edges):
return Classification.CLOCK_UNCERTAIN.value
kinds = {e.kind for e in witness_edges}
if kinds and kinds <= ANTI_DEPENDENCIES and len(witness) >= 3:
# elementary cycle list includes the closing node, e.g. [T1,T2,T1]
return Classification.WRITE_SKEW.value
return Classification.UNSAFE_CONFLICT.value
def diagnose(hist: History) -> Diagnosis:
g = build_graph(hist)
committed = hist.committed_txns()
in_flight = sorted(t for t, x in hist.txns.items()
if x.status == TxnStatus.IN_FLIGHT.value)
# definite MVSG restricted to committed transactions
definite_committed = set(committed)
witness = minimal_witness_cycle(g, definite_committed, definite_only=True)
witness_edges = _cycle_edges(g, witness, definite_only=True)
# possible (skew-dependent) cycle
possible_cycle: List[str] = []
if not witness:
possible_cycle = minimal_witness_cycle(g, definite_committed,
definite_only=False)
classification = _classify(hist, witness, witness_edges, possible_cycle)
serializable = classification == Classification.SERIALIZABLE.value
# Transactions to roll back: every in-flight transaction (it never
# durably committed) plus a minimum feedback vertex set of the committed
# MVSG when a definite cycle exists.
aborted = set(in_flight)
if witness:
aborted |= minimum_feedback_vertex_set(g, definite_committed,
definite_only=True)
kept = sorted(committed - aborted)
return Diagnosis(
classification=classification,
serializable=serializable,
committed=sorted(committed),
kept=kept,
aborted=sorted(aborted),
witness_cycle=witness,
witness_edges=[
{"src": e.src, "dst": e.dst, "kind": e.kind,
"definite": e.definite, "detail": e.detail}
for e in witness_edges
],
possible_cycle=possible_cycle,
in_flight=in_flight,
truncation_index=hist.truncation_index,
warnings=hist.warnings,
)
def diagnose_spec(spec: Dict[str, Any]) -> Diagnosis:
return diagnose(parse_history(spec))
# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #
def _main(argv: Optional[Sequence[str]] = None) -> int:
ap = argparse.ArgumentParser(description=__doc__.splitlines()[1])
ap.add_argument("path", nargs="?",
help="JSON history file (default: stdin)")
ap.add_argument("--demo", action="store_true",
help="run the built-in verification scenarios")
ap.add_argument("--graph", action="store_true",
help="also dump the MVSG")
args = ap.parse_args(argv)
if args.demo:
return demo()
if args.path:
with open(args.path, "r", encoding="utf-8") as fh:
spec = json.load(fh)
else:
spec = json.load(sys.stdin)
hist = parse_history(spec)
result = diagnose(hist)
out: Dict[str, Any] = {"diagnosis": result.to_dict()}
if args.graph:
out["graph"] = build_graph(hist).to_dict()
json.dump(out, sys.stdout, indent=2, sort_keys=True, default=str)
sys.stdout.write("\n")
return 0
Built-in verification scenarios. Append the scenario block below (this is exactly what is in /workspace/mvcc_recovery.py; it drives python3 mvcc_recovery.py --demo). The eight specs cover serial chains, write skew, torn/in-flight writers, aborted torn rollback, phantom predicate reads, clock uncertainty, mixed ww+rw cycles, and a three-way write-skew needing two aborts. (Full text in the repository file; representative excerpt below.)
# 2. Classic write skew: T1 reads {a,b} writes a; T2 reads {a,b} writes b.
_scenario(
"write_skew",
{
"epsilon": 1,
"log": [
{"txn": "T0", "op": "begin", "ts": 1},
{"txn": "T0", "op": "write", "key": "a", "value": 0, "ts": 2},
{"txn": "T0", "op": "write", "key": "b", "value": 0, "ts": 2},
{"txn": "T0", "op": "commit", "ts": 2},
{"txn": "T1", "op": "begin", "ts": 10},
{"txn": "T1", "op": "read", "key": "a"},
{"txn": "T1", "op": "read", "key": "b"},
{"txn": "T1", "op": "write", "key": "a", "value": 1, "ts": 14},
{"txn": "T1", "op": "commit", "ts": 14},
{"txn": "T2", "op": "begin", "ts": 11},
{"txn": "T2", "op": "read", "key": "a"},
{"txn": "T2", "op": "read", "key": "b"},
{"txn": "T2", "op": "write", "key": "b", "value": 1, "ts": 15},
{"txn": "T2", "op": "commit", "ts": 15},
],
"chains": {
"s0": {
"a": [{"value": 0, "writer": "T0", "ts": 2, "ver": 1},
{"value": 1, "writer": "T1", "ts": 14, "ver": 2}],
"b": [{"value": 0, "writer": "T0", "ts": 2, "ver": 1},
{"value": 1, "writer": "T2", "ts": 15, "ver": 2}],
}
},
},
Classification.WRITE_SKEW.value,
)
All commands were run in this environment.
$ ruff check mvcc_recovery.py test_mvcc_recovery.py
All checks passed!
$ python3 -m py_compile mvcc_recovery.py test_mvcc_recovery.py
(no output)
python3 mvcc_recovery.py --demo[PASS] serial_chain expected=serializable got=serializable
witness=[] aborted=[] kept=['T0', 'T1', 'T2'] possible=[]
! recovered commit of T0 from chain s0/<email>.0
[PASS] write_skew expected=write_skew got=write_skew
witness=['T1', 'T2', 'T1'] aborted=['T1'] kept=['T0', 'T2'] possible=[]
[PASS] torn_write_safe_retry expected=serializable got=serializable
witness=[] aborted=['T3'] kept=['T0', 'T1'] possible=[]
! recovered commit of T0 from chain s0/<email>.0
[PASS] aborted_torn_rollback expected=serializable got=serializable
witness=[] aborted=[] kept=['T0'] possible=[]
! recovered commit of T0 from chain s0/<email>.0
! rolled back torn version s0/<email>.0 from aborted txn T1
[PASS] phantom_predicate expected=write_skew got=write_skew
witness=['T1', 'T2', 'T1'] aborted=['T1'] kept=['T0', 'T2'] possible=[]
! recovered commit of T0 from chain s0/<email>.0
[PASS] clock_uncertain expected=clock_uncertain got=clock_uncertain
witness=[] aborted=[] kept=['T0', 'T1', 'T2'] possible=['T1', 'T2', 'T1']
[PASS] mixed_ww_rw_conflict expected=unsafe_conflict got=unsafe_conflict
witness=['T1', 'T3', 'T1'] aborted=['T1'] kept=['T0', 'T2', 'T3'] possible=[]
[PASS] three_way_write_skew expected=write_skew got=write_skew
witness=['T1', 'T2', 'T1'] aborted=['T1', 'T2'] kept=['T0', 'T3'] possible=[]
8/8 scenarios pass
| Scenario | What it proves |
|---|---|
serial_chain |
clean ww/wr/rt chain is serializable and unchanged |
write_skew |
committed all-rw 2-cycle → write_skew, min abort = 1 |
torn_write_safe_retry |
in-flight writer aborted; committed part stays serializable |
aborted_torn_rollback |
an aborted writer's durable version is rolled back |
phantom_predicate |
range read + insert closes an rw cycle (phantom) |
clock_uncertain |
no definite cycle, but an adversarial clock creates one |
mixed_ww_rw_conflict |
committed ww+rw cycle → unsafe_conflict |
three_way_write_skew |
complete anti-dependency graph needs 2 aborts (max kept) |
python3 -m unittest test_mvcc_recovery......................................................................
----------------------------------------------------------------------
Ran 22 tests in 0.105s
OK
Highlights of the 22 tests:
test_kept_is_maximum_induced_acyclic_subgraph brute-forces every subset of the committed transactions for three scenarios and asserts the reported kept set has the maximum acyclic size (not merely a valid one).test_fvs_complete_digraph_needs_n_minus_1 checks the exact minimum feedback vertex set on a complete digraph.test_minimal_cycle_is_shortest checks the witness is the shortest cycle.test_clock_uncertain_has_no_definite_cycle checks the definite MVSG is acyclic while a possible cycle is still reported.test_chain_proves_commit_when_log_truncated and test_abort_rolls_back_chain_version pin the reconciliation rules.test_fvs_exact_vs_greedy_agree_on_greedy_limit and test_possible_edges_ignored_for_definite_acyclicity check the fallback and the skew split.--graph too).ww/wr/rw conflicts; rt edges make it strict.a.ts + ε < b.ts), so no serial order exists.minimal_witness_cycle minimizes 1 + dist(v, u) over every edge u→v, hence returns the globally shortest cycle.max_exact=24; beyond that a greedy loop removes one high-degree cycle vertex per step.python3 mvcc_recovery.py --demo # 8/8 scenarios must pass
python3 -m unittest -v test_mvcc_recovery # 22 tests must pass
python3 mvcc_recovery.py --graph history.json # diagnose a real log
Input is a JSON object with epsilon, a log of {txn, op, ...} records (begin, read, read_range, write, commit, abort) and per-shard chains of {value, writer, ts, ver} versions. Output is a JSON diagnosis with classification, committed, kept (the maximal serializable history), aborted (roll back / safe-retry candidates), witness_cycle + witness_edges, possible_cycle, in_flight, and truncation_index.
The final artifact set is on disk: /workspace/mvcc_recovery.py, /workspace/test_mvcc_recovery.py, and the assembled /workspace/SOLUTION.md.
# Evidence - Problem class: python-serializable-mvcc-range-predicate-recovery - Model: openrouter/deepseek/deepseek-v4.1-flash - Solved: 2026-10-10T16:34:35.881Z - Verification: solution produced by pi in sandbox; see signatures.json
{"description": "Implement a recovery verifier for a sharded MVCC store whose transactions may read key ranges, write keys, and commit under a partially ordered timestamp service with bounded clock skew. Given a crash-truncated log and per-shard version chains, reconstruct the maximal serializable committed history or produce a minimal witness cycle, while distinguishing safe retries from write-skew anomalies.", "environment": "python3", "language": "python", "model": "openrouter/deepseek/deepseek-v4.1-flash", "problem_class": "python-serializable-mvcc-range-predicate-recovery", "provider": "openrouter", "solved_at": "2026-10-10T16:34:35.882Z", "version": "3.11"}