Queues and logs.

A queue is a to-do list that sits between two programs: one side writes down jobs, the other side picks them off and does them whenever it’s free, and the list remembers what’s still waiting even if either side crashes.

You’ll want one the first time a slow job is tied to someone waiting for it. Re-scoring a whole corpus with a new ranker checkpoint, summarizing every document a user imports, running a big batch of evals through the Spark overnight: done as one long loop, a single bad document or a restart halfway through leaves you not knowing what finished, and starting over. Done through a queue, a crash costs you one job’s worth of redo. “Put a queue in between” is also one of the most common moves in any system design interview, and the follow-up questions are exactly the ones this page answers.

It builds on Waiting vs working, where you met the rule linking how fast work arrives, how long each piece takes and how much is in progress at once, and on Rate limits and backpressure, where you saw a big enough waiting room hide the fact that you’re overloaded. Transactions helps for the part about making a repeated job harmless. Next, Data a model can eat uses the same ideas at training-data scale, and Batching is what a worker should do with a pile of jobs on a GPU.

Today

• Explain what a queue between two programs is, and the four reasons systems use one: the two sides run at different speeds, traffic arrives in bursts, failed work can be retried without the user knowing, and one event can feed many readers.
• Say exactly what happens when a worker dies halfway through a job: how the worker signals “done”, how the job is loaned out with a deadline so a dead worker’s job comes back, and where a job goes when it fails every single time.
• Explain the three promises a queue can make (a job runs at most once, at least once, or exactly once), why the last one is mostly a marketing word, and why the real answer is a worker that can safely do the same job twice.
• Tell a to-do-list queue (Celery, RQ, SQS) apart from a keep-everything event history (Kafka), where readers track their own position, the history is split into lanes, and order only holds inside one lane.
• Work out how many workers you need, watch how far behind they’re falling, and run a stand-in for your ranker over thousands of documents through a queue that survives workers being killed without warning.

You’ll want Waiting vs working (the arrivals-times-time rule and the fake LLM) and Rate limits and backpressure (why a bottomless waiting room hides overload, and retrying with growing pauses); Transactions helps for the “safe to do twice” part. Plan on one evening to read and one or two to build.

The Problem

You add a feature to the ranking service: for every document a user saves, call an LLM to write a two-line summary, then run the cross-encoder to score it against the user’s saved queries. The obvious code does both inside the request handler. It works on your laptop. Each save takes about 6 seconds, and the user stares at a spinner for all of it.

Then three things happen in one week.

A customer imports 4,000 documents at once. Your API server suddenly has 4,000 requests each holding a connection open for 6 seconds, the LLM provider starts answering 429, and because the handler retries inline, the requests take 30 seconds and the load balancer times them out. The user retries the import. Now there are 8,000.

You deploy a fix mid-afternoon. The rolling restart kills every in-flight request. Some summaries were written, some weren’t, and nothing records which.

And the research team asks you to re-score the whole corpus, 5 million documents, with the new ranker checkpoint. You write a loop. It dies at 3.2 million because one document has a malformed byte that crashes the tokenizer. You don’t know which 3.2 million finished. You start over.

All three have the same shape. Slow, failure-prone work is welded directly to whatever asked for it. If the asker is slow, impatient, or dead, the work suffers; if the work is slow or broken, the asker suffers. The fix is to put a buffer in between, a to-do list that both sides can read and write at their own pace and that remembers what’s done. That’s a queue.

The Idea

A to-do list between two programs

You already know the shape from a restaurant. The waiter doesn’t cook. They write the order on a ticket, clip it to the rail, and go back to the dining room. The cooks take tickets off the rail when they’re free. If ten tables order at once, the rail gets long, but nobody in the dining room is kept waiting at the kitchen door, and nobody in the kitchen is rushed.

In software, the waiter is the producerproducerThe program that creates jobs or messages and puts them on the queue. Your API server, usually., the cooks are consumerconsumerA program that takes messages off the queue and does something with them. When the something is a unit of work, people call it a worker.s (or workers), and the rail is a message queuemessage queueA separate service that holds messages from producers until consumers take them. The producer doesn’t need to know who consumes, or whether they’re up right now., run by a brokerbrokerThe server that stores the queue and hands messages out. Redis, RabbitMQ, SQS, and Kafka are all brokers of some kind.. A message is just bytes, usually a small JSON blob that says what to do: {"doc_id": 81723, "task": "summarize"}. Not the document itself; a pointer to it.

