atelier/examples/good/messaging-idempotent-consumer.md

Good Example: Idempotent Consumer with Dedup + DLQ

An orders-consumer that achieves exactly-once-via-idempotency: at-least-once delivery plus a TTL-bounded dedup store and a DLQ routing rule. Each aspect cites the messaging principle it satisfies. Exercises the idempotency + DLQ guidance in domains/messaging/delivery-semantics.md.

The Code

# The idempotency-key dedup store (P3). TTL-bounded (P6): a dedup
# store with no TTL is a memory leak. The TTL exceeds the broker's
# max-redelivery window; beyond it, the key is expired (the broker
# has given up).

import time, json

DEDUP_TTL_SECONDS = 24 * 3600   # > broker max-redelivery window

class DedupStore:
    """P3 (idempotent), P6 (TTL-bounded). seen() before process;
    mark() after process; the order gives at-least-once + dedup."""

    def __init__(self, backend):
        # backend is Redis or a shared DB. MUST be shared across
        # consumer instances (see messaging/pubsub.md on shared vs
        # independent subscriptions).
        self.backend = backend

    def seen(self, key: str) -> bool:
        ts = self.backend.get(key)
        if ts is None:
            return False
        if time.time() - ts > DEDUP_TTL_SECONDS:
            self.backend.delete(key)    # P6: expired; not a redelivery
            return False
        return True

    def mark(self, key: str):
        self.backend.set(key, time.time(), ttl=DEDUP_TTL_SECONDS)

# The idempotent consumer. Order: dedup BEFORE process, mark AFTER
# process, ack AFTER mark. A crash before mark re-processes (the
# dedup store lacks the key); a crash before ack redelivers and the
# dedup store makes the redelivery a no-op (P3). For a
# non-idempotent process (a payment that must not double-charge),
# process+mark are one DB transaction — exactly-once via
# idempotency (P4).

def consume_orders(broker, dedup: DedupStore, process_order):
    for message in broker.receive():
        payload = json.loads(message["body"])

        # P3: dedup BEFORE process. A redelivered message is a
        # no-op, not a double-apply.
        if dedup.seen(payload["idempotencyKey"]):
            broker.ack(message)             # already processed; skip
            continue

        try:
            # P4: the declared semantic is at-least-once + idempotent
            # dedup = exactly-once-via-idempotency. For a payment,
            # process_order + dedup.mark run in one DB transaction
            # so the mark commits iff the process commits.
            process_order(payload)
            dedup.mark(payload["idempotencyKey"])
            broker.ack(message)

        except TransientError as exc:
            # P6: bounded retry with backoff. Nack for redelivery;
            # the broker redelivers after exponential backoff.
            broker.nack(message, delay=backoff(payload.get("attempt", 0)))

        except (ValueError, SchemaError) as exc:
            # P5: poison message — unparseable. Route immediately,
            # do NOT retry (no retry will fix a bad schema).
            route_to_dlq(broker, message, exc, kind="poison")
            broker.ack(message)

        except PermanentError as exc:
            # P5: permanent failure (e.g., not-found dependency).
            # Retry will not fix it — DLQ now.
            route_to_dlq(broker, message, exc, kind="dlq")
            broker.ack(message)

# The DLQ routing rule (P5 dead-letter handling, P10 DLQ depth
# alert). Distinguishes poison (unparseable; never retried) from
# DLQ (exhausted retry budget on a transient). Both carry audit
# metadata; both emit a depth metric.

DLQ = "orders-dlq"
POISON = "orders-poison"
MAX_RETRY_TTL_SECONDS = 30 * 60   # 30 min retry window

def route_to_dlq(broker, message, reason, kind: str):
    target = POISON if kind == "poison" else DLQ
    broker.send(target, body=json.dumps({
        "original": message["body"],
        "reason": str(reason),
        "kind": kind,                       # poison vs dlq
        "deadLetteredAt": now_iso(),
        "redeliveryCount": message.get("attempt", 0),
    }))
    # P10: emit a metric so DLQ depth alerts fire. A DLQ that grows
    # with no alert is a silent correctness defect (P5/P10).
    metrics.increment(f"{kind}.depth", tags={"queue": "orders"})


