Queues and backpressure, from first idea to production

A customer sends a WhatsApp message to a business. Meta delivers it to your server as a webhook and waits for a 200 OK. Writing a good AI reply takes several seconds. If you do all of that before answering the webhook, a slow model call becomes a slow webhook, and a provider that keeps seeing slow or failed deliveries retries them and eventually stops sending.

So you don’t. You save the message, put a note on a list that says “this conversation needs a reply”, answer 200 in a few milliseconds, and let a separate process work through the list. That list is a queue, and almost every backend I’ve worked on has one at its center.

Queues are easy to add and easy to get wrong in one specific way: they never refuse anything. When work arrives faster than you can do it, a queue doesn’t fail. It gets longer. Your error rate stays at zero while customers wait minutes for a reply. This article is about using queues on purpose, so that overload becomes something you can see and decide about.

By the end you’ll be able to explain acknowledgments and why queues deliver some messages twice, predict waiting time from two numbers, configure Celery so one slow task type can’t starve another, choose between Redis, SQS, RabbitMQ and Kafka, and design backpressure from an asyncio.Queue to an HTTP 503. Start at Beginner if you’ve never run a background job; Intermediate is where the arithmetic lives, and Expert is where outages come from. The last section walks through the queues in ezychat.io, the multi-tenant AI messaging product I build and run.

Beginner: what a queue is for

Saying “got it” now and doing the work later

Some work doesn’t need to happen while somebody waits. Sending a welcome email, resizing a photo, writing an AI reply to a message that’s already saved: the person who asked only needs to know the request was received. The work can happen a second or a minute later.

A queue splits one slow operation into two fast steps and one slow one:

1. Accept: write down what needs doing and answer immediately.
2. Work: some other process picks the job up and does it.
3. Finish: that process marks the job done.

This split is called decoupling. The thing that creates work and the thing that does it no longer have to run at the same speed, at the same time, or on the same machine.

Producers, consumers and the line between them

A restaurant kitchen works the same way. The waiter clips your order to a rail above the pass and goes to the next table. Cooks take tickets from the rail, oldest first. When it’s busy, the rail fills up; when it’s quiet, cooks wait. The vocabulary maps one to one:

• A message (or task, or job) is one ticket: a small piece of data that says what to do, like {"task": "send_welcome", "user_id": 42}.
• A producer creates messages. The waiter. In software: your web API, a webhook handler, a scheduler that fires every minute.
• A consumer takes messages and does the work. The cook. In software it’s usually called a worker: a long-running process that takes a message, does it, and takes the next.
• The queue is the rail. It holds messages in order until a worker takes them.
• The broker is the system that stores the queue: Redis, RabbitMQ, Amazon SQS, Kafka. It’s a separate service so that producers and workers on different machines can share it.

Most job queues are FIFO: first in, first out, like the ticket rail. It’s the fair default when every job matters about the same, and you’ll spend half of this article qualifying it.

Your first queue

Python’s standard library has a thread-safe FIFO queue. Here’s a kitchen with one cook:

import queue
import threading
import time

orders = queue.Queue()  # first in, first out, and safe to share between threads


def cook():
    """A worker: take the oldest order, make it, repeat."""
    while True:
        order = orders.get()          # waits here while the queue is empty
        if order is None:             # a sentinel meaning "kitchen's closed"
            break
        time.sleep(0.1)               # the actual work
        print("served", order)
        orders.task_done()


worker = threading.Thread(target=cook)
worker.start()

started = time.perf_counter()
for dish in ["laksa", "nasi lemak", "roti canai", "satay"]:
    orders.put(dish)                  # the producer returns immediately
print(f"took 4 orders in {(time.perf_counter() - started) * 1000:.2f} ms")

orders.join()                         # wait until every order is served
orders.put(None)
worker.join()

# took 4 orders in 0.01 ms
# served laksa
# served nasi lemak
# served roti canai
# served satay

The producer took four orders in a hundredth of a millisecond. The work took 400 ms, but none of it happened while the producer was waiting. That gap is the whole point.

This queue lives inside one Python process. If the process dies, every order on the rail dies with it. Real systems keep the queue in a broker, outside the processes that use it, which raises a new question: how does the broker know a job was actually done?

What happens when a worker dies: acknowledgments

Suppose a worker takes a message, starts the work, and then the machine is restarted for a deploy. If the broker deleted the message when it handed it out, that job is gone forever. Nobody will ever know it existed.

So brokers don’t delete a message when they hand it out. They hide it. The worker has to send an acknowledgment (an ack) when it’s finished, and only then is the message deleted. If no ack arrives within a time limit, called the visibility timeout, the broker assumes the worker died and makes the message visible again, so another worker can take it.

Here’s that logic in a few lines, so you can see exactly where the guarantee comes from:

from collections import deque


class Broker:
    """A queue that remembers what it handed out until the worker says it's done."""

    def __init__(self, visibility_timeout=30):
        self.ready = deque()          # waiting to be picked up, oldest first
        self.in_flight = {}           # message id -> (message, deadline)
        self.timeout = visibility_timeout
        self.next_id = 1

    def publish(self, body):
        self.ready.append((self.next_id, body))
        self.next_id += 1

    def receive(self, now):
        self.requeue_expired(now)
        if not self.ready:
            return None
        msg = self.ready.popleft()
        self.in_flight[msg[0]] = (msg, now + self.timeout)  # hidden, not deleted
        return msg

    def ack(self, msg_id):
        del self.in_flight[msg_id]    # only now is the message gone for good

    def requeue_expired(self, now):
        for msg_id, (msg, deadline) in list(self.in_flight.items()):
            if now >= deadline:       # the worker never acked: assume it died
                del self.in_flight[msg_id]
                self.ready.appendleft(msg)


emails_sent = []
broker = Broker(visibility_timeout=30)
broker.publish("welcome email to aisha@example.com")

# Worker 1 takes the message, sends the email, then crashes before acking.
msg_id, body = broker.receive(now=0)
emails_sent.append(body)
print("worker 1 crashed; ready:", len(broker.ready), "in flight:", len(broker.in_flight))

# 30 seconds later the message becomes visible again and worker 2 gets it.
msg_id, body = broker.receive(now=31)
emails_sent.append(body)
broker.ack(msg_id)
print("after worker 2; ready:", len(broker.ready), "in flight:", len(broker.in_flight))
print("emails sent:", len(emails_sent))

# worker 1 crashed; ready: 0 in flight: 1
# after worker 2; ready: 0 in flight: 0
# emails sent: 2

The job wasn’t lost. It also ran twice: worker 1 had already sent the email when it crashed, and worker 2 sent it again. Nothing about this broker is buggy. It’s the unavoidable consequence of the ack coming after the work.

At-least-once delivery, and why duplicates are your problem

There are only two places to put the ack, and each gives you a different guarantee:

• Ack before the work and a crash loses the job. That’s at-most-once delivery: every message is handled once or not at all. Fine for a metrics ping, wrong for a payment.
• Ack after the work and a crash repeats the job. That’s at-least-once delivery: every message is handled once or more. This is what nearly every job system offers, and what you should usually choose.

You’ll see brokers advertise exactly-once delivery. Read the small print. Kafka’s transactions can record a message’s effect once, for reads and writes that stay inside Kafka. But once your worker sends an email or charges a card, a crash between “did it” and “recorded that I did it” can always happen, and no broker can see into your email provider.

So accept at-least-once delivery and make the work idempotent: safe to do twice, because the second run notices the first and does nothing.

Absorbing a burst

The second reason to use a queue is that traffic is lumpy. A campaign goes out and two thousand people reply in a minute. A cron job enqueues ten thousand reminders at midnight. If your workers can handle the average load, a queue lets them handle a burst too, by spreading it out over time:

arrivals = [60] * 10 + [5] * 50     # per second: a 10-second burst of 60/s, then a quiet 5/s
capacity = 20                       # the workers can finish 20 tasks a second

waiting, peak, drained_at = 0, 0, None
for second, arrived in enumerate(arrivals, start=1):
    waiting = max(0, waiting + arrived - capacity)
    peak = max(peak, waiting)
    if second > 10 and waiting == 0 and drained_at is None:
        drained_at = second
    if second in (1, 5, 10, 15, 20, 30):
        print(f"t={second:>2}s  waiting={waiting}")