That one change buys four separate things, and it’s worth knowing them by name, because each is a reason someone will give you in a design review.

1. Decouple speed. The API can accept 2,000 saves a second; the workers can summarize 20. Without a queue, the slow side sets the pace for the fast side. With one, the API stays fast and the backlog lives somewhere designed to hold it.
2. Absorb spikes. Traffic arrives in bursts: an import, a cron job, Monday morning. The queue stores the burst and the workers work it off over the next few minutes. You size the workers for the average rate, not the peak, as long as the backlog drains before the next burst.
3. Retry independently. If a summary fails because the LLM returned 503, the worker retries it later. The user’s request finished long ago and never knew. And if the API deploys, the queued jobs don’t care.
4. Fan out. One event, “document saved”, can be read by the summarizer, the ranker, the search indexer and the analytics pipeline, each on its own schedule. The producer writes once and doesn’t need to know any of them exist. Adding a fifth reader doesn’t touch the producer.

The price: the work is now asynchronous. The user gets “accepted” instead of the answer, so you need a way to report the result later (poll GET /jobs/<id>, a webhook, a websocket push). And you’ve added a new system that can fail. Most of this lesson is about making that new system fail safely.

How many workers? Little’s Law again

In Waiting vs working you met Little’s Law: in flight = arrival rate × time each one spends inside. A worker pool is the cleanest case of it there is. Each worker handles one job at a time, so “jobs in flight” is “busy workers”.

The same arithmetic sizes a batch job. Re-scoring 5 million documents at 20 ms each is 100,000 worker-seconds. With 16 workers that’s about 6,250 s, an hour and three quarters, if nothing else is the bottleneck. With 1 worker it’s 28 hours. You will check where “if nothing else is the bottleneck” stops being true in Measure.

Watch it

Here is a pool of workers draining a queue. Jobs arrive at random at the rate you set; each takes about half a second, so one worker does about 2 jobs per second. Try these in order:

1. Leave it at 6 arrivals/s and 4 workers (capacity 8/s). The queue stays short. Now drop to 3 workers (capacity 6/s, equal to arrivals). Watch the lag graph wander up and not come back down reliably.
2. Back to 4 workers, press Spike. The queue jumps by 60 and then drains at capacity minus arrivals, 8 − 6 = 2 jobs per second, so about 30 seconds. That’s the spike being absorbed.
3. Press Kill -9 a worker while it’s busy. Its job doesn’t vanish: it sits in “leased, owner dead” with a countdown, then reappears at the head of the queue, filled in dark (a retry). Count: redelivered goes up, lost stays 0.
4. Turn on Ack before work and kill again. Now the job is simply gone. Lost goes up.

Two lessons are hiding in step 3 and 4. The first is the difference between “the broker handed you the job” and “the job is done”, which is the whole subject of the next section. The second: in step 3, there’s a 3-second gap where the job is neither waiting nor running. That gap is the price of surviving crashes.

Handing over a job: acks and leases

When a worker takes a job, the broker has a decision to make: delete it now, or keep it until the worker says it’s finished? That “I’m finished” message is an ackackAcknowledgement. The consumer telling the broker “I’ve handled this message, you can forget it.” Until the broker gets one, it assumes the message might still need doing..

If the broker deletes on handover, a worker crash loses the job. If the broker keeps it until the ack, it needs to know when to give up waiting, because a crashed worker never sends anything. A dead process doesn’t announce that it’s dead. From the broker’s side, “still working on it” and “was kill -9’d ten minutes ago” look identical: silence.

So the broker gives each handed-out job a leaseleaseTemporary ownership of a job with a deadline. If the holder doesn’t finish (ack) or renew before the deadline, the job is given to someone else.. Amazon SQS calls the lease length the visibility timeoutvisibility timeoutSQS’s name for the lease: after a consumer receives a message, it’s hidden from other consumers for this long. If not deleted by then, it becomes visible again and gets redelivered. Default 30 s.: once a worker receives a message, it becomes invisible to everyone else for, say, 30 seconds. If the worker deletes it (the ack) in time, it’s gone for good. If not, it becomes visible again and some worker gets it. That’s redeliveryredeliveryThe broker handing the same message out again because the previous attempt never acked. The consumer gets a message it, or a dead colleague, may have already partly processed..

Two practical details follow from leases.

