#!/usr/bin/env python
"""Live check of the Idempotency-Key mechanics (erp_platform.idempotency, 0.5) on a real DataK3 bucket.
tests/test_hardening.py proves the rules on SQLite; the rules rest on what the ENGINE does with a plain INSERT,
so this runs the real Store and `begin` against DataK3:
1. N threads, one key, every one reading "no claim yet" (a stale snapshot) → exactly one may run;
2. the answer stored on the claim is replayed, and the same key with another request is refused 422;
3. a release (DELETE … WHERE status IS NULL) frees the key for a retry, and leaves an answered one alone;
4. a claim whose lease expired is taken over with generation n+1, with one winner among racers.
BUCKET=erp-platform-test DODIL_ORG=dodil python scripts/live_idempotency_check.py
Every statement goes through `dodil data pg` (erp_platform.livecheck), one statement per transaction, as the
Store's operations are. TIMESTAMPs come back from the CLI in the machine's zone and are made naive UTC here.
"""
import tempfile
import threading
import time
from contextlib import contextmanager
from datetime import datetime, timezone
from erp_platform import idempotency, livecheck as lc
TABLES = [idempotency.claims.name]
N = int(__import__("os").environ.get("THREADS", "10"))
UC, WHO = "check.add_note", "live-check"
class Conn(lc.Connection):
def execute(self, stmt):
res = super().execute(stmt)
for row in res:
for k, v in list(row.items()):
if isinstance(v, datetime) and v.tzinfo:
dict.__setitem__(row, k, v.astimezone(timezone.utc).replace(tzinfo=None))
return res
@contextmanager
def connect():
yield Conn()
STORE = idempotency.Store(connect)
class Blind(idempotency.Store):
"""Every read sees no claim: the plain INSERT is then the only guard (what a racer's snapshot can show)."""
def rows(self, *a):
return []
def race(store, key, digest, n):
verdicts, errs = [], []
def go():
try:
verdicts.append(idempotency.begin(store, WHO, UC, key, digest)[0])
except idempotency.KeyRefused as e:
verdicts.append(e.status)
except Exception as e: # noqa: BLE001 — counted, reported
errs.append(f"{type(e).__name__}: {str(e)[-160:]}")
ts = [threading.Thread(target=go) for _ in range(n)]
for t in ts:
t.start()
for t in ts:
t.join()
return verdicts, errs
def refused(fn):
"""The KeyRefused status, or None if it was not refused."""
try:
fn()
except idempotency.KeyRefused as e:
return e.status
return None
def claims(key):
return lc.rows(f"SELECT generation, status FROM {idempotency.claims.name} "
f"WHERE use_case = '{UC}' AND idempotency_key = '{key}' ORDER BY generation")
def main() -> int:
if not lc.BUCKET:
raise SystemExit("set BUCKET=erp-platform-test DODIL_ORG=dodil")
# The table exactly as platform.sql declares it — its NOT NULLs included: DataK3 checks them on an upsert's
# VALUES even when the row conflicts, which the code's own MetaData did not say until 0.5.3 (crm-demo, 2026-10-17).
from erp_platform import migrate
sql = migrate.platform_schema().read_text()
block = sql[sql.index("CREATE TABLE IF NOT EXISTS erp_idempotency"):]
with tempfile.NamedTemporaryFile("w", suffix=".sql", delete=False) as f:
f.write(block[:block.index(");") + 2] + "\n")
lc.create_schema(f.name, TABLES)
d1 = idempotency.request_hash(UC, {"text": "hello"})
v, errs = race(Blind(connect), "race-1", d1, N)
print(f" blind race: {N} threads → {v.count('run')} run, {v.count(409)} told 409, errors {errs[:2]}")
lc.check(f"claim: {N} racers that all saw no claim → exactly one runs", v.count("run") == 1 and not errs,
{"verdicts": v, "errs": errs[:2]})
won = lc.rows(f"SELECT * FROM {idempotency.claims.name} WHERE idempotency_key = 'race-1'")[0]
row = {k: won[k] for k in ("caller", "use_case", "idempotency_key", "generation", "request_hash")}
row |= {k: datetime.fromisoformat(str(won[k])).astimezone(timezone.utc).replace(tzinfo=None)
for k in ("claimed_at", "lease_until")}
STORE.finish(row, {"note_id": "N-1", "text": "hello"})
verdict, row = idempotency.begin(STORE, WHO, UC, "fin-1", d1)
STORE.finish(row, {"note_id": "N-2"}) # the row begin hands back is the WHOLE claim
lc.check("finish: the answer is stored on the claim begin returned (platform.sql's NOT NULLs hold)",
claims("fin-1") == [{"generation": 1, "status": 200}], claims("fin-1"))
verdict, answer = idempotency.begin(STORE, WHO, UC, "race-1", d1)
lc.check("replay: the stored answer comes back", verdict == "replay" and answer["note_id"] == "N-1",
(verdict, answer))
lc.check("replay: the same key with another request is 422",
refused(lambda: idempotency.begin(STORE, WHO, UC, "race-1",
idempotency.request_hash(UC, {"text": "other"}))) == 422)
verdict, row = idempotency.begin(STORE, WHO, UC, "rel-1", d1)
STORE.release(WHO, UC, "rel-1", row["generation"])
lc.check("release: a refused request leaves no claim", verdict == "run" and claims("rel-1") == [], claims("rel-1"))
STORE.release(WHO, UC, "race-1", 1)
lc.check("release: an answered claim is left alone", claims("race-1") == [{"generation": 1, "status": 200}],
claims("race-1"))
assert STORE.claim(WHO, UC, "take-1", 1, d1, lease=2.0)
lc.check("lease: inside it a repeat is 409",
refused(lambda: idempotency.begin(STORE, WHO, UC, "take-1", d1)) == 409)
time.sleep(3.0)
v, errs = race(STORE, "take-1", d1, N)
rows = claims("take-1")
print(f" takeover race: {N} threads → {v.count('run')} run; claims {rows}")
lc.check("takeover: after the lease one racer takes generation 2",
v.count("run") == 1 and not errs and [r["generation"] for r in rows] == [1, 2], {"v": v, "rows": rows})
return lc.finish(TABLES)
if __name__ == "__main__":
raise SystemExit(main())