print("peak:", peak, "| empty again at t =", drained_at, "s")

# t= 1s  waiting=40
# t= 5s  waiting=200
# t=10s  waiting=400
# t=15s  waiting=325
# t=20s  waiting=250
# t=30s  waiting=100
# peak: 400 | empty again at t = 37 s

For ten seconds, work arrives three times faster than it can be done. Nothing fails. The queue grows to 400, then drains at 15 a second (20 done, 5 arriving) until it’s empty at 37 seconds. A job that arrived at the peak waited about 20 seconds.

The queue didn’t make the workers faster. It traded time for capacity, and that trade only works if the burst ends. If arrivals stay above 20 a second, the queue grows forever and every job waits longer than the one before. Most of this article is about that case.

sent = set()                       # in production: a table with a unique constraint, not memory
emails_sent = []


def send_welcome(msg_id, address):
    """Safe to run twice: the second run finds the first one's mark and does nothing."""
    if msg_id in sent:
        return "already sent"
    emails_sent.append(address)    # the side effect
    sent.add(msg_id)               # record it, ideally in the same transaction as the work
    return "sent"


broker = Broker(visibility_timeout=30)
broker.publish("aisha@example.com")

msg_id, body = broker.receive(now=0)
print("worker 1:", send_welcome(msg_id, body), "then crashed before acking")
msg_id, body = broker.receive(now=31)
print("worker 2:", send_welcome(msg_id, body))
broker.ack(msg_id)
print("emails sent:", len(emails_sent))

# worker 1: sent then crashed before acking
# worker 2: already sent
# emails sent: 1

Intermediate: how long will it wait?

Little’s Law: L = λW

There’s one equation every queue obeys, and it’s simple enough to do in your head. In a stable system, over a long enough time:

L = λ × W

the average number of jobs in the system = the rate they arrive × the average time each one spends there

Here λ (lambda) is the arrival rate, W is the time from arriving to finishing (waiting plus being worked on), and L counts everything inside, queued and in progress. It’s Little’s Law, after John Little, who proved it in 1961. It doesn’t care how many workers you have, what order they take jobs in, or how variable the jobs are. Three worked examples, because this is the most useful arithmetic in the article:

• Webhooks arrive at 50 per second and each takes 0.2 s from arrival to done. On average 50 × 0.2 = 10 are in the system. With 10 worker slots, you have no headroom.
• A dashboard shows 1,200 jobs queued and workers finishing 20 per second. A new job will wait about 1,200 / 20 = 60 seconds. You turned a length, which means nothing on its own, into a wait, which customers feel.
• AI replies take 4 s and you want 30 a second: 30 × 4 = 120 in progress at once. That’s 120 concurrent model calls, which your vendor’s quota had better allow.

Here’s a simulation with random arrivals and random job lengths. It measures W from the jobs and counts L independently, by looking inside the system at 20,000 random moments:

import bisect
import heapq
import random


def simulate(arrival_rate, workers, mean_service, n_tasks=200_000, seed=7):
    """Random arrivals, `workers` identical workers, one shared FIFO queue."""
    rng = random.Random(seed)
    free_at = [0.0] * workers                 # when each worker is next free (a min-heap)
    arrived, finished = [], []
    t = 0.0
    for _ in range(n_tasks):
        t += rng.expovariate(arrival_rate)    # next arrival
        start = max(t, heapq.heappop(free_at))
        finish = start + rng.expovariate(1 / mean_service)
        heapq.heappush(free_at, finish)
        arrived.append(t)
        finished.append(finish)
    return arrived, finished


arrived, finished = simulate(arrival_rate=8, workers=5, mean_service=0.5)
span = arrived[-1]
lam = len(arrived) / span
W = sum(f - a for a, f in zip(arrived, finished)) / len(arrived)

# Measure L on its own: look at the system at 20,000 random moments and count who's inside.
done = sorted(finished)
rng = random.Random(1)
samples = [rng.uniform(0, span) for _ in range(20_000)]
L = sum(bisect.bisect_right(arrived, s) - bisect.bisect_right(done, s) for s in samples) / len(samples)

print(f"arrival rate λ = {lam:.2f}/s, time in system W = {W:.3f} s")
print(f"λ × W = {lam * W:.2f}   counted L = {L:.2f}")

# arrival rate λ = 7.99/s, time in system W = 0.773 s
# λ × W = 6.18   counted L = 6.17

It holds. And W is 0.773 s although the average job is only 0.5 s of work. The extra 0.27 s is waiting in line, which is the next thing to understand.

Utilization: why 90% busy is not 10% spare

Utilization, written ρ (rho), is the fraction of time the workers are busy: the arrival rate divided by the rate the workers can finish. Five workers that each finish 2 jobs a second can do 10 a second; at 8 arriving per second they’re 80% busy.

The usual intuition is that a system 80% busy has 20% spare and is fine. It’s the most expensive misconception about queues. Random arrivals bunch up: sometimes three arrive together and two have to wait. The busier the workers, the more often an arrival finds them all occupied, and the longer the line it finds.

For the simplest case (random arrivals, one worker, random job lengths; queueing theory calls it M/M/1), the average wait in line is:

average wait = ρ / (1 − ρ) task-lengths

Here’s the formula next to a simulation:

import heapq
import random


def mean_wait(arrival_rate, workers, mean_service, n_tasks=300_000, seed=3):
    """Average time a task waits in the queue before a worker starts it."""
    rng = random.Random(seed)
    free_at = [0.0] * workers
    t = total_wait = 0.0
    for _ in range(n_tasks):
        t += rng.expovariate(arrival_rate)
        start = max(t, heapq.heappop(free_at))
        heapq.heappush(free_at, start + rng.expovariate(1 / mean_service))
        total_wait += start - t
    return total_wait / n_tasks


print("busy   formula  simulated   (wait, in units of one task's service time)")
for rho in (0.5, 0.7, 0.8, 0.9, 0.95, 0.99):
    formula = rho / (1 - rho)
    simulated = mean_wait(arrival_rate=rho, workers=1, mean_service=1.0)
    print(f"{rho:>4.0%}  {formula:>7.1f}  {simulated:>9.1f}")

# busy   formula  simulated   (wait, in units of one task's service time)
#  50%      1.0        1.0
#  70%      2.3        2.3
#  80%      4.0        4.0
#  90%      9.0        8.9
#  95%     19.0       17.8
#  99%     99.0       76.2

Read it as a price list. Going from 50% to 80% busy costs three more task-lengths of waiting; going from 90% to 95% costs ten. At 99%, even 300,000 simulated jobs weren’t enough for the average to settle: near saturation the queue swings wildly, so the wait is long and unpredictable.

This is why I plan worker pools for about 70% utilization at peak, not 90%. The last 20% of “spare” capacity isn’t spare. It’s what keeps the line short.

More workers, one line

Real systems have more than one worker, and the number of lines matters as much as the number of workers. Compare four workers, each 80% busy, arranged two ways:

# mean_wait() is the simulator from the previous snippet.
# Four workers, each busy 80% of the time on average. Same work, two layouts:
separate = mean_wait(arrival_rate=0.8, workers=1, mean_service=1.0)  # each worker has its own queue
shared = mean_wait(arrival_rate=3.2, workers=4, mean_service=1.0)    # one queue feeds all four
print(f"4 queues, 1 worker each: wait {separate:.2f} service times")
print(f"1 queue, 4 workers:      wait {shared:.2f} service times")

# 4 queues, 1 worker each: wait 4.00 service times
# 1 queue, 4 workers:      wait 0.74 service times

Same workers, same work, five times less waiting. With separate lines, one worker can sit idle while the next line has three jobs in it; with one shared line, nobody waits while any worker is free. That’s why “one queue per worker” designs (hashing jobs to a fixed worker) cost more than they look. Pool when you can; split only for a reason, like isolating workloads.

Try arrivals at 9 with 5 workers (90% busy) and watch the wait, then drop to 7. Then push arrivals to 14, past the workers’ 10 a second, and leave it unbounded: the line climbs without end, and the Little’s Law line tells you the system isn’t steady any more.