The lease must outlast the job. If a job takes 90 seconds and the lease is 30, the broker hands the same job to a second worker while the first is still happily working on it, then a third. Nothing crashed, and the job runs three times. Long jobs either set a long lease, or heartbeat: every so often the worker says “still on it” and the broker pushes the deadline back (SQS: ChangeMessageVisibility). Celery’s Redis transport has exactly this trap: its visibility_timeout defaults to one hour, and a task scheduled or running longer than that gets delivered again.

A short lease means fast recovery, a long lease means slow recovery. When a worker really does die, its job sits untouched until the lease expires. In Figure 002 that was 3 seconds. With a one-hour lease, a crash means an hour’s delay for that job. You pick the lease from the job length (longest normal job, times a safety factor) and heartbeat if the spread is wide.

At-most-once, at-least-once, and the myth of exactly-once

Those three rows in Figure 003 have names, and you’ll hear them in every design discussion about messaging.

• at-most-onceat-most-onceEach message is processed zero or one times. Never duplicated, but can be lost. What you get when you ack (or delete) before doing the work.: ack first, then work. A crash loses the job. Fine for things where losing one is cheaper than doing one twice: a metrics sample, a “user is typing” ping.
• at-least-onceat-least-onceEach message is processed one or more times. Never lost (while the broker survives), but can be duplicated. What you get when you ack only after the work is done.: work first, then ack. A crash means a redelivery. Nothing is lost, but some things happen twice. This is the default for almost every serious queue.
• exactly-onceexactly-onceEach message has its effect exactly one time, even with crashes. Not achievable as a delivery guarantee across arbitrary systems; achievable as an effect, by combining at-least-once delivery with deduplication.: what everyone wants.

Here’s why you can’t get exactly-once delivery. Look at the bottom row again. The worker saved the result and then died before the ack reached the broker. Or the ack was sent and the network dropped it. Either way, the broker sees silence, and silence after a save looks exactly like silence before a save. The broker has to choose: redeliver (risk a duplicate) or not (risk a loss). There is no third option, because the broker can’t see inside the worker. Move the ack earlier and you’ve just moved the gap. This is a cousin of the classic “two generals” problem: two parties talking over a channel that can lose messages can never be sure they agree.

So the industry’s real answer is: at-least-once delivery, plus a consumer that makes duplicates harmless. An operation is idempotentidempotentDoing it twice has the same effect as doing it once. “Set score to 0.83” is idempotent; “add 0.83 to the total” is not. when repeating it changes nothing. You make a consumer idempotent with an idempotency keyidempotency keyA unique ID attached to a piece of work, so the receiver can recognize a repeat and skip it. For a job, the job ID or a natural key like (doc_id, model_version)., the same trick you saw for API retries in the API lifecycle:

• Write the result and the “done” mark in one transaction. INSERT INTO scores (doc_id, model, score) ... ON CONFLICT DO NOTHING, where (doc_id, model) is the primary key. The second delivery’s insert is a no-op. If you also bump a counter, bump it only when the insert actually inserted, in the same transaction. That’s Transactions earning its keep: the effect and the record of the effect can’t come apart.
• Make the operation itself a “set”, not an “add”. “Document 81723’s score is 0.83” can be written any number of times. “Increment the user’s processed count” can’t, unless you dedupe first.
• For external side effects (sending an email, charging a card, calling a paid API), pass the key along if the other side supports it (Stripe takes an Idempotency-Key header), or record “sent” before you’d risk sending twice and accept that a crash in between might mean zero sends. Somewhere, someone picks at-most-once or at-least-once. You just get to choose where.

When Kafka advertises “exactly-once semantics” (added in 2017), it means something precise and narrower: a producer’s retries don’t create duplicates in the log, and a job that reads from Kafka and writes back to Kafka can commit its output and its read position atomically. That’s real and useful. But the moment your consumer touches anything outside Kafka (a database, a model API, an email), you are back to at-least-once plus idempotency. People call the combination effectively-onceeffectively-onceAt-least-once delivery plus an idempotent consumer, so each message’s effect happens once even when the message is delivered more than once. The honest name for “exactly-once”..

Poison messages and the dead-letter queue

Remember the document with the malformed byte that crashes the tokenizer. Under at-least-once, it’s worse than you think. The worker claims it, crashes, the lease expires, another worker claims it, crashes… A poison messagepoison messageA message that fails every time it’s processed, usually because of bad data or a bug, not a temporary outage. Retrying it forever wastes workers and can block the queue. can kill your whole pool in a loop, one worker at a time, forever.

