Message Queues Part 1: Decoupling with Redis

1. Why Decouple Receipt from Processing

The core insight of a message queue is separation in time and in scaling. The producer records the work and moves on; the consumer picks it up whenever it is ready. In the gateway, the producer is the webhook or WebSocket receiver, and the consumers are worker processes that run the slow model call and generate a reply.

Absorbing bursts and smoothing load

Picture a highway toll plaza. If every car had to finish paying before the next could approach, a rush-hour surge would back traffic onto the highway. Instead, cars pull into a buffer of lanes and pay at their own pace. A message queue is that buffer: when inbound volume spikes above what your backend or LLM provider can handle, the queue holds the excess and workers pull only as fast as they can process.

This changes what a spike costs. Without a queue, a burst inflates response latency and the slowest platforms time out. With a queue, a burst inflates queue depth instead — the endpoint still responds in milliseconds because enqueuing is one fast write; the excess simply waits in line.

Isolating slow AI inference from fast ingress

The naive design does everything inside the HTTP handler: parse, look up context, call the LLM, generate a reply, respond. It collapses because the webhook thread does slow work synchronously on a latency-sensitive path. The decoupled pattern is: catch the event, validate minimally, drop the payload on the queue, and immediately return 200 OK. A separate worker pool consumes and does the slow processing at its own pace, so the receiver and workers scale independently.

Figure 8.1: Decoupled ingress. Platform → [Receiver: validate + XADD + return 200 OK] → Redis Stream (buffer) → [Worker Pool: XREADGROUP → LLM → reply → XACK]. The boundary between receiver and workers is the decoupling point — a slow LLM call never delays the 200 OK.

Backpressure and queue depth as a health signal

Backpressure is the mechanism by which a slow downstream component signals an upstream one to slow down instead of being overwhelmed. A queue makes backpressure both absorbable and visible: if the LLM provider is briefly slow, the queue holds the work; slow processing does not block ingestion and does not drop events — it just delays them. Because the excess accumulates in one observable place, queue depth becomes a monitorable signal of how far behind you are.

Key Takeaway: A queue decouples receipt from processing so ingress can acknowledge in milliseconds while workers do slow AI work at their own pace; bursts inflate queue depth (a health signal) instead of latency, and a slow downstream delays work rather than dropping it.

2. Redis as a Queue

Redis (Remote Dictionary Server) is an in-memory data-structure store used as a cache, database, and message broker. It offers three distinct ways to move messages between processes, and choosing correctly is the difference between a reliable gateway and one that silently drops user messages.

Lists with LPUSH/BRPOP: the simple work queue

A producer prepends items with LPUSH mylist "job" and a consumer removes them from the other end with RPOP, giving FIFO order. The blocking variant BRPOP mylist 5 parks the consumer for up to five seconds so idle consumers do not busy-loop. The fatal flaw is that BRPOP is destructive: the moment the item is popped it is gone. If the worker crashes one line later, that user's message is lost forever — lists give no acknowledgement, no redelivery, and no coordination between workers.

Pub/Sub vs queues: different delivery semantics

Redis pub/sub is a broadcast mechanism: a publisher sends to a channel and every currently-subscribed client gets a copy — excellent for real-time fan-out like a "typing…" indicator. But it is fire-and-forget with no memory: if no subscriber is connected when a message is published, it is gone. Pub/sub optimizes for many listeners each getting a copy now; a work queue needs one worker to reliably get each job eventually.

Redis Streams: consumer groups and the reliable log

Redis Streams is an append-only log of field/value entries, each stamped with an auto-generated, time-ordered ID of the form {milliseconds}-{sequence} (e.g. 1716998413541-0). Producers append with XADD; Redis assigns the ID. Unlike a list, a stream retains entries after reading; unlike pub/sub, it buffers entries for consumers not yet connected. The feature that turns a stream into a reliable work queue is the consumer group:

XGROUP CREATE mystream mygroup $ MKSTREAM

Here $ means "deliver only entries added after this point," 0 means "start from the beginning," and MKSTREAM atomically creates the stream if needed. A consumer group distributes messages so that each entry is delivered to exactly one consumer in the group. Add a second worker as a second consumer and Redis automatically splits the work — horizontal scaling with no rebalance logic to write.

Table 8.1 summarizes the three mechanisms.