Bounded and unbounded queues

An unbounded queue accepts everything. Python’s queue.Queue() with no arguments, Celery’s Redis lists, most broker defaults: they’ll take messages until memory or disk runs out.

That’s a trap. During a long overload an unbounded queue turns a capacity problem into a latency problem and hides it. Nothing errors; every job just waits longer, and by the time someone notices, the queue holds an hour of work for callers who gave up long ago.

A bounded queue has a maximum size, and when it’s full you must decide what happens next. That decision is the design. There are three answers.

Buffer, backpressure or shed

Buffering is what a queue does below its limit: absorb the difference between arrival rate and service rate for a while. Good for bursts, useless for sustained overload.

Backpressure means the full queue pushes back on whoever is adding to it. queue.put() blocks, a TCP connection stops reading so the sender’s buffer fills. The work isn’t refused; it’s held further upstream. That’s right when the producer can slow down (a batch import, a stream processor reading a log) and wrong when it can’t: a webhook sender won’t wait for your queue to drain. It times out and retries.

Load shedding means refusing new work when you’re full, quickly and explicitly. An HTTP API answers 503 Service Unavailable with a Retry-After header. The caller finds out in milliseconds, and the work you did accept still gets done on time.

Try both in the simulator. Set arrivals to 14 and pick Block the producer: the queue stays at its limit, and a dashed line shows tasks piling up in the producer’s hands. The waiting moved upstream, where somebody can decide about it. Pick Shed new work and, once the queue is full, the refused count climbs by about 4 a second: exactly the excess.

My rule of thumb: backpressure inside a system, shedding at its edge. Between components you control, blocking loses nothing and pushes the slowness to the place that can act on it. At the boundary with callers you don’t control, shed, because you can’t make the internet wait.

Retries, with backoff and jitter

Jobs fail: the database blips, a vendor answers 503, a call times out. Most such failures are transient, so job systems retry, and how they retry decides whether a blip stays a blip. Three rules:

1. Back off. Wait longer after each failure, usually doubling: 1 s, 2 s, 4 s, 8 s. A struggling dependency gets breathing room instead of an instant second wave.
2. Cap it. A maximum wait (say five minutes) and a maximum number of attempts (say five). After that, stop and put the job somewhere a person will look.
3. Add jitter. Randomize each wait. If a thousand jobs failed at the same instant and all retry after exactly 5 seconds, they all hit the dependency again at the same instant.

import random


def backoff(attempt, base=1.0, cap=300.0, rng=random):
    """Full jitter: a random wait between 0 and an exponentially growing ceiling."""
    ceiling = min(cap, base * 2 ** attempt)
    return rng.uniform(0, ceiling)


rng = random.Random(42)
for attempt in range(6):
    ceiling = min(300.0, 1.0 * 2 ** attempt)
    print(f"retry {attempt + 1}: wait up to {ceiling:>4.0f} s, this time {backoff(attempt, rng=rng):5.1f} s")

# 1,000 workers whose call failed at the same instant, retrying once:
rng = random.Random(1)
fixed = [5.0 for _ in range(1000)]
jittered = [backoff(3, rng=rng) for _ in range(1000)]


def busiest(waits):
    """The most retries landing in any one-second slot."""
    return max(sum(1 for w in waits if s <= w < s + 1) for s in range(10))


print("busiest second, fixed 5 s delay:", busiest(fixed))
print("busiest second, full jitter:    ", busiest(jittered))

# retry 1: wait up to    1 s, this time   0.6 s
# retry 2: wait up to    2 s, this time   0.1 s
# retry 3: wait up to    4 s, this time   1.1 s
# retry 4: wait up to    8 s, this time   1.8 s
# retry 5: wait up to   16 s, this time  11.8 s
# retry 6: wait up to   32 s, this time  21.7 s
# busiest second, fixed 5 s delay: 1000
# busiest second, full jitter:     135

This is “full jitter”: pick uniformly between zero and the ceiling. AWS’s architecture blog compared several jitter schemes and found full jitter among the best, and the simplest. The fixed delay sends all 1,000 retries in one second; full jitter’s worst second gets 135.

Retry only errors that can succeed later (timeouts, 503s). Send the ones that never will (validation failures, a missing template) straight to the dead-letter queue below; retrying a bug five times is just five failures. And every retry runs the job again, so a retry policy without idempotent consumers is a duplicate-generating machine.

Visibility timeouts: how long is “probably dead”?

The visibility timeout is the broker’s guess at how long a healthy worker could take. Wrong in either direction, it causes a specific bug:

• Too short: a slow job is assumed dead and handed to a second worker while the first is still on it. SQS’s default is 30 seconds; a report that takes 45 runs twice, every time, and nothing errors.
• Too long: a job whose worker really died sits invisible that whole time. A one-hour timeout means a crashed job’s customer waits an hour.

Set it comfortably above your slowest normal job, and give jobs a hard time limit below it, so a stuck job is killed before the broker hands it to someone else. Workers can extend the timeout as they go (SQS has ChangeMessageVisibility), but I’d rather split a long job into short ones than babysit a lease.

Dead-letter queues

Some messages fail every time: a corrupt PDF, a payload the code doesn’t understand, a reference to a deleted row. Retried forever, they waste work and keep coming back to the front of the line. A dead-letter queue (DLQ) is where a message goes after too many attempts: the broker counts receives, and past a limit it moves the message aside instead of handing it out again.

from collections import deque


class Broker:
    """Ready queue, in-flight table, and a dead-letter queue for messages that keep failing."""

    def __init__(self, visibility_timeout=30, max_receives=3):
        self.ready, self.in_flight, self.dead = deque(), {}, []
        self.timeout, self.max_receives = visibility_timeout, max_receives
        self.receives = {}            # message id -> how many times it has been handed out

    def publish(self, msg_id, body):
        self.ready.append((msg_id, body))

    def receive(self, now):
        for msg_id, (msg, deadline) in list(self.in_flight.items()):
            if now >= deadline:
                del self.in_flight[msg_id]
                self.ready.append(msg)
        while self.ready:
            msg = self.ready.popleft()
            self.receives[msg[0]] = self.receives.get(msg[0], 0) + 1
            if self.receives[msg[0]] > self.max_receives:
                self.dead.append(msg)  # park it where a person can look, out of everyone's way
                continue
            self.in_flight[msg[0]] = (msg, now + self.timeout)
            return msg
        return None

    def ack(self, msg_id):
        del self.in_flight[msg_id]


def handle(body):
    if body == "invoice-7781.pdf (corrupt)":
        raise ValueError("cannot parse PDF")


broker = Broker(visibility_timeout=30, max_receives=3)
for i, body in enumerate(["invoice-7780.pdf", "invoice-7781.pdf (corrupt)", "invoice-7782.pdf"], 1):
    broker.publish(i, body)

now, done = 0, []
while broker.ready or broker.in_flight:
    msg = broker.receive(now)
    if msg is None:
        now += 31                     # nothing visible: wait for a timeout to expire
        continue
    try:
        handle(msg[1])
        broker.ack(msg[0])
        done.append(msg[1])
    except ValueError:
        pass                          # no ack: it will become visible again
print("done:", done)
print("dead-lettered:", [body for _, body in broker.dead], "after", broker.receives[2] - 1, "tries")

# done: ['invoice-7780.pdf', 'invoice-7782.pdf']
# dead-lettered: ['invoice-7781.pdf (corrupt)'] after 3 tries

A DLQ is only useful if somebody looks at it. Alarm when anything lands there, keep messages long enough to investigate (SQS allows up to 14 days), and have a way to redrive them once the bug is fixed. An unwatched DLQ is a slower way to lose messages.

Priorities and fairness between tenants

FIFO treats every job as equally urgent. Two situations break that.

The first is different kinds of work: a customer waiting for a reply matters more than tonight’s analytics rollup. Strict priority (always take the most urgent first) starves low-priority work whenever urgent work never stops. Weighted schemes avoid that, but the fix that holds up best is usually separate queues with their own workers, which Advanced covers.

The second is different customers. With one shared FIFO queue, a customer who submits 300 jobs puts them in front of everyone else’s one:

from collections import OrderedDict, deque

