---
name: "message_queues"
description: "Design async task queues: when to queue vs do work inline, producers/consumers, dead-letter queues, retry with exponential backoff + jitter, idempotent consumers, and at-least-once vs at-most-once delivery semantics. Triggers on 'should I queue this', 'add a job queue', 'dead letter queue', 'retry backoff', 'idempotent consumer', 'task queue design'."
metadata: { "includeInPrompt": true }
---

> Differs from `webhook-handling` because: this is the ASYNC task-queue side — producers/consumers, DLQs, retry backoff, delivery semantics. `webhook-handling` is the synchronous receiving side: verifying and retrying inbound HTTP callbacks. Complements it; does not overlap.

## Purpose
Turn slow, bursty, or failure-prone work into reliable background jobs. Use this when a task asks for a job queue, delayed/scheduled work, cross-service messaging, or when a synchronous request handler is doing work it shouldn't.

## Workflow

### 1. Decide: queue it or do it inline
**Queue when** any of these are true:
- Work is slow (API calls, video rendering, email sending, report generation) and the caller can't wait.
- Load is bursty — spikes would overwhelm the request path (rate-limited APIs, onboarding surges).
- Work needs retry — a transient failure (network blip, 429) should not become a failed request.
- Work crosses a service boundary — producers and consumers should deploy/scale independently.
- Work is scheduled or deferred ("send this in 2 hours", "run nightly").

