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

```python
# 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)
```

```python
# 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)
```

```python
# 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.