"""Synthetic SQLite model of retry boundaries, not an Odoo or n8n runtime test. Run: python3 retry-model.py Uses only the Python standard library, temporary files, and no network. This model deliberately compares harmful schedules with an atomic target-side operation ledger. It does not install a production integration or prove Odoo's transaction/isolation, ACL, node retry, accounting, or deployment behavior. """ from concurrent.futures import ThreadPoolExecutor import hashlib import json from pathlib import Path import sqlite3 import tempfile import threading class ResponseLost(Exception): pass class PayloadConflict(Exception): pass class PermissionDenied(Exception): pass def database(path): conn = sqlite3.connect(path, timeout=10, isolation_level=None) conn.execute('PRAGMA foreign_keys = ON') conn.executescript(''' CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY, company TEXT NOT NULL, source_order TEXT NOT NULL, amount_cents INTEGER NOT NULL, currency TEXT NOT NULL, state TEXT NOT NULL DEFAULT 'draft' ); CREATE TABLE IF NOT EXISTS operations ( op_key TEXT PRIMARY KEY, payload_hash TEXT NOT NULL, order_id INTEGER NOT NULL REFERENCES orders(id) ); ''') return conn def payload(company='A', amount=12500): return {'company': company, 'source': 'shop', 'source_order': 'SO-77', 'command': 'create_draft_order', 'revision': 1, 'amount_cents': amount, 'currency': 'EUR'} def operation_key(p): return json.dumps([p[k] for k in ('source', 'company', 'source_order', 'command', 'revision')], separators=(',', ':')) def fingerprint(p): return hashlib.sha256(json.dumps(p, sort_keys=True, separators=(',', ':')).encode()).hexdigest() def create_order(conn, p): return conn.execute('INSERT INTO orders(company, source_order, amount_cents, currency) VALUES (?, ?, ?, ?)', (p['company'], p['source_order'], p['amount_cents'], p['currency'])).lastrowid def count(conn): return conn.execute('SELECT COUNT(*) FROM orders').fetchone()[0] def atomic_target(conn, p, *, authorized=('A', 'B'), fail_at=None): # Policy check still runs for a replay: prior success is not current authority. if p['company'] not in authorized: raise PermissionDenied(p['company']) key, digest = operation_key(p), fingerprint(p) conn.execute('BEGIN IMMEDIATE') try: existing = conn.execute('SELECT payload_hash, order_id FROM operations WHERE op_key=?', (key,)).fetchone() if existing: if existing[0] != digest: raise PayloadConflict(key) order_id = existing[1] else: order_id = create_order(conn, p) if fail_at == 'before_commit': raise ResponseLost('simulated failure before atomic commit') conn.execute('INSERT INTO operations(op_key, payload_hash, order_id) VALUES (?, ?, ?)', (key, digest, order_id)) conn.execute('COMMIT') except Exception: if conn.in_transaction: conn.execute('ROLLBACK') raise if fail_at == 'after_commit': raise ResponseLost('target committed; response deliberately discarded') return order_id def main(): results=[] def record(case, expected, observed, meaning): assert observed == expected, (case, expected, observed) results.append({'case':case,'expected':expected,'observed':observed,'meaning':meaning}) with tempfile.TemporaryDirectory(prefix='odoo-retry-model-') as tmp: def fresh(name):return database(str(Path(tmp)/(name+'.sqlite'))) p=payload() c=fresh('naive') create_order(c,p) # First commit occurs, but the caller does not retain its result. create_order(c,p) # Blind retry of the create command. record('blind_retry_after_commit',2,count(c),'Two draft orders for one intended operation; duplicate reproduced in the model.') c.close() c=fresh('premark') claimed={operation_key(p)} # Local receipt written before the external effect. # Crash before create_order. Restart sees the receipt and skips the event. if operation_key(p) not in claimed:create_order(c,p) record('mark_processed_before_effect',0,count(c),'A duplicate gate can lose work if receipt is treated as completion.') c.close() c=fresh('race') a=c.execute('SELECT id FROM orders WHERE company=? AND source_order=?',('A','SO-77')).fetchone() b=c.execute('SELECT id FROM orders WHERE company=? AND source_order=?',('A','SO-77')).fetchone() # A deterministic interleaving: both readers saw no match before either wrote. if a is None:create_order(c,p) if b is None:create_order(c,p) record('search_then_create_interleaving',2,count(c),'A prior search alone does not serialize competing writers.') c.close() c=fresh('atomic') try:atomic_target(c,p,fail_at='after_commit') except ResponseLost:pass recovered=atomic_target(c,p) second=atomic_target(c,p) record('target_atomic_ledger_lost_response',{'orders':1,'same_id':True,'state':'draft'}, {'orders':count(c),'same_id':recovered==second,'state':c.execute('SELECT state FROM orders').fetchone()[0]}, 'Replay returns the original draft record; it does not confirm, invoice, post or send it.') before=c.execute('SELECT amount_cents FROM orders').fetchone()[0] conflict=False try:atomic_target(c,payload(amount=13000)) except PayloadConflict:conflict=True record('same_key_changed_payload',{'conflict':True,'orders':1,'amount_unchanged':True}, {'conflict':conflict,'orders':count(c),'amount_unchanged':c.execute('SELECT amount_cents FROM orders').fetchone()[0]==before}, 'Changed business content requires explicit review or a new approved revision, not an unnoticed replay.') atomic_target(c,payload(company='B')) record('same_source_id_different_company',2,count(c),'Company is part of the operation namespace; the fixture explicitly authorizes A and B.') blocked=False try:atomic_target(c,p,authorized=('B',)) except PermissionDenied:blocked=True record('replay_after_authority_revoked',True,blocked,'An existing result does not bypass the current authorization check.') c.close() c=fresh('rollback') try:atomic_target(c,p,fail_at='before_commit') except ResponseLost:pass before_retry=count(c) atomic_target(c,p) record('failure_before_commit',{'before_retry':0,'after_retry':1}, {'before_retry':before_retry,'after_retry':count(c)},'The local atomic transaction rolls back the uncommitted order and receipt together.') c.close() path=str(Path(tmp)/'concurrent.sqlite') c=database(path);c.close() barrier=threading.Barrier(2) def worker(_): local=database(path) try: barrier.wait(timeout=5) return atomic_target(local,p) finally:local.close() with ThreadPoolExecutor(max_workers=2) as pool:ids=list(pool.map(worker,range(2))) c=database(path) record('two_concurrent_target_requests',{'orders':1,'same_id':True}, {'orders':count(c),'same_id':ids[0]==ids[1]},'SQLite serializes this fixture transaction; production Odoo/PostgreSQL concurrency still needs its own test.') c.close() # A client-only receipt cannot establish absence at an external destination. receipt={'status':'processing','target_id':None} visible_matches=[] decision='unknown' if receipt['status']=='processing' and not visible_matches else 'review' record('unresolved_client_only_receipt','unknown',decision,'Do not infer that no target write happened from a missing local result or an inconclusive lookup.') output={'kind':'executed synthetic Python/SQLite model','run_on':'2026-11-05', 'runtime':'Python standard library; no Odoo, n8n, network, PostgreSQL or AI model invoked', 'cases':results,'case_count':len(results),'all_expected_observations_matched':True, 'production_limits':['SQLite locking is not PostgreSQL/Odoo concurrency behavior.', 'The target-side ledger and order commit share one model database; a separate n8n data table does not create that atomicity.', 'No permissions, accounting posting, stock reservations, node retry semantics or external side effects were tested in Odoo.']} print(json.dumps(output,indent=2)) if __name__=='__main__':main()