Scheduling, Retries and Dead Letters

Some jobs run at a time: expire the holds every minute, reconcile with Payrail at 03:00, purge idempotency keys every hour. Some jobs fail and should try again: the mail provider returned 503 and will not in 10 seconds. Some jobs fail and must never try again: the seat label that crashes the renderer will crash it on the fifth attempt exactly as it did on the first. The worker needs a scheduler, a retry policy per job kind, and a dead-letter stream where the poison jobs go to be read by a human in the morning.

The three are one design, and they share the loop of Topic 44. The scheduler adds entries to the same jobs stream the workers consume, a retry is a job re-added with a delay and a higher attempt count, and a dead letter is the same entry moved to jobs:dead with its error attached. Nothing is a special case in the consumer, which is what makes the consumer 20 lines long. The hold-expiry job is the running example, because it is the one whose absence Chapter 6 promised to fix here.

Scheduled Jobs

A scheduler loop runs inside the worker, wakes every second, and enqueues whatever is due: expire_holds every minute, reconcile_payments at 03:00, purge_idempotency_keys on the hour for rows older than 24 hours, and purge_outbox nightly for rows published more than 7 days ago, in batches of 1,000 as Chapter 7 specified. The scheduled kinds go on the stream like any other, with a job_id the scheduler mints in place of the relay's outbox_id, and the workers treat the two alike. The hold-expiry handler is the sweep that Topic 34 of Chapter 6 relies on to keep at most one hold row per seat: it deletes every hold whose time has passed and returns the seats to the map.

async def expire_holds(job):
    async with conn.transaction():
        rows = await conn.execute(
            "DELETE FROM holds WHERE expires_at < now() RETURNING seat_id")
        seat_ids = [r.seat_id for r in await rows.fetchall()]
        await conn.execute(
            "UPDATE seats SET status = 'available', version = version + 1 "
            "WHERE id = ANY(%s) AND status = 'held'", (seat_ids,))
    # a second run a millisecond later deletes nothing and updates nothing

# the scheduler, on every worker, once a minute: only the holder of the lock enqueues
if await redis.set("sched:expire_holds", name, nx=True, ex=55):
    await redis.xadd("jobs", {"kind": "expire_holds", "job_id": uuid4().hex, "attempts": 0})

The handler deletes the expired holds in one statement, collects the seat ids the delete returned, and sets those seats back to available in a second statement that touches only seats still marked held, under one transaction. Run twice in the same minute, the second run finds nothing expired and changes nothing, which makes the job safe under every duplicate this chapter has described. The last three lines are the scheduler's guard, and they exist because Stagedoor will run more than one worker after Topic 47. Every worker's scheduler wakes at the same second; each tries SET sched:expire_holds worker-01 NX EX 55, which succeeds for exactly one of them and expires 55 seconds later, 5 seconds before the next tick; only the one that won adds the job. The alternative is a single scheduler process, which needs no lock and is one more thing to keep alive; Stagedoor takes the lock, because the workers are already there.

Delayed Jobs

"Send the reminder 24 hours before the event" is a job with a run time and no schedule. The stream has no native delay: an entry added now is delivered now. So the delay is a sorted set named delayed: ZADD delayed with the run time in epoch seconds as the score and the job's fields, serialized, as the member. The scheduler loop, on its one-second wake and under its lock, reads what is due with ZRANGE delayed 0 1758286400 BYSCORE LIMIT 0 100, the upper bound being the current time, adds each member to jobs, and removes it from the set. The book shows this one pattern and does not survey the others.

The move from the set to the stream is two commands on the same Redis and no transaction between them, so a crash between the XADD and the ZREM delivers the job now and again on the next tick. It is the relay's duplicate on a smaller scale, and it is tolerated by the same idempotent handler rather than by any cleverness in the scheduler. Retries use the same set, which is the next section.

Retry Per Kind

