幂等键归一到一次逻辑行动

from concurrent.futures import ThreadPoolExecutor
from contextlib import closing
import json
from pathlib import Path
import sqlite3
from tempfile import TemporaryDirectory
from threading import Barrier


with TemporaryDirectory(prefix="idempotency-scope-") as directory:
    database = Path(directory) / "orders.sqlite"
    with closing(sqlite3.connect(database)) as db:
        db.executescript("""
            CREATE TABLE effects (
                id INTEGER PRIMARY KEY, tenant TEXT NOT NULL,
                operation TEXT NOT NULL, quantity INTEGER NOT NULL CHECK (quantity > 0)
            );
            CREATE TABLE receipts (
                tenant TEXT NOT NULL, operation TEXT NOT NULL, request_key TEXT NOT NULL,
                fingerprint TEXT NOT NULL, result TEXT NOT NULL, expires_at INTEGER NOT NULL,
                PRIMARY KEY (tenant, operation, request_key)
            );
        """)

    def execute(tenant, operation, key, quantity, now, fail_before_receipt=False):
        # Tenant/operation are already authenticated/routed fixture identifiers.
        assert type(quantity) is int and quantity > 0
        fingerprint = json.dumps({"quantity": quantity}, separators=(",", ":"))
        scope = (tenant, operation, key)
        with closing(sqlite3.connect(database, timeout=5)) as db:
            with db:
                db.execute("BEGIN IMMEDIATE")
                previous = db.execute(
                    "SELECT fingerprint, result, expires_at FROM receipts "
                    "WHERE tenant=? AND operation=? AND request_key=?", scope,
                ).fetchone()
                if previous is not None and previous[2] > now:
                    if previous[0] != fingerprint:
                        return "conflict", None
                    return "replay", json.loads(previous[1])
                if previous is not None:
                    db.execute("DELETE FROM receipts WHERE tenant=? AND operation=? AND request_key=?", scope)
                cursor = db.execute(
                    "INSERT INTO effects (tenant, operation, quantity) VALUES (?, ?, ?)",
                    (tenant, operation, quantity),
                )
                result = {"effect_id": cursor.lastrowid, "quantity": quantity}
                if fail_before_receipt:
                    raise RuntimeError("injected failure after effect insertion, before receipt insertion")
                db.execute("INSERT INTO receipts VALUES (?, ?, ?, ?, ?, ?)",
                           (*scope, fingerprint, json.dumps(result), now + 60))
                return "created", result

    def counts():
        with closing(sqlite3.connect(database)) as db:
            return db.execute("SELECT (SELECT count(*) FROM effects), (SELECT count(*) FROM receipts)").fetchone()

    gate = Barrier(8)

    def attempt(_):
        gate.wait(timeout=5)
        return execute("north", "web-order", "intent-a", 2, 0)

    with ThreadPoolExecutor(max_workers=8) as pool:
        outcomes = list(pool.map(attempt, range(8)))
    assert [kind for kind, _ in outcomes].count("created") == 1
    assert [kind for kind, _ in outcomes].count("replay") == 7
    original = outcomes[0][1]
    assert all(result == original for _, result in outcomes) and counts() == (1, 1)
    print("eight independent connections produce one effect, one receipt, and seven replays")

    assert execute("north", "web-order", "intent-a", 3, 1) == ("conflict", None)
    assert counts() == (1, 1)
    assert execute("south", "web-order", "intent-a", 2, 1)[0] == "created"
    assert execute("north", "batch-order", "intent-a", 2, 1)[0] == "created"
    assert execute("north", "web-order", "intent-b", 2, 1)[0] == "created"
    assert counts() == (4, 4)
    print("changed parameters conflict; a different tenant, operation, or intent key has separate scope")

    try:
        execute("north", "web-order", "rolled-back", 1, 2, fail_before_receipt=True)
    except RuntimeError as error:
        assert str(error).startswith("injected failure")
    else:
        raise AssertionError("expected the transaction to fail")
    assert counts() == (4, 4)
    assert execute("north", "web-order", "rolled-back", 1, 2)[0] == "created"
    assert counts() == (5, 5)
    print("a raised exception rolls back the effect and receipt together; retry then creates one effect")

    _, committed = execute("north", "web-order", "committed-no-reply", 1, 3)
    # Discard the reply, then query through a fresh connection; no network loss is simulated here.
    assert execute("north", "web-order", "committed-no-reply", 1, 3) == ("replay", committed)
    assert counts() == (6, 6)
    print("a fresh connection replays a committed result without adding another effect")

    assert execute("north", "web-order", "intent-a", 2, 59) == ("replay", original)
    kind, after_expiry = execute("north", "web-order", "intent-a", 2, 60)
    assert kind == "created" and after_expiry["effect_id"] != original["effect_id"]
    assert counts() == (7, 6)
    print("at the explicit 60-second retention boundary, the same key creates a new effect")
    print("SQLite", sqlite3.sqlite_version, "verified counts: 7 effects, 6 retained receipts")