The fix is to count attempts. Every claim bumps the job’s attempt counter. When it fails (or its lease expires) and the count has reached a limit, say 3 or 5, the broker moves it aside into a dead-letter queuedead-letter queueA separate queue where messages go after failing too many times. Nobody processes it automatically; a human looks, fixes the bug or the data, and re-queues them. Often shortened to DLQ. instead of back into the main queue. SQS calls this a redrive policy with maxReceiveCount; RabbitMQ calls it a dead-letter exchange; Celery leaves it to you.

Three rules make a DLQ useful instead of a graveyard:

1. Store why it died. The last exception and traceback, the attempt count, the worker that saw it. Otherwise you’re debugging blind.
2. Alert on it. A DLQ that grows silently is the same as dropping messages.
3. Have a re-drive tool. After you fix the bug, you want one command that moves everything from the DLQ back to the main queue. Since the consumer is idempotent, re-driving something that half-succeeded is safe.

And separate retryable failures from permanent ones. A 503 from the LLM provider should be retried with backoff (and jitter, from Rate limits); a UnicodeDecodeError will fail identically on every attempt, so it can go straight to the DLQ.

Job queues and logs: two different machines

Everything so far describes a job queuejob queueA queue where each message is a unit of work, handed to exactly one worker, and deleted once acked. Celery, RQ, Sidekiq, SQS, RabbitMQ.: each message goes to one worker and is deleted when acked. Celery, RQ (Redis Queue) and Sidekiq are libraries built on brokers like Redis or RabbitMQ; SQS is Amazon’s hosted version. Under the hood in Redis, a queue is often just a list: producers push to one end, workers pop from the other.

Kafka is a different machine, and the difference matters. It’s a loglogAn append-only, ordered sequence of records. New records only go on the end; nothing is changed in place. Each record gets a position number, its offset.. Messages aren’t deleted when read. They’re appended to the end of a file and kept for a fixed time (the default is 7 days) or up to a size limit, whether anyone has read them or not. Each record gets a number, its offsetoffsetA record’s position in a Kafka partition: 0, 1, 2, … Consumers track “I’ve processed up to offset N” instead of deleting messages., and a consumer doesn’t ack individual messages; it remembers how far it has read, by periodically saving its current offset (a commit).

That design gives a log three powers a job queue doesn’t have:

• Many independent readers. The summarizer, the search indexer and the analytics loader each keep their own position in the same log. A slow reader doesn’t slow the others, and a new reader can start from the beginning.
• Replay. Found a bug in the summarizer that ran for two days? Fix it, reset its offset to two days ago, and it reprocesses everything. With a job queue, those messages are gone the moment they were acked.
• History as data. The log is a record of what happened, in order. That’s Kreps’ point: you can rebuild a search index, a cache, or a whole database table by reading the log from the start.

The job queue has its own advantages: per-message acks, per-message retries and DLQs, no need to think about positions, and simple scaling by just adding workers. Rule of thumb: if each message is a task to get done once, use a job queue. If each message is an event (a fact about something that happened) that several systems care about, use a log.

Partitions, consumer groups, and why ordering is per partition

One log on one disk can only go so fast. So a Kafka topictopicA named stream of records in Kafka, like “documents-saved”. Physically split into partitions. is split into partitionpartitionOne independent, ordered log inside a topic. Records within a partition have a strict order; records in different partitions have none. Partitions are spread across brokers.s, each its own little append-only log, living on different machines. When a producer writes a record, it picks a partition, usually by hashing the record’s key. Same key, same partition, every time.

On the reading side, consumers join a consumer groupconsumer groupA set of consumers that share the work of reading a topic. Kafka gives each partition to exactly one consumer in the group, so each record is processed once per group.. Kafka gives each partition to exactly one member of the group. That’s how a log does load balancing: not message by message, but partition by partition.

Three consequences you should be able to say out loud:

1. Ordering is only per partition. Within P1, the events for document 17 are in the order they were written. Across partitions there is no order at all: C1 might process an event from P0 written an hour after an event C2 hasn’t reached yet in P1. If order matters for some entity (the edits of one document, the events of one user), make that entity the key, so all its records share a partition. Global ordering would mean one partition, which means one consumer.
2. Parallelism is capped by partitions. Three partitions, at most three busy consumers per group. The fourth sits idle as a hot spare. Pick the partition count for the parallelism you’ll need later; adding partitions afterwards changes which partition each key hashes to, which breaks per-key ordering across the change.
3. One slow message blocks its whole partition. A consumer reads its partition in order. If one record takes a minute, everything behind it in that partition waits, even though other consumers are idle. A job queue doesn’t have this problem, since any free worker can take the next job. This is the deepest practical difference between the two, and why “Kafka as a job queue for slow, uneven jobs” often hurts. (Kafka 4.x is adding “share groups”, queue-style consumption on top of a topic, for exactly this reason. Check the docs for its status.)