JOB_SECONDS = 0.1
jobs = [("bigco", 0.0)] * 300 + [("cafe", 1.0), ("clinic", 1.0)]  # (tenant, submitted at)


def fifo(jobs):
    line, t, waits = deque(sorted(jobs, key=lambda j: j[1])), 0.0, {}
    while line:
        tenant, at = line.popleft()
        t = max(t, at)
        waits.setdefault(tenant, []).append(t - at)
        t += JOB_SECONDS
    return waits


def round_robin(jobs):
    """One small queue per tenant; the worker takes one job from each tenant in turn."""
    pending, t, waits = sorted(jobs, key=lambda j: j[1]), 0.0, {}
    lanes = OrderedDict()  # tenant -> its own FIFO queue, in turn order
    while pending or lanes:
        while pending and pending[0][1] <= t:
            tenant, at = pending.pop(0)
            lanes.setdefault(tenant, deque()).append(at)
        if not lanes:
            t = pending[0][1]
            continue
        tenant, lane = next(iter(lanes.items()))
        at = lane.popleft()
        lanes.move_to_end(tenant)      # back of the turn order
        if not lane:
            del lanes[tenant]
        waits.setdefault(tenant, []).append(t - at)
        t += JOB_SECONDS
    return waits


for name, policy in (("shared FIFO", fifo), ("per-tenant turns", round_robin)):
    waits = policy(jobs)
    print(f"{name:<17} cafe waited {waits['cafe'][0]:5.1f} s, "
          f"bigco's last job {waits['bigco'][-1]:5.1f} s")

# shared FIFO       cafe waited  29.0 s, bigco's last job  29.9 s
# per-tenant turns  cafe waited   0.2 s, bigco's last job  30.1 s

Fair turns cost the big tenant almost nothing (its last job finishes 0.2 s later) and save the small tenants half a minute each. This is fair queueing, the idea routers use to stop one flow hogging a link, and it’s what I’d reach for first.

A second approach keeps one FIFO queue: give each tenant a budget of job starts per window, and when an over-budget tenant’s job reaches the front, put it back with a delay. Skipping must be nearly free (no database read, no model call), so workers chew through the noisy tenant’s backlog quickly and reach everyone else.

Run the flood under all three. With a shared FIFO queue, B, C and D wait twenty seconds and more behind A’s campaign. With fair turns they barely notice. With deferral they wait a second or two, the cost of skipping A’s backlog. Now watch A: under deferral its flood takes far longer than under fair turns, because a budget caps A even when workers have spare capacity. Fair turns are work-conserving (no worker idles while there’s work); a budget isn’t, so it must be generous enough that only abuse hits it.

Advanced: Celery, brokers and poison messages

What a Celery worker actually does

Celery is the most common job queue in Python, and its defaults are tuned for throughput on short tasks. If yours aren’t short, they’ll surprise you. The moving parts:

• A worker is one celery worker process (Celery also calls it a node). It connects to the broker and consumes one or more queues.
• Inside it is a pool of execution slots. With the default prefork pool these are child processes, and --concurrency (default: the number of CPUs) sets how many. Each slot runs one task at a time.
• The worker prefetches: it reserves messages from the broker before it has a free slot to run them, so the next task is ready the instant a slot frees up. How many it reserves is worker_prefetch_multiplier × concurrency.
• By default it acks early, when a task is handed to a slot, not when the task finishes.

Here are those defaults on Celery 5.6, plus routing, which we’ll use shortly:

from celery import Celery

defaults = Celery("defaults").conf
print("prefetch multiplier:", defaults.worker_prefetch_multiplier)
print("acks_late:", defaults.task_acks_late)
print("reject_on_worker_lost:", defaults.task_reject_on_worker_lost)
print("ack failed tasks anyway:", defaults.task_acks_on_failure_or_timeout)

app = Celery("shop", broker="redis://localhost:6379/0")
app.conf.update(
    task_routes={
        "shop.ai.*": {"queue": "ai"},              # slow: seconds per task, blocks on a vendor
        "shop.messages.*": {"queue": "messages"},  # fast: milliseconds, someone is waiting
    },
    task_default_queue="messages",                 # anything unrouted still lands somewhere sane
)
for name in ("shop.ai.summarize", "shop.messages.store", "shop.reports.weekly"):
    print(name, "->", app.amqp.router.route({}, name)["queue"].name)

# prefetch multiplier: 4
# acks_late: False
# reject_on_worker_lost: None
# ack failed tasks anyway: True
# shop.ai.summarize -> ai
# shop.messages.store -> messages
# shop.reports.weekly -> messages

Prefetch: work a busy worker hides from idle ones

Prefetching is great when every task takes milliseconds. When lengths vary, it causes something that looks impossible: tasks waiting while a worker sits idle.

I measured it: two Celery 5.6 workers with one slot each, a local Redis 7.2 broker on my laptop (an M1 Pro), and one 6-second task enqueued just before seven half-second tasks. I recorded when each task started. Three runs per setting:

With the defaults, worker A took the long task and reserved three short ones behind it. Worker B ran out of work after two seconds but couldn’t take those three: they weren’t in the queue any more. They were in A’s hands.

A multiplier of 1 helps but doesn’t fix it: with early acks, a worker still reserves one task beyond the one it’s running. With acks_late, the running message isn’t acked yet and counts against the prefetch limit, so a one-slot worker reserves nothing extra. Celery’s documentation recommends this combination for long tasks, and my numbers agree.

The cost is throughput on tiny tasks, since each worker goes back to the broker for every message. Keep the defaults for millisecond tasks; for tasks that take seconds, or a mix, set prefetch to 1, and better still, give the long ones their own queue.

acks_late, and the bill that comes with it

acks_late=True moves the ack to after the task finishes. That fixes prefetch starvation and gives you at-least-once delivery: a worker killed mid-task leaves its message unacked, so it’s redelivered. It also brings three bills.

Duplicates. A redelivered task runs again from the start, so it must be idempotent. That’s the contract you sign when you turn it on.

The visibility timeout becomes load-bearing. With Redis, an unacked message is redelivered once it’s older than visibility_timeout (default one hour). I set it to 5 seconds and ran a 120-second task with two workers. Ninety-four seconds in, the second worker started the same task while the first was still running it. (Not at 5 seconds, because kombu, Celery’s messaging library, only checks for overdue messages periodically; in the version I tested, about every 100 seconds.) Any task that can outlive the visibility timeout will eventually run twice at once, and so will a countdown longer than it, because a scheduled task sits reserved and unacked until its time.

Failures are still acked. task_acks_on_failure_or_timeout defaults to True: if your task raises, the message is acked and gone. acks_late protects you from crashes, not exceptions; those are what self.retry and autoretry_for are for.

The poison pill that kills every worker

One more setting sounds like it completes the picture: task_reject_on_worker_lost. With it on, if the process running a task dies (killed for using too much memory, say), the message goes back on the queue instead of being acked as failed.

Think about a message whose content kills the worker: a PDF that makes the parser allocate 10 GB, a payload that segfaults a C extension. I tested it with a task that exits its own process:

Four hundred and ninety-one crashes in twenty seconds, with nothing else getting done. On a fleet, the message moves from worker to worker, killing each in turn. That’s a poison message.

The defense is the one for any repeated failure: count attempts and stop. SQS, and RabbitMQ quorum queues with a delivery limit, count for you and dead-letter after N deliveries. With Redis and Celery there’s no receive count, so keep one: increment a counter keyed by the message ID before the work, and past the limit, park the message and alert. The production section shows ezychat’s version.

Concurrency: processes, threads and what blocks

The prefork pool runs each task in its own process, which is right for CPU-heavy work like rendering and parsing. It’s wasteful for work that mostly waits: an AI task that spends 3.9 of its 4 seconds waiting on a model API holds a whole process, and its memory, to do nothing. For I/O-bound work, use a thread pool (--pool threads), gevent, or async code inside the task, so one process can hold many waiting tasks.

Whatever you pick, concurrency is a promise about the dependency behind it. Twenty slots calling a vendor with a quota of ten means ten tasks refused, retried and refused again. Size concurrency from what’s downstream, not from what the machine can hold.

One queue per workload