Key Takeaway: Lists are simple but destructive (a crash loses the message); pub/sub broadcasts with no memory (offline subscribers miss everything); Redis Streams with a consumer group is the reliable choice — an append-only log that retains entries, delivers each to exactly one consumer, tracks acknowledgements, and scales by simply adding consumers.

3. Producers and Consumers in Python

We build the pattern with redis-py, the official Python client, in its async mode (pip install "redis>=5"). The async client returns awaitables, so every call integrates cleanly with the asyncio event loop — no blocking.

Serializing messages and envelope design

A stream entry is a flat map of field/value pairs. Rather than scatter a message across many fields, serialize the payload once and wrap it in a small envelope carrying metadata for routing and idempotency:

import json, time, uuid

def make_envelope(platform: str, user_id: str, text: str) -> dict:
    return {
        "id": str(uuid.uuid4()),              # our own idempotency key
        "platform": platform,                 # slack, discord, webex
        "user_id": user_id,
        "ts": str(time.time()),
        "body": json.dumps({"text": text}),   # opaque JSON payload
    }

Keeping the entry small is good practice; a common production pattern stores only a small pointer (a jobId) on the stream and keeps the full record in a hash or database.

The producer: XADD

The producer — inside the fast webhook handler — appends the envelope and returns. redis-py maps XADD to xadd, with * passed implicitly so Redis assigns the ID:

import redis.asyncio as redis

STREAM = "gateway:messages"

async def enqueue(r: redis.Redis, envelope: dict) -> str:
    entry_id = await r.xadd(STREAM, envelope)   # Redis assigns the ID
    return entry_id

async def on_webhook(r, platform, user_id, text):
    env = make_envelope(platform, user_id, text)
    await enqueue(r, env)          # single fast Redis write
    return {"status": "ok"}        # return 200 immediately

The consumer: XREADGROUP and XACK

A worker reads with XREADGROUP using the special ID > ("entries this group has not yet delivered to anyone"). COUNT caps the batch — a backpressure knob — and BLOCK parks the client waiting for work. When an entry is handed out, Redis records it in the group's Pending Entries List (PEL). Only after processing succeeds does the consumer call XACK to clear it. This read → process → ack loop is the linchpin of at-least-once delivery.

STREAM = "gateway:messages"
GROUP  = "workers"

async def ensure_group(r: redis.Redis) -> None:
    try:
        await r.xgroup_create(STREAM, GROUP, id="$", mkstream=True)
    except redis.ResponseError as e:
        if "BUSYGROUP" not in str(e):   # already exists is fine
            raise

async def worker(consumer_name: str) -> None:
    r = redis.Redis(decode_responses=True)
    await ensure_group(r)
    while True:
        resp = await r.xreadgroup(
            GROUP, consumer_name,
            {STREAM: ">"}, count=10, block=5000,   # BLOCK is milliseconds
        )
        for _stream, entries in resp or []:
            for entry_id, fields in entries:
                try:
                    await handle(fields)                  # the slow work
                    await r.xack(STREAM, GROUP, entry_id) # clear the PEL
                except Exception:
                    log.exception("failed entry %s", entry_id)  # leave unacked

If handle raises, we deliberately do not ack — the entry stays in the PEL and can be reclaimed later. That is the whole point of at-least-once: a partial failure leaves the work recoverable rather than lost.

A worker pool

Because each entry goes to exactly one consumer, running several workers with distinct consumer names is all it takes:

async def main():
    async with asyncio.TaskGroup() as tg:
        for i in range(4):
            tg.create_task(worker(f"worker-{i}"))

Graceful shutdown matters: a worker should trap SIGTERM/SIGINT, stop reading, finish and ack the current entry, then exit — otherwise the in-flight entry is left unacked (recoverable, but a clean shutdown avoids an unnecessary reclaim).

Key Takeaway: The producer's job is one fast XADD then return; the consumer loops on XREADGROUP ... > (with COUNT for backpressure and BLOCK to avoid busy-looping), processes the entry, and only then calls XACK. Distinct consumer names in one group give a self-balancing worker pool for free.

4. Delivery Guarantees and Failure Handling

At-least-once, the PEL, and idempotency

