Dados, consistência e publicação de eventos

"""Original in-memory SQLite teaching lab, not a Spanner or Pub/Sub emulator.

All external calls below are Python dictionary operations. No network, credentials,
files, concurrency, vendor services or actual payments are used.
"""
import json
import platform
import sqlite3

checks = []


def check(name, condition):
    if not condition:
        raise AssertionError(name)
    checks.append(name)


def fails(call, exception):
    try:
        call()
    except exception:
        return True
    return False


db = sqlite3.connect(":memory:", isolation_level=None,
                     autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL)
db.executescript("""
CREATE TABLE orders (id TEXT PRIMARY KEY, amount INTEGER NOT NULL);
CREATE TABLE outbox (id TEXT PRIMARY KEY, amount INTEGER NOT NULL, sent INTEGER NOT NULL);
""")
unsafe_receipts = []


def unsafe_attempt(order_id, amount, abort=False):
    db.execute("BEGIN")
    try:
        # This fictional external list is outside SQLite's transaction boundary.
        unsafe_receipts.append((order_id, amount))
        db.execute("INSERT INTO orders VALUES (?, ?)", (order_id, amount))
        if abort:
            raise RuntimeError("injected failure before commit")
        db.execute("COMMIT")
    except Exception:
        db.execute("ROLLBACK")
        raise


def enqueue(order_id, amount, abort=False):
    db.execute("BEGIN")
    try:
        existing = db.execute("SELECT amount FROM orders WHERE id=?", (order_id,)).fetchone()
        if existing:
            if existing[0] != amount:
                raise ValueError("same identity with different intent")
            db.execute("COMMIT")
            return "existing"
        db.execute("INSERT INTO orders VALUES (?, ?)", (order_id, amount))
        if abort:
            raise RuntimeError("injected failure before outbox write")
        db.execute("INSERT INTO outbox VALUES (?, ?, 0)", (order_id, amount))
        db.execute("COMMIT")
        return "created"
    except Exception:
        db.execute("ROLLBACK")
        raise


external_effects = {}
delivery_calls = []


def fictional_receiver(order_id, amount):
    delivery_calls.append((order_id, amount))
    if order_id in external_effects:
        if external_effects[order_id] != amount:
            raise ValueError("receiver identity collision")
        return "existing"
    external_effects[order_id] = amount
    return "applied"


def relay(order_id, lose_response=False):
    row = db.execute("SELECT amount, sent FROM outbox WHERE id=?", (order_id,)).fetchone()
    if row is None:
        raise LookupError("no publication intent")
    if row[1]:
        return "already-sent"
    receipt = fictional_receiver(order_id, row[0])
    if lose_response:
        raise TimeoutError("injected response loss after receiver applied effect")
    db.execute("UPDATE outbox SET sent=1 WHERE id=?", (order_id,))
    return receipt


try:
    check("unsafe attempt aborts", fails(lambda: unsafe_attempt("U1", 80, True), RuntimeError))
    check("aborted database row absent", db.execute("SELECT * FROM orders WHERE id='U1'").fetchone() is None)
    check("external effect survives rollback", unsafe_receipts == [("U1", 80)])
    unsafe_attempt("U1", 80)
    check("retry creates second external effect", len(unsafe_receipts) == 2)
    check("one local row does not prove one external effect", db.execute("SELECT count(*) FROM orders WHERE id='U1'").fetchone()[0] == 1)
    check("safe enqueue abort injected", fails(lambda: enqueue("S1", 120, True), RuntimeError))
    check("order rolls back with failed intent", db.execute("SELECT * FROM orders WHERE id='S1'").fetchone() is None)
    check("no partial outbox row", db.execute("SELECT * FROM outbox WHERE id='S1'").fetchone() is None)
    check("successful atomic enqueue", enqueue("S1", 120) == "created")
    check("order and intent both present", db.execute("SELECT o.amount, x.amount FROM orders o JOIN outbox x USING(id) WHERE id='S1'").fetchone() == (120, 120))
    check("repeat original identity returns existing", enqueue("S1", 120) == "existing")
    check("different amount under same identity rejected", fails(lambda: enqueue("S1", 121), ValueError))
    check("collision does not change intent", db.execute("SELECT amount FROM outbox WHERE id='S1'").fetchone()[0] == 120)
    check("response loss follows external effect", fails(lambda: relay("S1", True), TimeoutError))
    check("effect exists despite timeout", external_effects == {"S1": 120})
    check("intent remains retryable", db.execute("SELECT sent FROM outbox WHERE id='S1'").fetchone()[0] == 0)
    check("receiver recognizes retry", relay("S1") == "existing")
    check("two calls one effect", len(delivery_calls) == 2 and len(external_effects) == 1)
    check("successful response marks sent", db.execute("SELECT sent FROM outbox WHERE id='S1'").fetchone()[0] == 1)
    check("sent intent is not sent again", relay("S1") == "already-sent" and len(delivery_calls) == 2)
    check("new ID creates distinct intent", enqueue("S2", 120) == "created")
    check("new ID is a new external effect", relay("S2") == "applied" and len(external_effects) == 2)
    check("receiver rejects mismatched retry payload", fails(lambda: fictional_receiver("S1", 999), ValueError))
    check("receiver preserved original effect", external_effects["S1"] == 120)
    check("relay rejects missing intent", fails(lambda: relay("missing"), LookupError))
    print(json.dumps({"passed": len(checks), "checks": checks,
                      "python": platform.python_version(), "sqlite": sqlite3.sqlite_version,
                      "database": ":memory:", "network": False,
                      "vendorExecution": False,
                      "scope": "Sequential SQLite atomic writes and synthetic receiver identity checks; no concurrency, actual transport, Spanner isolation or durable external idempotency tested."}))
finally:
    db.close()

Tópicos relacionados: Migração e reconciliação de dados · Idempotência e recuperação de incidentes · Critérios de aceitação e handover para RUN

Referência: Spanner transactions overview · Current linked standard guide; edition date unconfirmed (2026-09-30 inspection)