Every growing Celery system meets this. Fast tasks (store a message, about 50 ms) and slow ones (generate an AI reply, about 5 s) share a queue. A burst of slow tasks arrives, and a notification that needs 50 ms of work is delivered a minute late because 300 AI replies got in line first.

The fix is to route each kind of work to its own queue with task_routes, as in the config above. Then be precise about which of two levels of isolation you have:

• Ordering isolation: separate queues, same workers. A fast task no longer waits behind 300 slow ones in line; Celery’s Redis transport takes from a worker’s queues in rotation. But if every slot is busy with a slow task, the fast one still waits for a slot.
• Capacity isolation: separate queues and separate workers (celery worker -Q messages here, -Q ai there). Nothing on the AI side can take a messages slot. This is the bulkhead pattern, after the walls that stop one flooded compartment sinking a ship.

Start with ordering isolation: one line of config, and it fixes the common case. Move to capacity isolation when one workload can saturate the slots, or needs a different pool type.

Redis, SQS, RabbitMQ or Kafka?

These four aren’t interchangeable. Redis, SQS and RabbitMQ are queues: a message goes to one consumer and is deleted when acked. Kafka is a log: messages are appended and kept, and each consumer group tracks its own position.

How I choose:

• You already run Redis and have a modest volume of jobs: Redis behind Celery or RQ. One less system. Its durability is Redis’s persistence, and you build your own receive counting.
• You’re on AWS and want queues you never think about: SQS. Visibility timeouts, receive counts and DLQs are built in, and you never patch a broker.
• You need routing (one message to several queues by pattern, priorities): RabbitMQ.
• The messages are a history that several consumers read at their own pace, may replay, or need in order per key at high volume: Kafka. Not for slow, independent jobs: its unit of parallelism is the partition, and a slow message holds up everything behind it there.

Ordering and partitions

Global ordering and parallelism don’t mix: if every message must be processed in order, only one worker can process them. So systems order per key: all messages for one conversation go to the same partition (Kafka) or message group (SQS FIFO) and are processed in order there, while different keys run in parallel.

import zlib

PARTITIONS = 3


def partition_for(key):
    """Same key, same partition, so one conversation's events stay in order.
    (Kafka's default partitioner uses murmur2; crc32 keeps this short.)"""
    return zlib.crc32(key.encode()) % PARTITIONS


events = [("conv-a", "hello"), ("conv-b", "hi"), ("conv-c", "price?"), ("conv-a", "are you open?"),
          ("conv-d", "BROKEN PAYLOAD"), ("conv-b", "thanks"), ("conv-e", "book a table"), ("conv-c", "ok")]
logs = {p: [] for p in range(PARTITIONS)}
for key, text in events:
    logs[partition_for(key)].append((key, text))

for p, log in logs.items():
    print(p, log)

# A consumer reads each partition in order and can't skip ahead: if one event
# keeps failing, everything behind it on that partition waits.
for p, log in logs.items():
    for i, (key, text) in enumerate(log):
        if text == "BROKEN PAYLOAD":
            stuck = sorted({k for k, _ in log[i + 1:]})
            print(f"partition {p} is stuck at {key}; also waiting: {stuck}")

# 0 [('conv-a', 'hello'), ('conv-b', 'hi'), ('conv-a', 'are you open?'), ('conv-b', 'thanks')]
# 1 [('conv-c', 'price?'), ('conv-d', 'BROKEN PAYLOAD'), ('conv-c', 'ok')]
# 2 [('conv-e', 'book a table')]
# partition 1 is stuck at conv-d; also waiting: ['conv-c']

That’s the price of ordering: head-of-line blocking. One bad message for conversation D stops C too, because they share a partition, and on a log your position is a single offset. The standard fix is to move the failing message to a retry topic (later a dead-letter topic), commit past it, and carry on: you give up ordering for one key instead of progress for the whole partition.

Ask whether you need ordering at all. Often what you need is that a stale event can’t overwrite a newer one, and a version check in the consumer gives you that without ordering the queue.

The leaky bucket is a queue

A leaky bucket rate limiter is a bounded queue drained at a fixed rate: requests pour in, leak out one per interval, and spill over when it’s full. It’s backpressure in miniature, and the right tool in front of a dependency with a hard rate limit.

Expert: overload, and the feedback loops that keep it going

Retry amplification

Retries are a good idea at every layer, which is what makes them dangerous. A mobile app retries a failed request 3 times; the API it calls retries downstream 3 times; that service retries its query 3 times. When the database slows down, one tap becomes up to 3 × 3 × 3 = 27 queries, aimed at the component that was already struggling.

The fixes:

• Retry at one layer, usually the one that owns the job. Inner layers fail fast and report.
• Use a retry budget: retries may add at most, say, 10% to the request rate; past that, fail instead. Google’s SRE book describes per-client retry budgets of this kind.
• Honor Retry-After and back off with jitter, so retries arrive after the trouble, not during it.

Metastable failures: when the system stays down

This is the most important simulation in the article. One server handles 100 requests a second; normal traffic is 80. Clients give up after 1 second. From second 10 to 15 traffic spikes to 160 a second, then returns to 80. The server works its queue in order and doesn’t know which clients have given up.

from collections import deque

TICK = 0.001        # simulate in 1 ms steps
SERVICE = 10        # ticks per request: one server doing 100 requests/second
TIMEOUT = 1000      # ticks: the client gives up after 1 second


def new_per_second(second):
    return 160 if 10 <= second < 15 else 80   # a 5-second spike, then back to normal