Kafka’s delivery semantics are the same story as before, told with offsets. If you commit the offset before processing, a crash skips records: at-most-once. If you commit after, a crash replays from the last commit: at-least-once, possibly a few hundred duplicates. The client’s default, auto-commit every 5 seconds in the background, can do either depending on timing, which is why serious consumers turn it off and commit after they’ve processed.

Consumer lag: the one number to watch

How do you know whether a queue is healthy? Not by its size alone; a big backlog that’s shrinking is fine. You watch consumer lagconsumer lagHow far behind the consumers are. In Kafka, the newest offset minus the committed offset, per partition. In a job queue, the number of waiting messages, or better, the age of the oldest one..

• In Kafka, lag for each partition is the latest offset minus the group’s committed offset: “how many records written but not yet processed”.
• In a job queue, it’s queue depth, and even better, the age of the oldest message. Depth tells you how much; age tells you how late. 10,000 tiny jobs might be 5 seconds of work; 100 huge ones might be an hour. SQS publishes this as ApproximateAgeOfOldestMessage.

Lag has three shapes, and you saw all three in Figure 002. Flat near zero: keeping up. Spike then decay: a burst being absorbed, draining at (capacity − arrival rate). A steady ramp: arrivals exceed capacity, and it will never recover on its own. The ramp’s slope is literally the shortfall. If lag grows by 600 jobs a minute, you’re 10 jobs per second short, which by Little’s Law is 10 × (job time) more workers.

The time to drain a backlog B, once you add capacity, is B / (capacity − arrival rate). Note what’s on the bottom: with a backlog of 100,000 jobs, capacity 110/s and arrivals 100/s, draining takes 10,000 s, almost three hours, even though you’re “over capacity”. Headroom is what clears backlogs.

And notice what a queue does to backpressure. In Rate limits you saw that an unbounded buffer hides overload: the producer never feels the pressure. A queue is a very large buffer, on purpose. It is your choice to absorb rather than push back, which is fine for spikes and terrible for sustained overload. So bound it (a max length, or reject new jobs when the oldest is older than your deadline), and alert on lag. Otherwise the queue fills disk or memory and the failure is a lot bigger than a 429 would have been.

The outbox pattern: writing to the database and the queue at once

One more trap, and it shows up the first day you wire a queue into a real app. The save handler does two things: insert the document row into Postgres, and enqueue a “summarize this” job.

db.execute("INSERT INTO documents ...")   # 1
db.commit()
queue.enqueue({"doc_id": doc_id})         # 2

If the process dies between 1 and 2, the document exists and its job never will. Swap the order and the opposite happens: a job for a document that was rolled back. You can’t wrap both in one transaction, because the database and the broker are two different systems. This is the dual writedual writeWriting the same change to two systems (say, a database and a queue) as two separate steps. A crash between them leaves the systems disagreeing, and neither knows. problem.

The fix is the outbox patternoutbox patternWrite the outgoing message into an “outbox” table in the same database transaction as the business change. A separate relay process reads the outbox and publishes to the broker, deleting or marking rows once sent.: write the message to an outbox table in the same database transaction as the document. Now they either both exist or neither does. A separate small process, the relay, reads unsent outbox rows, publishes them to the queue, and marks them sent.

The relay can poll the table every few hundred milliseconds, or read the database’s own write-ahead log with a change data capturechange data captureReading a database’s internal change log (Postgres WAL, MySQL binlog) to stream every committed change to other systems. Debezium is the common tool. Often shortened to CDC. tool like Debezium. Notice that CDC is Kreps’ argument made literal: the database already has a log; you’re just letting other systems read it.

Which one do I reach for?

Batch inference is a queue workload

Back to the 5 million documents. A batch job is a queue with one giant spike at the start: all the arrivals happen at once, and lag is “how much is left”. Treat it that way and the 3.2-million-and-crash problem disappears.