def consume_with_retry_budget(broker, dedup, process_order):
    # Combines TTL-with-backoff for transient failures (P6) with
    # poison-queue + DLQ + alert (P5/P10).
    for message in broker.receive():
        payload = json.loads(message["body"])
        if dedup.seen(payload["idempotencyKey"]):
            broker.ack(message); continue

        first_attempt_ts = payload.get("firstAttemptTs", time.time())
        attempt = payload.get("attempt", 0)

        try:
            process_order(payload)
            dedup.mark(payload["idempotencyKey"])
            broker.ack(message)

        except TransientError as exc:
            # P6: if the retry window is exhausted, route to DLQ;
            # otherwise redeliver with exponential backoff.
            if time.time() - first_attempt_ts > MAX_RETRY_TTL_SECONDS:
                route_to_dlq(broker, message, exc, kind="dlq")  # P5
                broker.ack(message)
            else:
                broker.nack(message, delay=backoff(attempt))

The Scenario

An orders queue delivers order.created events to the consumer at at-least-once (the declared semantic, P4). The broker redelivers on consumer crash or ack-timeout. Three things happen:

1. Normal delivery — the consumer dedups by idempotencyKey, processes, marks, acks.
2. Redelivery after a crash before ack — the consumer crashed after mark but before ack. The broker redelivers; seen() returns true; the consumer acks without re-processing (P3).
3. Poison message — a malformed JSON body. The consumer routes it to the poison queue immediately (no retry will fix a parse error), acks the origin, and emits a poison.depth metric. The operator is paged on poison-queue growth (P10).

A transient downstream failure (the payments API is briefly 503) retries with exponential backoff for 30 minutes (P6); if it exceeds the budget, the message routes to the DLQ with reason, redeliveryCount, and deadLetteredAt — auditable, drainable, observable (P5). The DLQ depth metric alerts the operator; the DLQ entry's audit metadata lets the operator replay after the bug is fixed (P5 reversibility).

Principles Demonstrated

Consumers are Idempotent (Messaging P3, C1)

• The consumer dedups by idempotency key before processing. A redelivered message is a no-op, not a double-apply. The process → mark → ack order gives at-least-once + idempotent dedup; for a non-idempotent process, process + mark are one DB transaction (exactly-once via idempotency, P4).
• See domains/messaging/delivery-semantics.md (idempotency-key dedup store) and domains/messaging/first-principles.md P3.

Delivery Semantics are Explicit (Messaging P4, C1, C2)

• The channel is declared at-least-once + idempotent consumer — the engineering practice that collapses to exactly-once under correct dedup (P3). The semantic is not emergent; it is the declared choice per channel. The tradeoff (dedup-store cost, transactional-process complexity) is conscious and documented.
• See domains/messaging/queues.md (the three-semantics comparison table) and domains/messaging/delivery-semantics.md (exactly- once via idempotency).

Dead-Letter Handling is Defined (Messaging P5, C1, C5)

• Poison messages (unparseable) route immediately to the poison queue — no retry will fix them. Transient failures retry with backoff until the TTL, then route to the DLQ. Both carry audit metadata (reason, redeliveryCount, deadLetteredAt); both are drainable and observable. The DLQ is the reversibility mechanism — a dead-lettered message can be reprocessed after the bug is fixed.
• See domains/messaging/delivery-semantics.md (dead-letter strategy comparison table, DLQ routing rule) and domains/messaging/first-principles.md P5.

Messaging is Observable (Messaging P10, C7, C1)

• DLQ depth and poison-queue depth are emitted as metrics and wired to alerts. A DLQ that grows silently is a correctness defect; the alert makes it visible. Silent backlog is a bug, not a feature — the operator is paged on growth, not on a customer report.
• See domains/messaging/delivery-semantics.md (DLQ depth as an alert) and domains/messaging/first-principles.md P10.

Cross-Domain Links

• domains/messaging/delivery-semantics.md — the idempotency-key dedup-store pattern and the dead-letter strategy comparison table exercised here (TTL-with-backoff + poison-queue + DLQ + alert).
• domains/messaging/queues.md — the three-semantics comparison table; the ack/nack/redelivery model this consumer uses.
• domains/messaging/first-principles.md — P3, P4, P5, P10 are the principles demonstrated.
• domains/concurrency/patterns — the in-process retry/backoff analog (Pattern 6, Timeout on Every Block); messaging owns the broker-backed instance where redelivery comes across a network.
• domains/errors/patterns — errors as data: a DLQ entry is the async-messaging instance of an error log (observable, auditable, drainable).
• domains/observability/metrics — the generic SLI/SLO discipline the DLQ-depth alert builds on.
• review/anti-patterns.md — the messaging-unbounded-retry and messaging-shared-subscription chaos anti-patterns are the inverse of this example's bounded retry + independent-consumer discipline.