The Queue and the Worker

A queue is where the api puts work and the worker takes it, and the guarantees it offers are what the worker must be written for. Stagedoor's queue is one Redis stream named jobs with one consumer group named workers. XADD appends a job, XREADGROUP lets a worker claim one, XACK says it is done, and a job that was claimed and never acknowledged is what a crashed worker leaves behind. Everything the handler has to tolerate follows from those four facts.

This topic is the mechanics of that loop as a client of the stream: what an entry is, what the group tracks, where the acknowledgement goes, how a dead worker's jobs are taken over, and what an untrimmed stream costs. The broker's internals and the choice between brokers are Middleware Deep Dive's subject, and the book holds that line on purpose: the same loop, with the same failure modes and different command names, is what a RabbitMQ channel, an SQS receive and a Kafka consumer group hand the reader.

The Stream

A stream is an append-only log of entries, each with an id and a flat list of field-value pairs. The id is assigned by Redis when the entry is added with *: a millisecond timestamp and a sequence number, 1758200000000-0, which means the id of every job says when it was queued, a fact Topic 47 turns into the queue-age metric for free. The relay of Chapter 7 is the one thing in Stagedoor that adds to jobs, and it adds one entry per outbox row.

# the relay, once per outbox row; Redis assigns the id and returns it
XADD jobs MAXLEN ~ 100000 * kind render_tickets order_id 4471 outbox_id 88210 request_id 7f3a9c2e trace 4bf92f35
→ "1758200000000-0"

# once, at first deploy: the group starts at the end of the stream ($)
XGROUP CREATE jobs workers $ MKSTREAM

# each worker, in its loop: up to 10 new entries (>), waiting at most 5 s for one
XREADGROUP GROUP workers worker-01 COUNT 10 BLOCK 5000 STREAMS jobs >

# after the handler returns, never before
XACK jobs workers 1758200000000-0

The first command adds one job. Its fields carry the kind, which selects the handler; the order id, which is everything the handler needs to load the rest; the outbox id, which is the job's identity for Topic 45; and the request id and trace from Topic 21 of Chapter 4, so the job's log lines join the checkout that queued it. The entry carries nothing the handler could look up, which is why it is 5 fields and not the order. The second command creates the group once, starting at the end of the stream so the first worker does not replay history. The third is the read a worker sits in, asking for up to 10 entries it has not seen and waiting up to 5 seconds for one. The fourth is the acknowledgement, and the comment on it is where the next section starts.

The Consumer Group

Without a group, every reader of a stream sees every entry, which is a broadcast and not a queue. The group is what turns it into one. It records a cursor, the last id delivered to any of its consumers, and a pending entries list of every entry delivered and not yet acknowledged, with the consumer's name, the time of delivery and a count of how many times it has been delivered. When worker-01 reads with the special id >, it receives entries past the cursor, the cursor moves, and those entries go on the pending list under its name; worker-02 reading a millisecond later receives the entries after those. Two live workers never receive the same entry.

The exception is the sentence that the rest of the chapter is about. An entry stays on the pending list until it is acknowledged, and a worker that dies mid-handler acknowledges nothing. From Redis's side, that entry is still delivered and still worker-01's, and it will stay that way until something claims it. The group tracks delivery, not completion; completion is what the acknowledgement adds, and it comes from the worker's code, at the moment the worker chooses.

The Loop

The worker's loop reads, and for each entry it parses the payload into a typed job, runs the handler, and acknowledges. The parse is the boundary of Topic 14 of Chapter 3 arriving from a different direction: the fields on the stream are bytes written by another process, and the malformed order id that crashed every worker in turn in the spring was a payload that was trusted instead of parsed. A payload that fails to parse goes to the dead-letter stream of Topic 46 and is acknowledged on jobs, because it will not parse better on the fifth delivery.

slots = asyncio.Semaphore(8)                              # at most 8 jobs in flight per worker

async def consume(redis, name):                             # name is the host: "worker-01"
    while True:
        batches = await redis.xreadgroup("workers", name, {"jobs": ">"}, count=10, block=5000)
        for _, entries in batches:
            for entry_id, fields in entries:
                await slots.acquire()
                asyncio.create_task(handle(redis, entry_id, fields))