• Each job is a chunk, not a document. Enqueue {"shard": 412, "start": 412000, "end": 413000}, not the documents themselves. One message per document means 5 million broker round trips; one per 1,000 documents means 5,000, and the worker can run the ranker on full GPU batches. Chunk size is a trade: bigger chunks are more efficient but redo more work when a worker dies mid-chunk.
• Results are keyed, so redelivery is free. Write scores with primary key (doc_id, model_version) and ON CONFLICT DO NOTHING (or overwrite; the score is the same either way). Re-running a chunk is harmless, so you never need to know exactly which ones finished. You just re-enqueue anything not in the results table.
• Poison data goes to the DLQ, not into a crash loop. The malformed document fails its chunk three times and gets parked, with its traceback, while the other 4,999 chunks keep moving. Even better, catch per-document errors inside a chunk and record them, so one bad document doesn’t send 999 good ones to the DLQ.
• Little’s Law sizes it, lag tracks it. Workers × throughput per worker tells you the finish time before you start. Lag over time tells you whether you’re on schedule, and its slope tells you when you’ll finish.

This is also what the batch APIs from model providers are under the hood: a durable queue of your requests, worked off whenever they have spare capacity, which is why they’re cheaper.

Build It

Everything goes in ~/workbench/queues/. You’ll build a Redis-backed job queue with leases, acks, retries and a dead-letter queue; move slow work out of a request; then try to break it with kill -9. Redis is used as a fast shared data-structure server here; you don’t need to know it beyond the handful of commands below.

brew install redis && brew services start redis    # or: docker run -d -p 6379:6379 redis:7
redis-cli ping                                     # PONG
mkdir -p ~/workbench/queues && cd ~/workbench/queues
python3 -m venv .venv && source .venv/bin/activate
pip install redis fastapi uvicorn httpx

# work.py: a fake ranker. CPU-bound, deterministic, ~20 ms per document.
import hashlib, os, time
ITERS = int(os.environ.get("ITERS", "20000"))

def score(doc: str) -> float:
    if "POISON" in doc:
        raise ValueError(f"cannot tokenize {doc!r}")
    h = doc.encode()
    for _ in range(ITERS):
        h = hashlib.sha256(h).digest()
    return h[0] / 255

def bench():
    t = time.perf_counter(); score("hello"); print(f"{(time.perf_counter() - t) * 1000:.1f} ms")

# setup_db.py: fresh results database
import os, sqlite3
if os.path.exists("results.db"): os.remove("results.db")
db = sqlite3.connect("results.db")
db.executescript("""
CREATE TABLE scores(job_id TEXT PRIMARY KEY, doc_id INT, score REAL);
CREATE TABLE scores_naive(job_id TEXT, doc_id INT, score REAL);
CREATE TABLE stats(n INT); INSERT INTO stats VALUES (0);
""")
db.commit()

@app.post("/docs", status_code=202)
def save(doc: Doc):
    job_id = f"doc-{doc.doc_id}"            # natural key: re-saving the same doc is a no-op job
    q.enqueue({"doc_id": doc.doc_id, "text": doc.text}, job_id=job_id)
    return {"job_id": job_id}

@app.get("/jobs/{job_id}")
def status(job_id: str):
    ...  # "done" + score if it's in results.db, else "pending"/"dead", from Redis

import json, time, uuid, redis
r = redis.Redis(decode_responses=True)
PENDING, INFLIGHT, DLQ = "q:pending", "q:inflight", "q:dlq"

_claim = r.register_script("""
local id = redis.call('RPOP', KEYS[1])
if not id then return false end
redis.call('ZADD', KEYS[2], ARGV[1], id)
local n = redis.call('HINCRBY', 'job:' .. id, 'attempts', 1)
return {id, redis.call('HGET', 'job:' .. id, 'payload'), n}
""")

def claim(lease_s: float):
    """Returns (job_id, payload_dict, attempts) or None."""
    res = _claim(keys=[PENDING, INFLIGHT], args=[time.time() + lease_s])
    ...

def enqueue(payload: dict, job_id: str | None = None) -> str: ...
    # write job:<id> (payload, attempts=0), then LPUSH onto pending.
    # Question for you: what if job:<id> already exists (same doc saved twice)?

def ack(job_id: str) -> bool: ...
    # remove from inflight and delete job:<id>. Atomic? What if the lease already
    # expired and someone else holds the job now? (Hint: ZREM returns 0 or 1.)

def extend(job_id: str, lease_s: float) -> bool: ...
    # heartbeat: push the deadline back, only if still in inflight (ZADD with XX)

def fail(job_id: str, error: str) -> None: ...
    # record last_error, make the lease expire now so reap() handles it

def reap(max_attempts: int = 3) -> tuple[int, int]: ...
    # every expired lease: attempts >= max_attempts -> DLQ, else back to pending.
    # Returns (requeued, dead). Must be atomic too: two workers reaping the same
    # expired job must not requeue it twice. Write it as a second Lua script.