A retry is a decision about the job, not about the error. The email send retries 5 times with the full-jitter backoff of Topic 38 of Chapter 7, 1 second doubling to a cap of 60, because a mail provider's 503 is a condition that passes. The PDF render retries once, because a render that failed on an order's data is unlikely to succeed on the same data, and the one retry covers a seat-map read that timed out on a replica busy with reports. The reconciliation does not retry at all; the next night's run covers it, and a retry of a job that walks every payment of the day is a second walk on the same night. The policy lives in the kind's definition, next to the handler, and the loop's wrapper reads it; a handler never contains a retry loop of its own, because a handler that retries inside a loop that also retries is the 27 attempts Chapter 7 counted.

The wrapper is where the attempt count lives. On a failure it reads the kind's policy and the entry's attempts field, and if another attempt is allowed it adds the job to delayed with attempts plus 1 and a score of now plus the backoff, then lets the loop acknowledge the original. The stream never holds a sleeping task; a job waiting 60 seconds for its fourth attempt is a member of a sorted set, and the 8 slots on the worker are doing other work. When the count reaches the limit, the wrapper does not re-add. It dead-letters.

The Dead-Letter Stream

A job that has exhausted its retries, or whose payload failed to parse at the loop's edge, is added to jobs:dead with everything it had plus the error and the attempt count, and then acknowledged on jobs. Nothing is lost, because the entry is on a stream that is trimmed at 10,000 and read by nobody but people. Nothing loops forever, because the ack on jobs means the group will not deliver it again. The malformed seat label from the spring is the canonical resident.

XRANGE jobs:dead - + COUNT 1
1) 1) "1758203641877-0"
   2)  1) "kind"           2) "render_tickets"
       3) "order_id"       4) "4471"
       5) "outbox_id"      6) "88210"
       7) "attempts"       8) "2"
       9) "error"         10) "SeatLabelInvalid: '14-C' is not a row letter and a seat number"
      11) "traceback"     12) "  File stagedoor/tickets/render.py, line 88, in barcode_for ..."
      13) "request_id"    14) "7f3a9c2e"
      15) "trace"         16) "4bf92f35"

The entry is the original job with three fields added: the attempt count, which says the render was tried twice as the policy allows; the error, which names the seat label the barcode encoder refused; and the traceback, which names the line. The request id and trace are still there, so the checkout that queued it is one search away in Chapter 13. What Marek does with it is a data fix, an organizer's seat map with a label typed as 14-C instead of 14C, and then the replay of the last section. What the alert on jobs:dead did is more important than what the entry says: Topic 47 puts its length on the dashboard, and a length that went from 0 to 1 at 21:14 is a page, because every entry there is a buyer without a ticket.

Poison Messages

A poison message is the job that kills the worker rather than failing inside it. In the spring it was an order id that could not parse, trusted by a handler that then dereferenced it, and every worker that claimed the entry died in turn, restarted, and claimed it again, which is the loop Topic 14 of Chapter 3 called Chapter 8's poison message. Two things stop it now. The loop's wrapper catches every exception a handler raises, so an unhandled error is a failed attempt and not a dead process, and the attempt counter decides its fate. And for the failure no wrapper can catch, the out-of-memory kill in the middle of a render, the counter that survives the process is the group's own: XPENDING reports how many times each pending entry has been delivered, the reclaim loop reads it before XAUTOCLAIM, and an entry delivered 3 times goes to jobs:dead without being run again.

The wrapper never raises. It returns, or it has re-added the job to delayed, or it has dead-lettered it, and in all three cases the loop's finally acknowledges the original entry. That invariant is what lets the consumer be simple: the only way an entry stays pending is a process that died, and the only way a process dies repeatedly on one entry is the delivery count, which has its limit.

Replaying

A dead letter that was a code bug is fixed by a deploy, and a dead letter that was a data bug is fixed by an organizer correcting a label. Either way the job then needs to run, and the replay is an XADD jobs with the entry's original fields, attempts reset to 0, and the dead entry deleted from jobs:dead with XDEL. A small tool does it for one id or for every entry of a kind, because after a bug in the barcode encoder there are 40 of them and not 1. Nothing in the tool knows what the first run did before it failed, and it does not need to: Topic 45 wrote the handlers to reach a state, so a render whose first attempt stored the PDF and died before recording the key stores it again and records it, and a send whose first attempt reached the provider is deduplicated by the provider. Idempotency was built for the duplicate the queue produces, and replay is the second reason it was worth building.