def run(retries=0, lifo=False, drop_expired=False, seconds=60):
    queue = deque()
    give_up = {}                   # tick -> requests whose client gives up then
    answered = [0] * seconds       # answers that reached a client still waiting, per second
    busy_until, owed = 0, 0.0
    for now in range(seconds * 1000):
        owed += new_per_second(now // 1000) * TICK
        while owed >= 1:           # fresh requests arrive evenly spaced
            owed -= 1
            req = {"deadline": now + TIMEOUT, "tries": 0, "served": False}
            queue.append(req)
            give_up.setdefault(req["deadline"], []).append(req)
        for req in give_up.pop(now, []):
            if not req["served"] and req["tries"] < retries:
                retry = {"deadline": now + TIMEOUT, "tries": req["tries"] + 1, "served": False}
                queue.append(retry)        # the abandoned original is still in the queue too
                give_up.setdefault(retry["deadline"], []).append(retry)
        while now >= busy_until and queue:
            req = queue.pop() if lifo else queue.popleft()
            if drop_expired and req["deadline"] <= now:
                continue                   # its client has gone: skip it at no cost
            req["served"] = True
            busy_until = now + SERVICE
            if busy_until <= req["deadline"] and busy_until < seconds * 1000:
                answered[busy_until // 1000] += 1
    return answered


def report(name, answered):
    print(f"{name:<30}" + "".join(f"{sum(answered[s:s + 10]) / 10:>7.0f}" for s in range(0, 60, 10)))


print(f"{'useful answers per second':<30}" + "".join(f"{f'{s}-{s + 10}s':>7}" for s in range(0, 60, 10)))
report("FIFO, no retries", run())
report("FIFO, 3 retries", run(retries=3))
report("FIFO, 3 retries, drop expired", run(retries=3, drop_expired=True))
report("LIFO, 3 retries", run(retries=3, lifo=True))

# useful answers per second       0-10s 10-20s 20-30s 30-40s 40-50s 50-60s
# FIFO, no retries                   80     26     40     80     80     80
# FIFO, 3 retries                    80     26      0      0      0      0
# FIFO, 3 retries, drop expired      80     52     86     80     80     80
# LIFO, 3 retries                    80    100     82     80     80     80

Read the second row slowly. The spike ended at 15 seconds and traffic went back to what the server handled comfortably before. For the next 45 seconds it answered nothing in time. It was 100% busy the whole time, working for clients who had already left, and its queue ended the run at 11,319 requests.

The loop: the spike builds a queue more than a second deep, so every client times out and retries, adding a copy while the original is still queued. Arrivals jump from 80 to as many as 320 a second against a capacity of 100. The queue never drains, so every new request times out too. The trigger is gone; the overload sustains itself.

This is a metastable failure: two stable states, healthy and overloaded, where a temporary push moves the system from one to the other and it doesn’t come back by itself. The term comes from a 2021 HotOS paper by Bronson and colleagues, followed by an OSDI 2022 study of real cloud incidents by Huang and colleagues, in which retries are among the most common sustaining loops. You get out by cutting load hard (block at the edge, scale up a lot, turn retries off) until the queue drains, then letting traffic back slowly. Without retries (row one), the same spike costs about 25 seconds and the system recovers on its own.

Drop work nobody is waiting for

Rows three and four keep the retries and break the loop.

Drop expired work. Before starting a request, check whether its caller is still waiting. That needs the request to carry its deadline, a good habit anyway (gRPC propagates deadlines across calls; elsewhere, use a header or a message field). The server stops spending capacity on the dead, the queue drains, and it recovers within seconds of the spike.

Serve the newest first (LIFO). Under overload, the request most likely to still have a waiting client is the one that just arrived. Here LIFO reached 100 useful answers a second during the spike. The cost is fairness: stale requests pile up at the bottom (480 at the end of this run) and wait until the queue empties, if ever. So LIFO is an overload mode, not a default. Facebook’s “Fail at Scale” paper (Ben Maurer, ACM Queue, 2015) describes this pairing: switch to LIFO under load, with a CoDel-style queue timeout that drops old requests.

CoDel: limit the wait, not the length

A length bound has a dilemma. Set it low and you drop bursts that would have cleared in a second; set it high and, under sustained overload, everyone waits the maximum. What you care about is time.

CoDel (controlled delay), from Kathleen Nichols and Van Jacobson in 2012, was designed for router queues. It measures how long packets sojourn in the queue and drops only when the minimum sojourn stays above a target (5 ms in the paper) for a whole interval (100 ms). A burst that clears quickly is left alone; a queue that never empties is a standing queue, and that’s what gets dropped.

Facebook’s adaptation for server queues is easy to copy: if the queue has emptied within the last interval, it’s a burst, so allow a long wait; if not, it’s a standing queue, so new arrivals get a short timeout. Here it is against two length bounds:

from collections import deque


def serve(arrivals, policy, rate=100, seconds=30):
    """One server finishing `rate` requests/second. Returns (shed, waits of the requests it served)."""
    queue, waits, shed = deque(), [], 0
    arrivals = deque(sorted(arrivals))
    last_empty, t, step = 0.0, 0.0, 1 / rate
    while t < seconds:
        while arrivals and arrivals[0] <= t:
            at = arrivals.popleft()
            limit = policy(len(queue), t - last_empty)
            if limit is None:
                shed += 1                       # refused at the door
            else:
                queue.append((at, at + limit))  # (arrived, give up waiting at)
        while queue:
            at, expires = queue.popleft()
            if t > expires:
                shed += 1                       # waited too long: drop it unserved
                continue
            waits.append(t - at)
            break
        if not queue:
            last_empty = t
        t += step
    return shed, waits


def length_cap(n):
    """Refuse new work once n requests are waiting. No limit on how long they wait."""
    return lambda queue_len, busy_for: None if queue_len >= n else float("inf")


def sojourn_limit(queue_len, busy_for, interval=2.0, normal=2.0, standing=0.1):
    """Adaptive queue timeout (CoDel's idea, as Facebook describe it for server queues):
    if the queue has emptied within the last `interval`, it's a burst, so allow a long wait;
    if it hasn't, it's a standing queue, so new arrivals may wait only briefly."""
    return normal if busy_for < interval else standing


def pct(waits, q):
    return sorted(waits)[int(len(waits) * q)]


burst = [0.0] * 100 + [i / 40 for i in range(40 * 30)]   # 100 at once, then a calm 40/s
overload = [i / 120 for i in range(120 * 30)]            # a steady 120/s against capacity 100/s
policies = (("cap at 10", length_cap(10)), ("cap at 50", length_cap(50)), ("sojourn limit", sojourn_limit))
for name, arrivals in (("burst, then calm", burst), ("20% overload", overload)):
    for label, policy in policies:
        shed, waits = serve(arrivals, policy)
        print(f"{name:<17} {label:<14} shed {shed:>4}   median wait {pct(waits, 0.5):.2f} s   "
              f"p99 {pct(waits, 0.99):.2f} s")

# burst, then calm  cap at 10      shed   91   median wait 0.01 s   p99 0.01 s
# burst, then calm  cap at 50      shed   51   median wait 0.01 s   p99 0.42 s
# burst, then calm  sojourn limit  shed    0   median wait 0.01 s   p99 0.93 s
# 20% overload      cap at 10      shed  590   median wait 0.10 s   p99 0.10 s
# 20% overload      cap at 50      shed  550   median wait 0.50 s   p99 0.50 s
# 20% overload      sojourn limit  shed  588   median wait 0.10 s   p99 0.36 s

The small cap handles overload well and throws away 91 burst requests that would have been served within a second. The large cap keeps more of the burst and makes everyone wait half a second under overload. The sojourn limit keeps the whole burst and holds overload waits as low as the small cap. Every policy sheds about 600 in the overload case, because 20% too much work for 30 seconds is 600 requests too many. The policy only decides which ones, and how long the survivors wait.

Admission control: limit what’s in flight

All of this assumes you know the server’s capacity. Usually you don’t, and it changes: a deploy slows a query, a cache goes cold. Admission control caps the work in progress and adjusts the cap from measurements, using TCP’s AIMD rule: while latency is under target, allow one more request in flight; when it’s over, cut the limit by a fraction. Netflix’s open-source concurrency-limits library does this for services. Requests over the limit wait briefly in a small bounded queue, then get a 503.

Backpressure end to end in async Python

In async Python, backpressure is in the standard library if you ask for it. asyncio.Queue() with no maxsize is unbounded; give it a bound, and await queue.put() pauses the producer when it’s full:

import asyncio
import time


async def producer(queue, n):
    """Reads n events as fast as it can and hands them on. `put` waits when the queue is full."""
    started = time.perf_counter()
    for i in range(n):
        await queue.put(i)                 # backpressure: this line is where the producer slows down
    return time.perf_counter() - started


async def consumer(queue, seen):
    while True:
        item = await queue.get()
        await asyncio.sleep(0.01)          # 10 ms of work, e.g. one database write
        seen.append(item)
        queue.task_done()


async def main(maxsize):
    queue, seen = asyncio.Queue(maxsize=maxsize), []
    workers = [asyncio.create_task(consumer(queue, seen)) for _ in range(4)]
    peak = 0

    async def watch():
        nonlocal peak
        while True:
            peak = max(peak, queue.qsize())
            await asyncio.sleep(0.001)

    watcher = asyncio.create_task(watch())
    produced_in = await producer(queue, 400)
    await queue.join()
    for task in workers + [watcher]:
        task.cancel()
    print(f"maxsize={maxsize:<4} producer finished in {produced_in:.2f} s, "
          f"queue peaked at {peak}, processed {len(seen)}")


asyncio.run(main(maxsize=0))     # 0 means unbounded
asyncio.run(main(maxsize=20))

# maxsize=0    producer finished in 0.00 s, queue peaked at 396, processed 400
# maxsize=20   producer finished in 0.96 s, queue peaked at 20, processed 400

Same work, same total time. The unbounded version held 396 items at once; the bounded one never more than 20, because the producer was slowed to the consumers’ pace. If the producer read from a socket, that slowdown would reach the sender through TCP flow control. That’s end-to-end backpressure: each stage slows the one before it, back to the source.

At the edge, where the source is a client you don’t control, you tell it to come back instead. The same queue with put_nowait becomes an intake that sheds, and a semaphore bounds whatever expensive thing sits behind it:

import asyncio
import math


class Intake:
    """The edge of a job system: accept while there's room, refuse with a useful answer when there isn't."""

    def __init__(self, maxsize, jobs_per_second):
        self.queue = asyncio.Queue(maxsize=maxsize)
        self.rate = jobs_per_second          # measured throughput of the workers behind the queue

    def submit(self, job):
        try:
            self.queue.put_nowait(job)
            return 202, {}                   # accepted: it will run
        except asyncio.QueueFull:
            drain = self.queue.qsize() / self.rate
            return 503, {"Retry-After": str(math.ceil(drain))}


model_slots = asyncio.Semaphore(4)           # at most 4 calls to the model vendor at once
in_flight = peak = 0


async def call_model(prompt):
    global in_flight, peak
    async with model_slots:                  # the fifth caller waits here, not at the vendor
        in_flight += 1
        peak = max(peak, in_flight)
        await asyncio.sleep(0.05)            # stand-in for a slow API call
        in_flight -= 1
        return f"summary of {prompt}"


async def main():
    intake = Intake(maxsize=3, jobs_per_second=2)
    print([intake.submit(f"job {i}") for i in range(5)])
    results = await asyncio.gather(*(call_model(f"doc {i}") for i in range(20)))
    print(len(results), "model calls, at most", peak, "at once")


asyncio.run(main())

# [(202, {}), (202, {}), (202, {}), (503, {'Retry-After': '2'}), (503, {'Retry-After': '2'})]
# 20 model calls, at most 4 at once

Two details. The intake answers 202 Accepted, not 200: the job is queued, not done. And a full queue answers 503, not 429. A 429 says “you are over your limit”; a 503 says “I am overloaded,” which is what a full queue means. Retry-After comes from how long the backlog takes to drain: Little’s Law again.

Deadlines travel with the work

The simulations all point one way: the most valuable fact about a queued job is often when it stops being worth doing. A customer reply after an hour; a “typing” event after five seconds; a scheduled tick after the next tick has fired. Put an expiry on jobs that have one. Celery takes expires on a task (an expired task is discarded when a worker picks it up), or put a deadline in the payload and check it first. One line turns “process a backlog of stale work” into “skip it in microseconds.”

Interview walk-through: design a job system

“Design a background job system” (or its cousins: a notification service, a webhook delivery system, a task scheduler) tests whether you can reason about load and failure, not whether you can name Kafka. Here’s a structure that works.

1. Requirements. Ask first.

• Functional: submit a job (returns an ID at once), run it, check its status. Several job types with different costs. Failures retry; repeated failures are kept for inspection. Some jobs are scheduled.
• Non-functional: 5 million jobs a day, peak 4× average, about 2 s of worker time each, mostly waiting on an external API. No accepted job is lost. At peak a job starts within 30 s. One tenant must not delay the others.

2. Estimates. Out loud, with Little’s Law:

import math

jobs_per_day = 5_000_000
peak_factor = 4            # busiest minute vs the daily average
service_s = 2.0            # worker time per job: mostly waiting on a model API
target_busy = 0.7          # leave headroom: waits explode near 100%
slots_per_process = 20     # I/O-bound, so one process can hold 20 jobs at once
wait_budget_s = 30         # a job may wait this long at peak before we refuse new ones
job_bytes = 2_000

avg = jobs_per_day / 86_400
peak = avg * peak_factor
busy_slots = peak * service_s                 # Little's Law: L = λ × W
slots = math.ceil(busy_slots / target_busy)
print(f"average {avg:.0f} jobs/s, peak {peak:.0f} jobs/s")
print(f"slots busy at peak {busy_slots:.0f}, provision {slots} -> {math.ceil(slots / slots_per_process)} processes")
print(f"queue bound at peak: {peak * wait_budget_s:,.0f} jobs")
print(f"payload per day: {jobs_per_day * job_bytes / 1e9:.0f} GB")

# average 58 jobs/s, peak 231 jobs/s
# slots busy at peak 463, provision 662 -> 34 processes
# queue bound at peak: 6,944 jobs
# payload per day: 10 GB

Say what they mean. 231 jobs a second is modest for any broker. The real constraint is 463 jobs in progress at once: 463 concurrent calls to the external API, so the vendor’s concurrency limit, not the broker, shapes the design.

3. API. POST /jobs returns 202 with a job ID, or 503 with Retry-After when the intake is full; GET /jobs/{id} returns status. Accept an idempotency key on submit, so a retried submit doesn’t create two jobs.

4. High-level design. The submit handler writes the job row (queued) and then enqueues its ID, with a sweeper that re-enqueues queued rows older than a few minutes, because the enqueue can fail after the commit. Separate queues and worker pools per job class (interactive, bulk, scheduled). Workers ack late, run idempotently by job ID, and update the row.

5. Deep dives. Have an answer ready for each:

• Broker: SQS on AWS (receive counts and DLQs built in), or Redis and Celery if already there. Not Kafka: these are independent jobs, not an ordered history.
• Delivery: at least once, idempotent workers, visibility timeout above the hard time limit.
• Retries: capped exponential backoff with full jitter; DLQ after five receives, with an alarm and a redrive tool.
• Fairness: fair turns across tenants, or a per-tenant budget on starts that defers instead of dropping.
• Overload: bounded intake shedding with 503; autoscale on the age of the oldest job; deadlines where jobs have them.
• Poison messages: receive counts and a DLQ; never reject_on_worker_lost without a count.

6. Observability. Per queue: oldest message age, depth, throughput, DLQ arrivals, retry rate. Alarm on age; it’s the wait itself.

In production: the queues in ezychat.io

ezychat.io is an AI customer-conversation product: businesses connect WhatsApp, Instagram, Messenger and web chat, and AI and people answer their customers. Almost everything slow runs on Celery workers with Redis as the broker. Here’s how the queues are laid out and why, from the code as it is today.

The webhook door answers fast and remembers everything

Meta’s webhooks arrive at a separate webhook-ingress service: the API’s image, started in a mode that registers only the webhook routes, behind its own load balancer target group. The handler caps the body size, checks Meta’s signature over the raw bytes, saves the payload to a webhook_logs table, enqueues the processing task, and answers 200. The docstring:

No business logic lives here: the POST handler’s only job is to verify the request came from Meta, durably log the raw payload, and hand off to the ingestion pipeline

The order matters. The row is written before the enqueue, so if the enqueue fails or the task is lost, the unprocessed row is still there for a sweep to find. The obligation lives in the database; the queue is only a fast path to it.

Lanes, chosen at the door

A post can get hundreds of comments a minute, and comments share the processing task with messages. So the ingress picks the queue from the payload:

def _queue_for(payload: dict) -> str:
    for entry in payload.get("entry", []) or []:
        for change in entry.get("changes", []) or []:
            if change.get("field") in COMMENT_CHANGE_FIELDS:
                return ENGAGEMENT_QUEUE
    return MESSAGES_QUEUE

The reason is recorded where the queue is declared, with a measurement:

Bounded by reach rather than by MAC, so one campaign can put thousands of ingest tasks in front of a paying customer’s message — measured at ~900 backed-up tasks, a probe behind them waited 1.45s against 0.06s on a quiet queue. Celery’s Redis transport rotates between queues rather than draining one before the next, so this alone restores fairness: no second worker, and the existing one consumes it.

That’s ordering isolation, in production: a separate lane, the same worker. Anything unrecognized goes to the messages lane, a choice the docstring calls “wrong for volume, right for correctness”.

Six more queues each carry a comment saying what they keep apart: ai for reply turns and other blocking model calls; ingest for bursty knowledge syncs, so “a large sync never starves live reply generation”; happiness for deal-room scoring that “arrives as a fleet-wide burst every fifteen seconds”; and publishing, billing and docwork, each kept off messages.

To be precise about what that buys: today one worker service consumes all eight queues. That’s ordering isolation, not capacity isolation, and the infrastructure comment says when that changes: “When AI volume justifies it, ai/ingest split onto their own task definition.” I think that’s the right order: measure the contention, then pay for separate workers.

The “Common mistake” from Advanced is recorded here too: a test asserts that every routed queue appears in the worker’s --queues flag in both the Compose file and the ECS task definition, because, in its docstring’s words, “a routed-but-unconsumed queue strands its tasks in Redis forever while the producer sees a successful enqueue”.

The sweeper rides a different lane

AI replies run on the ai queue. The sweep that finds AI turns that were never answered runs every 60 seconds, and it is routed, deliberately, to messages:

# The lost-turn sweep stays off the `ai` queue: it is what notices that
# queue falling behind, and its age metric must keep publishing then.
"ezychat.ai.tasks.sweep_unanswered_turns": {"queue": MESSAGES_QUEUE},
"ezychat.ai.tasks.*": {"queue": AI_QUEUE},

The specific route comes before the wildcard so it wins. On the AI queue, the sweep would sit in the backlog it exists to detect, and the reply-age alarm would go quiet exactly when replies were late.

Coalescing a burst of messages into one turn

People send “hi”, then “is the shop open”, then “on Sunday”, seconds apart. Three messages shouldn’t mean three replies. Each message stamps a per-conversation marker in Redis with its own ID and schedules the AI turn 4 seconds out (ai_debounce_seconds: int = 4). A starting turn checks the marker:

marker = await redis.get(_debounce_key(str(conversation_id)))
...
if marker is not None and marker.decode() != token:
    return {"status": "superseded"}

If a newer message has arrived, this turn exits, and the newest turn answers them all, since a turn reads every inbound message since the last reply. Instead of deduplicating work after it’s done, make stale copies cheap no-ops. If Redis is unavailable, the docstring says it “fails open to processing (a possible duplicate turn)”, never a dropped one.

acks_late where it’s safe, and a poison-pill brake

The webhook processing task is the one place I turned on acks_late, and the comment above it records both halves of the trade from the Advanced section:

# Acknowledged only after the task finishes, so a worker stopped mid-task
# (a deploy, a container kill) leaves the message to be redelivered rather
# than gone for good. Safe because ingestion is idempotent: every write
# dedupes on Meta's own ids, the property Meta's own redeliveries rely on.
#
# Deliberately *not* `reject_on_worker_lost`: a delivery that kills its
# worker process (out of memory, say) would be requeued at once and kill
# the next one, forever. The replay sweep below re-enqueues such a log
# within minutes instead, and abandons it after three attempts.
@celery_app.task(name=PROCESS_WEBHOOK_LOG_TASK_NAME, acks_late=True)
def process_webhook_log(log_id: str) -> None:

That “kill the next one, forever” is the 491 crashes in 20 seconds from my experiment. Instead, the replay sweep (every two minutes) counts replays per log in Redis and stops at three:

if _replay_attempt(log_id) > _REPLAY_LIMIT:
    abandoned.append(log_id)
    continue

An abandoned log is marked “abandoned after 3 replays without finishing” and logged at CRITICAL, because “something in it is killing the worker, and replaying it would keep killing the worker.” A dead-letter queue built from a column and a counter.

Late is not lost

The same sweep has a rule I’d put in every replay mechanism. An unprocessed webhook might be lost, or just waiting behind a backlog, and replaying the second kind is the retry amplification from Expert:

# While its queue has a backlog, a log is most likely queued, not lost:
# replaying it would add work exactly when the worker is behind. Past this
# it is replayed regardless, which bounds how long a lost one can wait.
_REPLAY_PATIENCE = timedelta(minutes=15)

So the sweep checks the lane’s depth first. While it has a backlog, a log younger than 15 minutes is left alone. And since every write dedupes on Meta’s IDs, a replay that turns out to be a duplicate is harmless.

Fairness on the comment lane

Separating comments protects messages. It doesn’t protect one business from another’s campaign, because engagement is still one FIFO queue. The fairness module’s docstring states the design:

The queue stays FIFO; the over-quota tenant’s work becomes cheap to skip.

Each organization may start a generous number of batches a minute; past that, the triage task checks the budget before any model call or database read and re-enqueues itself with a 30-second countdown. Nothing is dropped. It’s “Defer past share” from the demo; the rate limiter article covers the numbers. The budget fails open, because “there is no contention to arbitrate when the thing arbitrating it is down.”

The same task retries a transient failure with capped backoff:

raise self.retry(exc=exc, countdown=min(60 * 2**self.request.retries, 900)) from exc

One minute, two, four, eight, then the fifteen-minute ceiling, with max_retries=5. It has no jitter; each retry belongs to one post, but if a vendor outage failed thousands of posts in one second, jitter is what I’d add.

A tick that expires

Deal-room scoring is driven by a tick every 15 seconds. A tick that waited longer than that in a backed-up queue would duplicate the next one, so it expires at its cadence:

"tick-happiness-scoring": {
    "task": "ezychat.happiness.tasks.tick",
    "schedule": 15.0,
    # A tick that waited out its own cadence in a backed-up queue would
    # fan out the same rooms the next tick already did.
    "options": {"expires": 15.0},
},

The smallest queue: one per WebSocket

Inbox updates reach each browser over a WebSocket, and each connection has its own asyncio.Queue(maxsize=256) with a single writer. The send used to be awaited inline, so a browser that stopped reading stopped the relay loop, and Redis decided what happened next: the handbook records a storm in which 16,926 of 19,992 frames were delivered while the database showed nothing undispatched.

The overflow policy is neither block nor shed. A full queue is emptied and replaced by one {"type": "resync"} frame, telling the client to refetch:

So the queue is bounded, and going over it means the connection stops trying to describe what changed and tells the client to go and look instead.

A 5-second cooldown stops it asking twice. Before it, a probe holding a socket deaf drew “35 resync frames from a single overload”: 35 refetches from one tab. With it, one.

Admission control on the API, and age as the alarm

The API has a bounded queue too: admission control, a concurrency limit derived from a 200 ms latency target, and a short wait for a slot (api_admission_queue_timeout_seconds: float = 0.5) before a 503 with Retry-After. Webhooks are exempt, because “Meta disables a webhook subscription that keeps failing.”

And the alarms watch the age of the oldest unfinished thing, not depth. WebhookUnprocessedAgeSeconds alarms when a delivery has waited over five minutes for two consecutive five-minute periods; AiTurnPendingAgeSeconds past ten minutes, with a description written for whoever is paged: “A customer message has waited over 10 minutes for its AI turn.” A depth of 500 might be fine or a disaster; ten minutes of waiting is a customer.

What I’d copy

• Write the obligation to the database before you enqueue, and sweep for rows the queue lost. The queue is the fast path, not the record.
• Choose the lane at the door, from what the work is, and give every new queue a consumer before anything routes to it.
• Put the watchdog on a different lane from the work it watches.
• Turn on acks_late only for idempotent tasks, never reject_on_worker_lost without a receive count, and make abandoning a message loud.
• Don’t replay what’s merely late. Check whether the queue is busy before adding to it.
• Make over-quota work cheap to skip instead of making the queue smarter.
• Alarm on age. It’s the number your customers feel.

Cheat sheet

• Queue: producers add messages, workers take them in order. Decouples the two and absorbs bursts by trading time for capacity.
• Ack after the work: at-least-once delivery. Duplicates are normal; make consumers idempotent.
• Visibility timeout: above the job’s hard time limit, or slow jobs run twice at once.
• Little’s Law: L = λ × W. Queue length ÷ throughput = wait.
• Utilization: wait ≈ ρ / (1 − ρ) task-lengths for one worker. 80% busy waits 4, 95% waits 19. Plan for about 70%.
• One shared line beats many: pool workers behind one queue unless you need isolation.
• Bound every queue. When full: backpressure (block the producer) inside the system, shed (503 + Retry-After) at the edge.
• Retries: exponential backoff, full jitter, a cap, a maximum number of attempts, then a dead-letter queue someone watches.
• Fairness: fair turns per tenant, or a per-tenant budget that defers cheaply. Strict priority starves.
• Celery: prefetch 4 and early acks by default. For long tasks, worker_prefetch_multiplier=1 and acks_late=True. Route workloads to separate queues; separate workers for capacity isolation.
• Poison messages: count receives and stop. reject_on_worker_lost without a count is a crash loop.
• Brokers: Redis (simple, already there), SQS (managed, DLQs built in), RabbitMQ (routing), Kafka (ordered, replayable log; parallelism per partition).
• Ordering: per key, not global; it costs head-of-line blocking.
• Retries multiply across layers. Retry at one layer, with a budget.
• Metastable failure: retries keep a system overloaded after the trigger is gone. Drop work whose caller has gone, or serve newest first under overload.
• CoDel: limit how long work waits, not how many items wait.
• Async Python: asyncio.Queue(maxsize=N) for backpressure, put_nowait to shed, Semaphore around every finite dependency.
• Alarm on the age of the oldest message, per queue.