FastAPI Backend Development Bootcamp · レッスン

冪等なコンシューマーとExactly-Onceセマンティクス

メッセージ処理を重複排除してチェックポイントを記録し、下流で実質的なExactly-Once配信を実現します。

レッスン 4/413 ステップ

「冪等なコンシューマーとExactly-Onceセマンティクス」はCoddyKit上の無料FastAPI Backend Development Bootcampレッスンです。これはレッスン4/4です。下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。これはFastAPI Backend Development Bootcamp学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 FastAPI Backend Development Bootcampコースには全4レッスンが含まれています。

このレッスンの一部はまだ翻訳されておらず、英語で表示されています。

The Delivery Problem

Message brokers like Kafka and Pulsar default to at-least-once delivery. If a consumer crashes after processing a message but before acknowledging it, the broker redelivers that message on restart.

For a FastAPI backend draining an order-events topic, that means the same OrderPlaced event can hit your handler twice. Without protection, you charge the customer twice or send two confirmation emails.

At-most-once: ack before processing — fast, but you lose messages on crash.
At-least-once: ack after processing — no loss, but duplicates.
Exactly-once (effective): at-least-once delivery plus idempotent handling.

Idempotency Is the Real Goal

True exactly-once delivery across a network is impossible in the general case. What you can build is effective exactly-once: the broker may deliver a message many times, but your consumer applies its effect exactly once.

A handler is idempotent when running it twice with the same input yields the same final state as running it once. The strategy: give every message a stable unique key, record keys you have already processed, and skip anything you have seen before.

Choosing a Deduplication Key

The dedup key must be stable across redeliveries. The broker offset is NOT stable — a rebalanced partition can change offsets, and Pulsar message IDs differ from Kafka offsets.

Prefer a business-level identifier carried in the message itself, or a producer-assigned event id:

event_id set by the producer (a UUID) — best general choice.
A natural key like order_id when each order produces one event of a kind.
(topic, partition, offset) only as a last resort.

The pure dedup function below just remembers keys it has seen.

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))

In-Memory Dedup Is Not Enough

The set-based deduper resets every time the FastAPI process restarts — and crashes are exactly when redeliveries happen. You need a durable dedup store that survives restarts and is shared across worker replicas.

Two common backends:

Database table of processed ids — strong consistency, joins with your business data, supports transactions.
Redis with TTL — fast, great when redeliveries only happen within a bounded window.

The right choice depends on how long a duplicate can realistically arrive after the original.

The Inbox Table Pattern

The inbox pattern stores each processed message id in a table with a UNIQUE constraint. Before applying effects, you try to insert the id; a unique-violation means it is a duplicate, so you skip.

Crucially, you insert the dedup row and apply the business change in the same database transaction. Either both commit or neither does — there is no window where the effect happened but the id was not recorded.

-- 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;

Idempotent Consumer in FastAPI

Here is the dedup-in-transaction idea wired into an async consumer using SQLAlchemy. The INSERT of the event id and the business mutation share one transaction. If the insert raises IntegrityError, we already processed this event and roll back.

Because this depends on a running database and SQLAlchemy session, it is framework/infrastructure code — not something an online judge can execute standalone.

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"

Checkpointing: Commit Offsets After Effects

Idempotency handles duplicates; checkpointing controls when the broker advances past a message. The golden rule: commit the offset only after the effect is durably persisted.

Disable auto-commit and commit manually after your transaction succeeds. This guarantees at-least-once: a crash before the commit replays the message, and idempotency absorbs the replay.

Kafka: enable.auto.commit=false, then consumer.commit().
Pulsar: turn off automatic ack, call consumer.acknowledge(msg) after success.

# 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

Why Ordering of Ack and Effect Matters

The sequence of operations is the whole ballgame. Compare two orderings after the effect is applied but before a crash:

Commit-then-process: offset advances first. A crash loses the message — at-most-once. Bad for money.
Process-then-commit: effect persists first. A crash replays the message — at-least-once. Dedup makes the replay harmless.

Always choose process-then-commit and lean on idempotency. The simulation below shows a redelivery being absorbed.

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

Redis Dedup With a TTL Window

When duplicates can only arrive within a bounded window (say, broker retry plus rebalance time), Redis is a lightweight dedup store. Use SET key value NX EX ttl: it sets the key only if absent and auto-expires it.

The atomic NX flag makes the check-and-claim a single operation, so two concurrent workers cannot both win the same event id. Below is a pure simulation of that logic with an expiry clock.

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)

Naturally Idempotent Operations

Sometimes you can sidestep an explicit dedup store by making the operation itself idempotent, so applying it twice equals applying it once.

Upserts: INSERT ... ON CONFLICT DO UPDATE keyed by a business id.
Set assignments: status = 'PAID' instead of a relative toggle.
Conditional writes: apply only if a version or state guard matches.

Relative mutations like balance = balance - 50 are NOT idempotent — running them twice double-counts. Convert such effects to absolute state or guard them with a processed-id check.

-- 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';

End-to-End Exactly-Once Pipeline

Putting it together for a FastAPI consumer:

Producer stamps every event with a stable event_id (UUID).
Consumer reads with auto-commit disabled.
In one DB transaction: insert the event_id into the inbox (dedup) and apply the business effect.
On IntegrityError, skip — the event was already handled.
Only after the transaction commits, commit/ack the offset to the broker.

This yields effective exactly-once: at-least-once delivery from the broker, exactly-once effect from idempotency, and no lost messages from checkpoint ordering.

Quick Check

You run an at-least-once Kafka consumer that decrements account balances. To achieve effective exactly-once, what is the correct combination?

対話的なレッスンで答える

Recap

You learned how to turn at-least-once delivery into effective exactly-once processing:

Idempotency is the goal — make double-processing produce the same state as single processing.
Pick a stable dedup key, preferably a producer-assigned event_id, never a raw broker offset.
Use a durable inbox (DB unique constraint or Redis NX+TTL) so dedup survives restarts and replicas.
One transaction for the dedup insert and the business effect — all or nothing.
Checkpoint last: commit/ack the offset only after the effect is persisted (process-then-commit).
Prefer naturally idempotent operations (upserts, absolute state) over relative mutations.

無料で開始

AI チューターと学ぶ FastAPI Backend Development Bootcamp — 無料

ブラウザでリアルコードを書いて実行し、24/7 の AI チューターから瞬時にサポートを受け、ウェブまたはアプリで続きから学習できます。

コース: 21
レッスン: 84

無料で学習を始めるコース全体を表示

よくある質問

「冪等なコンシューマーとExactly-Onceセマンティクス」レッスンは無料ですか？

はい。「冪等なコンシューマーとExactly-Onceセマンティクス」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、FastAPI Backend Development Bootcampコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 FastAPI Backend Development Bootcampコースには全4レッスンが含まれています。

「冪等なコンシューマーとExactly-Onceセマンティクス」で何を学びますか？

メッセージ処理を重複排除してチェックポイントを記録し、下流で実質的なExactly-Once配信を実現します。ブラウザで直接実行するハンズオンコードでFastAPI Backend Development Bootcampを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

FastAPI Backend Development Bootcampを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのFastAPI Backend Development Bootcampは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。これはレッスン4/4です。

「冪等なコンシューマーとExactly-Onceセマンティクス」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このFastAPI Backend Development Bootcampレッスンでコードを書いて実行できますか？

はい。すべてのFastAPI Backend Development Bootcampレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

← FastAPI Backend Development Bootcampに戻る