import os, sqlite3, time, traceback
import q, work

IDEM = os.environ.get("IDEM", "1") == "1"
db = sqlite3.connect("results.db", timeout=30, isolation_level=None)
db.execute("PRAGMA journal_mode=WAL")   # lets many worker processes write safely

def apply(job_id: str, doc_id: int, s: float) -> None:
    # IDEM=1: in ONE transaction, INSERT OR IGNORE into scores, and bump stats.n
    #         only if a row was actually inserted (cursor.rowcount).
    # IDEM=0: plain INSERT into scores_naive and always bump stats.n.
    ...

while True:
    ...

# produce.py: python produce.py N   (every 1000th doc is poison)
import sys, q
q.r.delete(q.PENDING, q.INFLIGHT, q.DLQ)
for k in q.r.scan_iter("job:*"): q.r.delete(k)
n = int(sys.argv[1])
for i in range(n):
    doc = f"doc {i}" + (" POISON" if i % 1000 == 7 else "")
    q.enqueue({"doc_id": i, "text": doc}, job_id=f"doc-{i}")
print("enqueued", n)

# chaos.py: python chaos.py WORKERS   (CHAOS=0 to just run, no kills)
import os, random, signal, subprocess, sys, time, q
W, kills = int(sys.argv[1]), 0
spawn = lambda: subprocess.Popen([sys.executable, "worker.py"])
procs = [spawn() for _ in range(W)]
t0 = time.time()
while q.r.llen(q.PENDING) + q.r.zcard(q.INFLIGHT) > 0:
    time.sleep(0.3)
    if os.environ.get("CHAOS", "1") == "1":
        i = random.randrange(W)
        os.kill(procs[i].pid, signal.SIGKILL); procs[i].wait(); kills += 1
        procs[i] = spawn()                      # the "supervisor" restarts it
for p in procs: p.kill()
print(f"done in {time.time() - t0:.1f} s, kills={kills}, dlq={q.r.llen(q.DLQ)}")

# verify.py
import sqlite3
db = sqlite3.connect("results.db")
one = lambda sql: db.execute(sql).fetchone()[0]
print("idempotent rows:", one("select count(*) from scores"),
      "| naive rows:", one("select count(*) from scores_naive"),
      "| naive distinct:", one("select count(distinct job_id) from scores_naive"),
      "| counter:", one("select n from stats"))

docker run -d --name rp -p 9092:9092 docker.redpanda.com/redpandadata/redpanda:latest \
  redpanda start --mode dev-container --kafka-addr 0.0.0.0:9092 --advertise-kafka-addr localhost:9092
docker exec rp rpk topic create docs-saved -p 3
pip install confluent-kafka

Measure

1. Nothing lost, nothing doubled. Run the chaos test four ways and fill in the table. ACK_DELAY=0.01 widens the gap between “saved” and “acked” from microseconds to 10 ms, so kills land in it often enough to count.

What to expect: lost should be zero in every row: 1,998 distinct jobs scored plus 2 in the DLQ, because a job only leaves Redis on ack, and ack comes after the commit. With IDEM=1, rows and counter are exactly 1,998 every time. With IDEM=0, rows exceed distinct jobs by roughly the number of kills that landed between commit and ack. When I ran a version of this harness on an M4 Pro (with a Python stand-in for Redis, not real Redis), the IDEM=0, ACK_DELAY=0.01 run had 23 kills and produced 2,009 rows for 1,998 distinct documents, so 11 duplicates and a counter of 2,009; the IDEM=1 run with the same settings had 24 kills and exactly 1,998. With ACK_DELAY=0 the window is tiny and you may see zero duplicates: that’s the dangerous case, because the bug is still there, just rare enough to reach production.

2. Throughput vs worker count. CHAOS=0, 2,000 documents (no kills), for 1, 2, 4, 6, 8, 10, 12, 16 and 24 workers. Throughput is documents / elapsed seconds.

Then use your number. If one machine tops out at P docs/s, re-scoring 5 million documents takes 5,000,000 / P seconds. Little’s Law in reverse: to hit a deadline of D seconds, you need throughput 5,000,000 / D, and so that many divided by your per-worker rate, spread over enough machines to stay below each one’s knee. On the Spark the ranker would run on the GPU in batches, so the knee will be GPU throughput, not CPU cores, and chunk size becomes the dial.