**Do NOT queue when**: the caller needs the result in this request; the work is trivially fast (<~100ms, no I/O); strict global ordering matters (queues guarantee ordering only per partition/queue, not across workers — check your broker's docs); or adding infrastructure would exceed the reliability the task actually needs.

### 2. Model the pieces
- **Producer**: the code that creates the job. Message = small payload + unique ID + attempt count. Keep the payload lean; store large data elsewhere and reference it.
- **Queue/topic**: the durable buffer. One queue per job type (don't mix 1s and 1h jobs — slow jobs head-of-line-block fast ones).
- **Consumer**: pulls a message, does the work, then **acks only after work completes**. No early acks.
- **Visibility timeout / redelivery**: while a consumer works, the message is hidden. If the consumer crashes without acking, the broker makes it visible again for someone else — this is how at-least-once delivery works. Timeout must exceed the 99th-percentile job duration, or long jobs get double-processed.

### 3. Know your delivery semantics
- **At-most-once**: ack before doing the work (or fire-and-forget). Message lost if the consumer crashes. Use only for work that's safe to drop (metrics, heartbeats).
- **At-least-once**: ack after work completes; crashes redeliver. Duplicates WILL happen — so make the consumer idempotent. This is the default for background jobs.
- **Effectively-once**: at-least-once + an idempotent consumer (see step 5). Duplicates arrive but have no effect.
- **Exactly-once**: cannot be delivered by the broker alone. Only achievable end-to-end via idempotency — beware anyone claiming a broker setting "guarantees" it.

### 4. Design retry: backoff, caps, then dead-letter
Retry formula: `delay = base * 2^attempt + jitter`, where jitter is a random fraction of the base. The `2^attempt` spreads retries exponentially so a flapping dependency isn't hammered; jitter (add `random.uniform(0, base)`) stops all workers from retrying in lockstep (thundering herd). Cap the delay (e.g. max 5–15 minutes) and cap attempts (3–5 is typical).
- Retry ONLY transient failures: timeouts, 429s, 5xx from dependencies. Never retry poison messages (malformed payloads, permanent 4xx) — send those straight to the dead-letter queue.
- **Dead-letter queue (DLQ)**: a second queue where messages land after max attempts. DLQ triage procedure: (1) inspect the message + error (add the failing payload and exception to the DLQ record); (2) decide — fixable bug → fix code and replay the message; bad data → discard or route to manual review; (3) alert on DLQ growth (a growing DLQ is an incident, not storage). Set a retention policy so the DLQ doesn't fill silently.

### 5. Make the consumer idempotent
Because at-least-once redelivers, the same message can arrive twice. Three patterns, in preference order:
1. **Natural idempotency**: design the operation so repeats are no-ops ("set status=done", "upsert by ID").
2. **Dedupe store**: keep a set/table of processed message IDs; skip IDs already seen. Store the ID durably (DB, Redis) if duplicates can arrive across restarts.
3. **Conditional writes**: guard the write itself ("UPDATE ... WHERE status != 'done'"), so the second attempt changes nothing.

### 6. Run the demo (stdlib only)
Save this as `queue_demo.py`, then run `python3 queue_demo.py`. It demonstrates producer/consumer, transient retry with backoff + jitter, DLQ routing for a poison message after 3 attempts, and a duplicate delivery skipped by the idempotency check.

```python
"""queue_demo.py — async task queue demo (Python stdlib only).
Producer/consumer, retry with exponential backoff + jitter, dead-letter
queue routing after N failures, and an idempotent consumer (dedupe set).
Run: python3 queue_demo.py"""
import queue, random, threading, time

MAX_ATTEMPTS, BASE_DELAY = 3, 0.1
work_q, processed_ids, DLQ = queue.Queue(), set(), []
transient_seen, lock = set(), threading.Lock()
random.seed(42)  # deterministic jitter for the demo


def backoff(attempt):  # delay = base * 2^attempt + jitter
    return BASE_DELAY * 2 ** attempt + random.uniform(0, 0.05)


def do_work(tid):
    if tid == 3:  # poison message: always fails
        raise RuntimeError("poison message")
    with lock:  # task 5: one transient failure, then succeeds
        if tid == 5 and tid not in transient_seen:
            transient_seen.add(tid)
            raise TimeoutError("transient failure")


def producer(tasks):
    for t in tasks:
        work_q.put({"id": t, "attempts": 0})
        print(f"[producer] enqueued task {t}")


def consumer():
    while True:
        task = work_q.get()
        if task is None:  # shutdown sentinel — still ack it, or join() hangs
            work_q.task_done()
            break
        tid, attempts = task["id"], task["attempts"]
        with lock:
            if tid in processed_ids:
                print(f"[consumer] task {tid}: duplicate delivery — skipped (idempotent)")
                work_q.task_done()
                continue
        try:
            do_work(tid)
        except Exception as e:
            attempts += 1
            if attempts >= MAX_ATTEMPTS:
                DLQ.append({"id": tid, "error": str(e)})
                print(f"[consumer] task {tid}: failed {attempts}x ({e}) -> DEAD-LETTER QUEUE")
            else:
                d = backoff(attempts)
                print(f"[consumer] task {tid}: attempt {attempts} failed ({e}); retry in {d:.2f}s")
                time.sleep(d)
                work_q.put({"id": tid, "attempts": attempts})
        else:
            with lock:
                processed_ids.add(tid)
            print(f"[consumer] task {tid}: done (ack)")
        work_q.task_done()  # ack ONLY after work completes or is parked


t = threading.Thread(target=consumer)
t.start()
producer([1, 2, 3, 4, 5, 6])
producer([2])  # simulate at-least-once redelivery of an already-done task
work_q.join()  # drain all work first — retries requeue during this
work_q.put(None)  # sentinel only after the queue is truly empty
t.join()
print(f"\nprocessed IDs: {sorted(processed_ids)}")
print(f"DLQ: {[m['id'] for m in DLQ]}")
```

Verified output (Python 3, stdlib only, run locally):

```
[producer] enqueued task 1
[producer] enqueued task 2
[producer] enqueued task 3
[producer] enqueued task 4
[producer] enqueued task 5
[producer] enqueued task 6
[producer] enqueued task 2
[consumer] task 1: done (ack)
[consumer] task 2: done (ack)
[consumer] task 3: attempt 1 failed (poison message); retry in 0.23s
[consumer] task 4: done (ack)
[consumer] task 5: attempt 1 failed (transient failure); retry in 0.20s
[consumer] task 6: done (ack)
[consumer] task 2: duplicate delivery — skipped (idempotent)
[consumer] task 3: attempt 2 failed (poison message); retry in 0.41s
[consumer] task 5: done (ack)
[consumer] task 3: failed 3x (poison message) -> DEAD-LETTER QUEUE

processed IDs: [1, 2, 4, 5, 6]
DLQ: [3]
```

### 7. Pick a real broker when outgrowing this
The demo above is the mental model, not production. When you need durability across restarts, multiple workers, or cross-service messaging, reach for a real broker — RabbitMQ, Apache Kafka, Redis Streams, or AWS SQS/SNS. All of them implement some version of: queues/topics, consumers, ack/redelivery (visibility timeout), retries, and DLQs. Broker APIs and CLIs vary by version and deployment, so check your broker's docs for the exact commands rather than trusting recalled syntax.

## Output Contract
When this skill fires, deliver:
1. The queue/inline verdict with the deciding checklist line (which bullet applied).
2. The design: job types → queue mapping, retry policy (base, cap, max attempts, what's retried), DLQ + triage plan, and the idempotency mechanism.
3. The ack discipline stated explicitly (ack only after work completes) and the visibility-timeout vs job-duration check.

## Operating Rules
1. Never ack before work completes — an early ack is at-most-once by accident.
2. Retry only transient failures; route poison messages straight to the DLQ.
3. Always backoff + jitter; never retry in a tight loop (it turns one outage into a self-inflicted DDoS).
4. Assume duplicates: every consumer is idempotent by design, not by luck.
5. Cap both delay and attempts; every retry policy ends in a DLQ, not in infinite retry.
6. Monitor DLQ depth and consumer lag — queue metrics are the health metrics.
7. Do not claim exactly-once delivery from broker configuration; it's idempotency end-to-end or it isn't real.
8. Name only real brokers; give exact CLI/API commands only when verified — otherwise say "check your broker's docs".