#!/usr/bin/env python
"""Live check of optimistic versions (erp_platform.repository, `Aggregate(..., version=...)`) on DataK3.
tests/test_versions.py proves the rules on SQLite, which serialises writers. DataK3 does not: it has no
read-your-writes inside a transaction, and it decides a duplicate key at the INSERT or only at COMMIT. So this
runs the real Aggregate.save through a REAL engine (erp_platform.livecheck.engine — multi-statement
transactions, which the `dodil data pg` harness cannot give), and measures:
1. N concurrent saves from the same expected version → exactly 1 commits, N-1 are Conflict, and the stored
rows are the winner's alone (the losers' children never landed);
2. a stale save (expected < stored) → Conflict before anything is written;
3. after a reload the next save succeeds as a normal change;
4. for the record: `UPDATE … WHERE version = :v` under the same race — how many commit, and what the
losers are told (the reason the claim is an INSERT and not this).
BUCKET=erp-platform-test DODIL_ORG=dodil python scripts/live_version_check.py
"""
import os
import threading
from collections import Counter
from sqlalchemy import Column, Integer, MetaData, String, Table, TIMESTAMP, text
from erp_platform import livecheck as lc, repository
from erp_platform.repository import Aggregate, Conflict, Part
N = int(os.environ.get("THREADS", "10"))
md = MetaData()
root = Table("vcheck_person", md, Column("person_id", String, primary_key=True), Column("name", String),
Column("version", Integer), Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
kids = Table("vcheck_slice", md, Column("person_id", String, primary_key=True),
Column("slice_no", Integer, primary_key=True), Column("amount", String))
PERSON = Aggregate(Part(root, ("person_id",)), Part(kids, ("person_id", "slice_no")), version="version")
TABLES = ["vcheck_person", "vcheck_slice", "vcheck_person_versions"]
SCHEMA = """CREATE TABLE IF NOT EXISTS vcheck_person (
person_id VARCHAR NOT NULL,
name VARCHAR,
version INTEGER,
created_at TIMESTAMP,
updated_at TIMESTAMP,
PRIMARY KEY (person_id)
);
CREATE TABLE IF NOT EXISTS vcheck_slice (
person_id VARCHAR NOT NULL,
slice_no INTEGER NOT NULL,
amount VARCHAR,
PRIMARY KEY (person_id, slice_no)
);
""" + "\n".join(l for l in PERSON.version_ddl().splitlines() if not l.startswith("--"))
def rows(name, n_slices=1):
return {"vcheck_person": [{"person_id": "p1", "name": name}],
"vcheck_slice": [{"person_id": "p1", "slice_no": i, "amount": f"{name}-{i}"} for i in range(1, n_slices + 1)]}
def main() -> int:
lc.drop_all(TABLES)
for stmt in SCHEMA.split(";\n"):
if stmt.strip():
lc.pg(stmt.strip().rstrip(";"))
eng = lc.engine(pool_size=N + 2)
def save(r, v):
with repository.conflicts(), eng.begin() as c:
return PERSON.save(c, r, expected_version=v)
def load():
with eng.connect() as c:
return PERSON.load(c, {"person_id": "p1"})
lc.check("a new aggregate is saved as version 1", save(rows("first"), 0) == 1)
lc.check("load returns the version", PERSON.version_of(load()) == 1)
outcomes, gate = [], threading.Barrier(N)
def racer(i):
try:
gate.wait()
save(rows(f"w{i}", n_slices=1 + i % 3), 1)
outcomes.append(("ok", i))
except Conflict:
outcomes.append(("conflict", i))
except Exception as exc: # noqa: BLE001 — counted, and a failure
outcomes.append(("error", f"{type(exc).__name__}: {str(exc).splitlines()[0][:100]}"))
threads = [threading.Thread(target=racer, args=(i,)) for i in range(N)]
[t.start() for t in threads]
[t.join(120) for t in threads]
tally = Counter(o[0] for o in outcomes)
print(f" {N} concurrent saves at expected_version=1: {dict(tally)}")
lc.check(f"{N} concurrent saves from one version: exactly 1 commits, {N - 1} are Conflict",
tally == Counter({"ok": 1, "conflict": N - 1}), outcomes)
winner = next(i for k, i in outcomes if k == "ok")
after = load()
lc.check("the stored aggregate is the winner's alone, children included (the losers' rolled back)",
after["vcheck_person"][0]["name"] == f"w{winner}" and PERSON.version_of(after) == 2
and [r["amount"] for r in sorted(after["vcheck_slice"], key=lambda r: r["slice_no"])]
== [f"w{winner}-{j}" for j in range(1, 2 + winner % 3)], after)
claims = lc.rows("SELECT version FROM vcheck_person_versions ORDER BY version")
lc.check("one claim per version: [1, 2]", [int(r["version"]) for r in claims] == [1, 2], claims)
try:
save(rows("stale"), 1)
stale = None
except Conflict as c:
stale = c
lc.check("a stale save is Conflict, with the stored version", stale is not None and stale.stored == 2, stale)
lc.check("after a reload, the next save is a normal change",
save(rows("reloaded"), PERSON.version_of(load())) == 3 and load()["vcheck_person"][0]["name"] == "reloaded")
# For the record: why not UPDATE … WHERE version = :v.
told, upd, gate2 = [], [], threading.Barrier(N)
def updater(i):
with eng.connect() as c:
try:
gate2.wait()
with c.begin():
n = c.execute(text("UPDATE vcheck_person SET name = :n, version = 4 "
"WHERE person_id = 'p1' AND version = 3"), {"n": f"u{i}"}).rowcount
told.append(n)
upd.append("committed")
except Exception as exc: # noqa: BLE001
upd.append(f"{getattr(getattr(exc, 'orig', exc), 'sqlstate', None) or type(exc).__name__}")
threads = [threading.Thread(target=updater, args=(i,)) for i in range(N)]
[t.start() for t in threads]
[t.join(120) for t in threads]
print(f" UPDATE … WHERE version = 3, {N} racers: rowcount reported {dict(Counter(told))}, "
f"outcomes {dict(Counter(upd))}")
eng.dispose()
return lc.finish(TABLES)
if __name__ == "__main__":
raise SystemExit(main())