3. Lag under a spike. Run 8 workers with no chaos and a producer that enqueues at a steady 200 docs/s (add a time.sleep to produce.py), then enqueue 5,000 extra at once. Sample LLEN q:pending every second (a 10-line script, or watch -n1 redis-cli llen q:pending) and plot it. It should look like the spike in Figure 002: a jump, then a straight-line decline with slope (capacity − 200) docs/s. Measure the slope and check it against your throughput table.

Break It

1. Lease shorter than the job. Set the lease to 0.01 s with 20 ms jobs, no kills. Every job’s lease expires mid-work, reap hands it to someone else, and the naive table fills with duplicates while workers are perfectly healthy. Then add a heartbeat (extend every half-lease, from a thread in the worker) and watch it stop. This is the most common queue bug in production, and it looks like “random duplicates under load”, because jobs slow down under load.
2. No attempt limit. Set max_attempts to a huge number. The poison jobs bounce between pending and in-flight forever, using a worker each time. With one poison message per 1,000 jobs and a slow failure (say the tokenizer hangs 10 s before crashing), a growing fraction of your pool ends up doing nothing but failing. Then put the limit back.
3. Ack before work. Move ack before apply and rerun the chaos test. Now you’ll be short: fewer than 1,998 rows, nothing in the DLQ for them, and no error anywhere. Silent loss is worse than duplicates, because nothing tells you.
4. Producer faster than consumers, no bound. Run 2 workers and a producer at 500 docs/s for a few minutes. Watch redis-cli info memory climb and the oldest job’s age grow without limit. Then add a guard to enqueue: if LLEN q:pending exceeds a limit, raise, and have the API return 503 with Retry-After. That’s backpressure from Rate limits, applied at the door of the queue.

Explain It Back

Answer out loud, in plain words, as if in an interview. Then check.

Go Deeper

• Jay Kreps, “The Log: What every software engineer should know about real-time data’s unifying abstraction” (LinkedIn Engineering blog, December 2013). The essay behind this lesson. Long, clear, and it will change how you see databases, replication and data pipelines. Expanded into the short O’Reilly book I Heart Logs (Kreps, 2014).
• Jay Kreps, Neha Narkhede, Jun Rao, “Kafka: a Distributed Messaging System for Log Processing” (NetDB workshop, 2011). The original Kafka paper: why LinkedIn built it, and the design choices (sequential disk writes, consumer-held offsets, no per-message acks) that made it fast.
• Martin Kleppmann, Designing Data-Intensive Applications, the chapter “Stream Processing”. The best single explanation of message brokers vs logs, delivery semantics, and change data capture, with the trade-offs laid out. The chapter on batch processing is worth reading alongside it.
• Apache Kafka documentation, “Design” section (kafka.apache.org/documentation). Covers persistence, the producer and consumer, and delivery semantics, written by the people who built it. Read the “Message Delivery Semantics” part twice.
• Amazon SQS Developer Guide: “Amazon SQS visibility timeout” and “Using dead-letter queues”. The clearest official write-up of leases, redelivery, and redrive policies, with the real defaults and limits.
• Tyler Treat, “You Cannot Have Exactly-Once Delivery” (Brave New Geek blog, 2015). A short, sharp argument for why delivery guarantees stop at at-least-once, and what people really mean by “exactly-once”.
• Neha Narkhede, “Exactly-once Semantics Are Possible: Here’s How Kafka Does It” (Confluent blog, 2017). The other side: what Kafka’s idempotent producer and transactions actually guarantee, and where the boundary is. Read it right after Treat.
• Chris Richardson, “Pattern: Transactional outbox” (microservices.io). The outbox pattern and its relatives (polling publisher, log tailing) in one page.

Connections

This lesson leans on Waiting vs working for Little’s Law and on Rate limits and backpressure for the idea that a buffer hides overload: a queue is a deliberately enormous buffer, so you need lag alerts and a bound to get the pressure signal back. The idempotent write is Transactions at work, and it’s the same idempotency-key trick as API retries in the API lifecycle. The “where’s the knee” measurement is the same shape as p99 and load testing, and the “is the bottleneck cores or something shared?” question comes from Processes, threads and the GIL, since each worker is a separate process exactly so CPU-bound scoring can use every core.

Looking ahead: Data pipelines is this lesson at training-data scale, where chunked, idempotent, replayable jobs are the whole design. Batching is what a worker should do with a chunk once it has one on a GPU. And Observability is where consumer lag and oldest-message age become dashboards and alerts.