"""Stdlib-only tests (no pytest): plain asserts in functions called from a __main__ block. Run: `python3 test_mvcc_store.py`. The headline test pins the whole demo scenario — 52 versions, the horizon sweep freeing nothing, the precise sweep freeing 49, and the identical horizon sweep freeing 50 the instant the idle reader commits (commentary.html section 6). The rest are unit checks on the visibility rule and on the transaction lifecycle. """ import io import contextlib import demo from mvcc_store import HORIZON, PRECISE, Store, Version, retain, visible def build_scenario(): """T1 seeds a and b; R reads only a and stays live; 50 writers update b.""" store = Store() t1 = store.begin() store.write(t1, "a", "A0") store.write(t1, "b", "B0") assert store.commit(t1) == 1 r = store.begin() assert r.snapshot == 1 assert store.read(r, "a") == "A0" for i in range(1, 51): w = store.begin() store.write(w, "b", f"B{i}") store.commit(w) return store, r def test_idle_reader_pins_a_key_it_never_read(): """The headline. One idle reader turns a 52-version heap into garbage that the horizon rule may not touch — and the precise rule frees 49 of it.""" store, r = build_scenario() assert store.clock == 51 assert len(store.chains["b"]) == 51 assert store.total_versions() == 52 # R never asked for b, and still sees the value b had when it began. assert store.read(r, "b") == "B0" fresh = store.begin() assert store.read(fresh, "b") == "B50" store.commit(fresh) # R's snapshot of 1 drags the horizon down to 1, so nothing has died # "before the horizon" and the sweep frees nothing at all. assert store.horizon() == 1 assert store.snapshots() == [1, 51] assert store.vacuum(HORIZON) == 0 assert store.total_versions() == 52 # The precise rule asks each version whether ANY live snapshot sees it. # B1..B49 are invisible to both snapshot 1 and snapshot 51. assert store.vacuum(PRECISE, dry_run=True) == 49 # Commit R, change nothing else, run the identical horizon sweep again. store.commit(r) assert store.horizon() == 51 assert store.vacuum(HORIZON) == 50 assert store.total_versions() == 2 assert [v.value for v in store.chains["b"]] == ["B50"] def test_chain_hops_are_paid_by_the_stale_reader(): """This toy prepends, so distance from the head is the read cost.""" store, r = build_scenario() fresh = store.begin() assert store.probe(r, "b") == ("B0", 51) assert store.probe(fresh, "b") == ("B50", 1) def test_visibility_is_two_comparisons(): """`visible` is the whole rule: born at or before me, not yet dead to me.""" v = Version("x", xmin=5, xmax=9) assert visible(v, 4) is False # not born yet at snapshot 4 assert visible(v, 5) is True # xmin <= snap, so xmin is inclusive assert visible(v, 8) is True assert visible(v, 9) is False # xmax > snap, so xmax is exclusive live = Version("y", xmin=5, xmax=None) assert visible(live, 5) is True assert visible(live, 1000) is True def test_uncommitted_write_is_visible_only_to_its_own_transaction(): store, r = build_scenario() w = store.begin() store.write(w, "b", "B-uncommitted") assert store.read(w, "b") == "B-uncommitted" # reads its own buffer assert store.read(r, "b") == "B0" # R is untouched other = store.begin() assert store.read(other, "b") == "B50" # so is everyone else assert store.total_versions() == 52 # nothing was published def test_abort_leaves_no_trace(): """Buffered writes make abort free: there is nothing published to undo.""" store, _r = build_scenario() w = store.begin() store.write(w, "b", "B-doomed") store.abort(w) assert store.clock == 51 assert store.total_versions() == 52 assert [v.value for v in store.chains["b"][:1]] == ["B50"] def test_read_only_transaction_burns_no_commit_id(): """A transaction that wrote nothing does not advance the clock, so it cannot push the horizon forward for anyone else either.""" store, _r = build_scenario() reader = store.begin() assert store.read(reader, "a") == "A0" assert store.commit(reader) == 51 assert store.clock == 51 def test_horizon_rule_ignores_xmin_on_purpose(): """A version born after the horizon must survive: a future snapshot will want it. That is why the horizon test only looks at xmax.""" store, r = build_scenario() newest = store.chains["b"][0] horizon, snaps = store.horizon(), store.snapshots() assert newest.xmin == 51 and newest.xmax is None assert horizon == 1 assert newest.xmin > horizon # invisible to R, the only reader assert visible(newest, r.snapshot) is False assert retain(newest, HORIZON, horizon, snaps) is True assert retain(newest, PRECISE, horizon, snaps) is True def test_demo_output_is_byte_identical_across_runs(): """No clock, no RNG: the demo is a pure function of its script.""" runs = [] for _ in range(2): buf = io.StringIO() with contextlib.redirect_stdout(buf): demo.main() runs.append(buf.getvalue()) assert runs[0] == runs[1] assert "GC horizon=1 reclaims 0, leaving 52" in runs[0] assert "GC precise would reclaim 49, leaving 3" in runs[0] assert "GC horizon=51 reclaims 50, leaving 2" in runs[0] TESTS = [ test_idle_reader_pins_a_key_it_never_read, test_chain_hops_are_paid_by_the_stale_reader, test_visibility_is_two_comparisons, test_uncommitted_write_is_visible_only_to_its_own_transaction, test_abort_leaves_no_trace, test_read_only_transaction_burns_no_commit_id, test_horizon_rule_ignores_xmin_on_purpose, test_demo_output_is_byte_identical_across_runs, ] if __name__ == "__main__": failed = 0 for test in TESTS: try: test() except AssertionError as exc: failed += 1 print(f"FAIL {test.__name__}: {exc}") else: print(f"PASS {test.__name__}") if failed: print(f"\n{failed} of {len(TESTS)} tests FAILED") raise SystemExit(1) print(f"\nAll {len(TESTS)} tests PASSED")