Redis drops an entry from the PEL only when the consumer explicitly XACKs it. If a consumer crashes after XREADGROUP but before XACK, the entry stays pending and can be reclaimed. This guarantees a message is delivered at least once — but it may be delivered more than once. So duplicates are a normal operating condition, compounded by platforms (WhatsApp and similar) that also redeliver webhooks at-least-once. Therefore consumer logic must be idempotent. The standard defense is deduplication keyed on a stable ID with a short TTL:

async def handle(fields: dict) -> None:
    msg_id = fields["id"]
    # SET key value NX (only if absent) EX 3600 (expire in 1h)
    first_time = await r.set(f"seen:{msg_id}", "1", nx=True, ex=3600)
    if not first_time:
        return                      # duplicate — already handled, skip
    reply = await call_llm(fields)  # the slow, non-idempotent work
    await send_reply(fields["platform"], fields["user_id"], reply)

Claiming stuck messages with XAUTOCLAIM

When a consumer crashes mid-processing, the entries it was handed remain in the PEL indefinitely — nothing re-delivers them automatically. XPENDING reports, per entry, the owning consumer, the idle time (ms since last delivery), and the delivery count. Idle time tells you an entry is abandoned; delivery count tells you it keeps failing. XAUTOCLAIM (Redis 6.2+) scans the PEL from a cursor, reassigns entries idle longer than a threshold to a target consumer, and returns a continuation cursor:

async def reap(r, consumer_name: str, min_idle_ms: int = 60_000):
    cursor = "0-0"
    for _ in range(10):   # max_pages safety net
        cursor, claimed, _deleted = await r.xautoclaim(
            STREAM, GROUP, consumer_name,
            min_idle_time=min_idle_ms, start_id=cursor, count=100,
        )
        for entry_id, fields in claimed:
            try:
                await handle(fields)                  # process like new work
                await r.xack(STREAM, GROUP, entry_id)
            except Exception:
                log.exception("reap failed %s", entry_id)
        if cursor == "0-0":
            break         # full sweep complete

The textbook pattern is that each healthy consumer is its own reaper: on a timer it calls XAUTOCLAIM with itself as target, then processes what it claimed as if new. The min_idle_time guard makes this safe — an entry actively being processed has a low idle time and will not be stolen. Critically, do not recover your own pending entries by reading with an explicit ID like XREADGROUP ... 0 every loop: that re-delivers all pending entries continuously and resets their idle counters to zero, keeping genuinely stuck entries permanently below the threshold.

Dead-letter handling and poison messages

Some entries fail every time — a malformed payload that crashes every consumer. This is a poison message. Its endless reclaim wastes worker time and inflates the PEL; the delivery count is how you detect it. The remedy is a dead-letter queue (DLQ): once an entry's delivery count crosses a threshold (say five), move it out of the live flow into a separate stream carrying enough context to diagnose it, then XACK the original so it stops cycling:

DLQ = "gateway:messages:dead"

async def to_dead_letter(r, entry_id, fields, reason, attempts):
    await r.xadd(DLQ, {
        "orig_id": entry_id, "reason": reason,
        "attempts": attempts, **fields,
    })
    await r.xack(STREAM, GROUP, entry_id)   # stop it cycling

New entries keep flowing past the poison pill the whole time — XREADGROUP > still delivers fresh work — so quarantining one bad message never stalls the queue. The dead-letter stream is inspectable with XRANGE gateway:messages:dead - + and replayable once the bug is fixed. On Redis 7.0+, XAUTOCLAIM returns a third element — deleted IDs (entries trimmed away by a MAXLEN limit) — removed from the PEL without an XACK; log and route them for audit.

Observability: lag and pending

XINFO GROUPS reports each group's lag (entries not yet read by anyone) and pending (delivered but unacked); XINFO CONSUMERS gives per-consumer counts and idle times. The diagnostic rule:

• Growing lag, short consumer idle times → producers outpace healthy consumers. Add workers.
• Growing pending, small lag → consumers receive but do not ack. They crash mid-message, or the reaper/ack logic is broken.

Key Takeaway: At-least-once makes duplicates normal, so handlers must be idempotent (dedup on a message ID with a short TTL). Recover crashed work with XAUTOCLAIM on an idle-time threshold — each consumer reaping for itself — quarantine poison messages to a separate dead-letter stream once their delivery count is exceeded, and monitor lag and pending to know whether to add workers or fix a broken ack loop.