幂等消费者与恰好一次语义

def make_deduper():
    seen = set()

    def process(event_id, payload):
        if event_id in seen:
            return "skipped (duplicate)"
        seen.add(event_id)
        return f"processed {payload}"

    return process


if __name__ == "__main__":
    handle = make_deduper()
    events = [
        ("evt-1", "order#100"),
        ("evt-2", "order#101"),
        ("evt-1", "order#100"),  # redelivery
    ]
    for eid, payload in events:
        print(eid, "->", handle(eid, payload))

-- Inbox table for the consumer (Postgres)
CREATE TABLE processed_messages (
    event_id   TEXT PRIMARY KEY,
    consumer   TEXT NOT NULL,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Same transaction: dedup insert + business write
BEGIN;
INSERT INTO processed_messages (event_id, consumer)
VALUES ('evt-1', 'order-service');  -- fails if already present

UPDATE accounts SET balance = balance - 50
WHERE id = 'acct-7';
COMMIT;

from sqlalchemy.exc import IntegrityError
from sqlalchemy import text

async def handle_event(session, event_id: str, order_id: str, amount: int):
    try:
        async with session.begin():
            await session.execute(
                text("INSERT INTO processed_messages (event_id, consumer) "
                     "VALUES (:eid, 'order-service')"),
                {"eid": event_id},
            )
            await session.execute(
                text("UPDATE orders SET total = total + :amt WHERE id = :oid"),
                {"amt": amount, "oid": order_id},
            )
    except IntegrityError:
        # Duplicate delivery: effect already applied, safe to ignore
        await session.rollback()
        return "duplicate"
    return "processed"

# Kafka consumer loop with manual commit AFTER processing
from confluent_kafka import Consumer

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "order-service",
    "enable.auto.commit": False,   # we commit ourselves
    "auto.offset.reset": "earliest",
})
consumer.subscribe(["orders"])

while True:
    msg = consumer.poll(1.0)
    if msg is None or msg.error():
        continue
    process_with_dedup(msg)          # durable, idempotent
    consumer.commit(message=msg)     # advance offset only now

def consume(messages, crash_after=None):
    seen = set()
    balance = 0
    committed = 0
    for i, (eid, amount) in enumerate(messages):
        if eid not in seen:        # idempotent effect
            seen.add(eid)
            balance += amount
        committed = i + 1          # commit AFTER effect
        if crash_after is not None and i == crash_after:
            break
    return balance, committed


if __name__ == "__main__":
    stream = [("e1", 50), ("e2", 30)]
    # Crash before committing e2, then e2 is redelivered
    bal, _ = consume(stream, crash_after=0)
    replay = stream[1:] + [("e2", 30)]  # redelivery duplicate
    seen = {"e1"}
    for eid, amount in replay:
        if eid not in seen:
            seen.add(eid)
            bal += amount
    print("final balance:", bal)  # 80, not 110

class FakeRedisNX:
    def __init__(self):
        self.store = {}

    def set_nx_ex(self, key, now, ttl):
        exp = self.store.get(key)
        if exp is not None and exp > now:
            return False          # still claimed -> duplicate
        self.store[key] = now + ttl
        return True               # claimed -> first time


if __name__ == "__main__":
    r = FakeRedisNX()
    print(r.set_nx_ex("evt:1", now=0, ttl=60))   # True
    print(r.set_nx_ex("evt:1", now=10, ttl=60))  # False (dup)
    print(r.set_nx_ex("evt:1", now=70, ttl=60))  # True (expired)

-- Idempotent upsert keyed by business id
INSERT INTO order_totals (order_id, total)
VALUES ('order-100', 250)
ON CONFLICT (order_id)
DO UPDATE SET total = EXCLUDED.total;

-- Idempotent state transition (absolute, not relative)
UPDATE orders SET status = 'PAID'
WHERE id = 'order-100' AND status = 'PENDING';