async def handle(redis, entry_id, fields):
    try:
        job = parse_job(fields)                             # bytes from another process: typed or refused
        await run_with_policy(job, entry_id)                # the handler; CPU work inside goes to a thread
    except MalformedJob as e:
        await dead_letter(redis, entry_id, fields, e)       # jobs:dead, with the error (Topic 46)
    finally:
        await redis.xack("jobs", "workers", entry_id)     # after the work; a crash above leaves it pending
        slots.release()

The loop asks for up to 10 entries and waits up to 5 seconds for the first, then hands each entry to a task, holding at most 8 in flight so that a burst of 12,000 jobs does not become 12,000 tasks. Each task parses the fields, runs the handler through the retry wrapper of Topic 46, and acknowledges. The acknowledgement is in the finally clause and nothing above it can skip it, but a process that dies never reaches it, which is the property the design wants: a crash at any line before the ack leaves the entry pending and claimable, not lost. Two details are easy to miss. The BLOCK read is the one Redis call in the codebase whose socket timeout is longer than the 100 milliseconds Chapter 7 set for Redis, because waiting is its job. And the dead-letter branch acknowledges too, since the poison entry has been moved and must not be delivered again.

Pending and Reclaiming

XPENDING jobs workers - + 10 lists the first 10 entries that were delivered and not acknowledged: the id, the consumer, the milliseconds since delivery, and the delivery count. On a healthy night the list is short and every idle time is under 5 seconds. An entry idle for 3 minutes under worker-01 is either a job that is genuinely slow or a worker that is gone, and from Redis's side those look the same. The reclaim loop decides by threshold: XAUTOCLAIM jobs workers worker-02 60000 0-0 COUNT 10, run every 30 seconds by every worker, takes ownership of up to 10 entries idle for more than 60 seconds, from the start of the list, and returns them as if worker-02 had just read them. Their delivery count goes up by 1, which Topic 46 reads as the poison counter.

The threshold is the decision. 60 seconds is 15 times the render's P95 of 4 seconds, so a live worker in the middle of a slow job is not robbed of it; a threshold of 5 seconds would reclaim jobs from workers that are fine, and every reclaimed job would run twice. The other half of the design is what a worker does on restart: before reading with >, worker-01 reads its own pending list with the id 0, which returns the entries still assigned to its name from before it died, and works through them first. Between the two, no entry waits forever. This is at-least-once delivery, stated as a mechanism: every entry is handled, and an entry whose worker died after the work and before the ack is handled twice.

Concurrency and Ordering

One stream, several workers, and no ordering guarantee across them. worker-01 receives the entry for order 4471's placement and worker-02 receives the one for its refund a second later; if the placement's render is a 4-second job and the refund is 50 milliseconds, the refund finishes first. Even one worker with 8 tasks in flight finishes entries in an order the stream did not promise. The book's answer is not to serialize the workers. Jobs that must run in order carry the order id, and the handler reads the current state of the order and acts on that: a refund handler that finds tickets cancels them, a placement handler that finds an order already refunded renders nothing. Topic 45 makes this the general rule.

The second concurrency question is inside one worker. The render is 4 seconds of CPU, and a worker whose loop ran it inline would stop reading for 4 seconds, which at 8 tasks in flight is not a loop at all. So the handler calls asyncio.to_thread(render_pdf, order), the render runs in the thread pool of Topic 04 of Chapter 1, 8 threads on the free-threaded build, and the loop's own thread stays free to issue the next XREADGROUP, answer the reclaim loop and send acknowledgements. One slow render then costs one thread for 4 seconds and nothing else.

Trimming

An acknowledged entry leaves the pending list and stays on the log. Nothing about the group deletes it, so a stream that is only ever appended to is Redis memory that grows by every job ever queued: at a few hundred bytes per entry and 12,000 jobs on an on-sale night, a season of nights is hundreds of megabytes that nothing reads. MAXLEN ~ 100000 on every XADD keeps the stream to roughly its newest 100,000 entries, trimming whole internal nodes at a time, which is why the tilde is there; an exact MAXLEN would trim on every add and cost more than it saves. The number is far above any backlog Stagedoor tolerates, because trimming does not respect the pending list: an entry trimmed while still unacknowledged is a job with an id and no fields, and the reclaim loop returns those as deleted rather than as work.

The dead-letter stream of Topic 46 is trimmed separately and much more slowly, at 10,000, because its entries are the ones a human has not read yet. What happens without any trimming is the failure the sessions of Chapter 5 are exposed to: Redis reaches its memory limit, the eviction policy chooses the keys that have been idle longest, and those are the sessions, so a stream that nobody trimmed signs the buyers out.