Making a slow queue keep up

I clicked “Run,”went to make coffee, came back, and the spinner was still spinning. Made lunch. Still spinning. The job eventually finished — twenty to sixty minutes later — and here's the maddening part: nothing was broken. No errors. No crashes. CPU relaxed. Every dashboard a calm, confident green. The queue was just quietly making everyone wait. This is the story of finding the five places it was strangling itself, and fixing them without renting a bigger machine.

The setup

The service runs AI vision checks over construction drawings — roughly, “does the rebar in this blueprint match the spec?” A job splits into blocks, each block is a few steps, and most steps call a hosted vision model that happily chews on a single image for 10–25 minutes of real wall-clock time. Work is dispatched onto AWS SQS and chewed through by Celery workers (via kombu) on ECS Fargate.

The bug report was one sentence: “I click go and nothing happens for ages.” Measured queue waits were 21–64 minutes. And yet CPU was low, nothing was erroring, and the vision API wasn't rate-limiting us. By every metric I was looking at, the system was fine.

Everything was fine. That was the problem.

The trap was that there was no single “the bottleneck.” Throughput was pinched in five places at once, and heroically fixing any one of them alone changed the wait time by approximately zero:

• Worker concurrency was 1. Each worker process did exactly one thing at a time, like a very focused, very slow intern.
• A vision-API semaphore of 1. An internal limit added to “avoid rate limits” also meant one — and only ever one — model call could be in flight.
• Autoscaling on CPU that never fired. The scale-out trigger was CPU 70%. But the work is I/O-bound — it's mostly waiting on an API — so CPU peaked around 57% and the fleet never grew, no matter how deep the backlog got. The autoscaler's one job was to notice we were drowning; it was watching the wrong gauge and having a lovely, relaxed day.
• One shared queue for two very different jobs. A quick interactive step a human is actively waiting on stood in the same line as everyone's 25-minute heavy passes.
• Effectively short polling. The queue's wait-time was 0s and the library was on its own low default, so workers busy-asked “anything for me? anything for me?” and still picked things up slowly.

Almost everything below is the same idea wearing different hats: let independent work actually run in parallel, and scale on the thing that's actually backed up.

One queue for two jobs is the original sin

The worst symptom was head-of-line blocking. The very first step of a job — the quick one that decides how to carve up the drawing, the one a human is staring at a spinner for — was queued behind other people's 25-minute heavy passes. A five-second task was waiting half an hour because it stood in the same line. It's the express checkout being closed so you're stuck behind someone's monthly grocery haul, holding a single banana.

The fix isn't more workers. It's a second lane. The interactive step gets its own dedicated queue; the heavy work stays on the main one. Same reason a design tool never makes “move this shape” wait behind “export this 4K video.”

The part I'm quietly proud of is how it ships. The producer picks the queue at dispatch time, and if the fast lane isn't configured yet, it gracefully falls back to the main queue. So the code can go to production before the infrastructure even exists — it changes nothing until you set one environment variable, and rolling back is just unsetting it. No heroics, no big-bang cutover.

import os

HEAVY_QUEUE = "jobs"                        # 10–25 min vision passes
FAST_QUEUE  = os.environ.get("FAST_QUEUE")  # unset → fast lane not wired up yet

def queue_for_interactive_step() -> str:
    # Graceful fallback: no fast lane configured → use the main queue.
    # Deploying this is a no-op until FAST_QUEUE is set; rollback = unset it.
    return FAST_QUEUE or HEAVY_QUEUE

def dispatch_blocksplit(job_id: str):
    send_task("worker.blocksplit", args=[job_id],
              queue=queue_for_interactive_step())

One sharp edge, delivered at 3am courtesy of kombu (Celery's SQS transport): it will crash-loop with an UndefinedQueueExceptionif you route to a queue that isn't in the worker's known set. So the fast queue is defensively aliased to the main queue — if the two configs ever disagree, the worker quietly degrades to the main lane instead of face-planting on boot.

Scale on the backlog, not the CPU

The autoscaler was measuring the wrong thing. For a worker that spends its life waiting on an API, CPU is a terrible proxy for “are we behind?” The honest signal is the queue itself: how many messages are sitting there, visible and unloved.

So the scale-out target became backlog per running task — roughly messages_waiting / running_tasks— with CPU demoted from “decision-maker” to “guardrail.”

# Grow the fleet when the QUEUE is backing up, not when CPU is high.
target = 5                                   # aim for ~5 waiting messages per task
backlog_per_task = messages_visible / max(running_tasks, 1)

if backlog_per_task > target:
    desired = ceil(messages_visible / target)
    set_desired_count(min(desired, MAX_TASKS))  # cap so a spike can't melt the bill

The protection job: don't reap a worker mid-render

Here's the catch that scaling on the backlog creates. When the backlog drains, the autoscaler wants to scale in — kill some tasks to save money. Fine, except a 25-minute vision pass can't be resumed. If a scale-in event reaps the exact task holding a job that's 24 minutes deep, that block starts over from zero — and re-bills the model call. Autoscaling and long non-resumable work are natural enemies.

The fix is a small “protection job”: while a worker is actually processing a heavy pass, it tells ECS “do not reap me.” ECS calls this UpdateTaskProtection. You switch it on when you pick up real work and off the moment you're done, so the autoscaler is free to reclaim idle tasks but never a busy one.

import boto3

ecs = boto3.client("ecs")

def set_protection(task_arn: str, on: bool, minutes: int = 60):
    # expiresInMinutes is the safety valve: even if we crash without
    # clearing it, protection auto-lapses so a task can't become immortal.
    ecs.update_task_protection(
        cluster=CLUSTER,
        tasks=[task_arn],
        protectionEnabled=on,
        expiresInMinutes=minutes if on else None,
    )

def run_block(msg):
    set_protection(SELF_TASK_ARN, True)      # entering a long pass — shield me
    try:
        do_vision_pass(msg)                  # 10–25 min, not resumable
    finally:
        set_protection(SELF_TASK_ARN, False) # done — safe to reclaim again

A concurrency budget, not a vibe

With the lanes split, backlog scaling on, and busy tasks protected, the last knob is how many model calls to allow in flight. My first instinct was, of course, to crank it to the moon.

The rate limiter had other plans — and here's the thing people miss: a 429 doesn't slow a job, it failsa block. It's not a speed bump, it's a trapdoor. So I stopped guessing and treated in-flight calls as an arithmetic budget:

in_flight = tasks * concurrency * max_inflight    # must stay <= vision-API quota

The immediate win was raising max_inflight from 1 to 2 — which unlocked a fan-out that already existed in the code but was welded shut by that semaphore of 1. And instead of hoping we stayed under quota, the worker honors it explicitly and backs off politely on the rare 429 rather than stampeding into it:

import asyncio, random

gate = asyncio.Semaphore(MAX_INFLIGHT)     # was 1 (!); now a computed budget

async def call_model(payload):
    async with gate:                       # never exceed the budget
        for attempt in range(5):
            try:
                return await vision.generate(payload)
            except RateLimited as e:
                # honor Retry-After when given; otherwise exponential + jitter
                delay = e.retry_after or (2 ** attempt + random.random())
                await asyncio.sleep(delay)
        raise                              # give up loudly, don't fail silently

From there the plan is to ramp, not jump: 2 → 4, guided by measurements, never in one leap. And you can only ramp safely if you can seeit — so every model call now logs its own latency, token count, and cost. That per-call log is the ramp's steering wheel; without it, “add more concurrency” is just gambling with a production credit card.

Long-poll instead of busy-waiting

Short polling is a worker asking “anything for me?”, being told “no” instantly, and immediately asking again — thousands of empty round-trips, and, ironically, slower pickup. Long polling lets one request wait up to 20 seconds (the AWS maximum) for a message to show up. Fewer empty receives, and a new message gets grabbed almost the instant it lands.

# 20s is the AWS maximum. Set it on the QUEUE and the CONSUMER —
# the library has its own low default that silently wins otherwise.
BROKER_TRANSPORT_OPTIONS = {
    "wait_time_seconds": 20,      # long poll (kombu default is a sad 10)
    "visibility_timeout": 3600,   # >= the longest job, or it gets redelivered mid-run
}

Jobs people abandoned lived forever

One more quiet money leak: someone starts a job, sees the spinner, thinks “nope” and closes the tab. The job kept right on going — burning real vision-model spend — because the reconciler only ever looked at jobs marked running, and the front-end's cancel signal was never wired up. Nothing swept these away. They just accumulated, like gym memberships.

The fix is a whole-job heartbeat: the most recent activity anywhere in the job — the job row, any of its blocks, any of its steps. If that heartbeat has been quiet for over an hour, a background reconciler cancels the job.

select greatest(
  j.updated_at,
  coalesce(max(b.updated_at), j.updated_at),
  coalesce(max(s.updated_at), j.updated_at)
) as last_seen
from jobs j
left join blocks b on b.job_id = j.id
left join steps  s on s.job_id  = j.id
where j.id = :job_id
group by j.updated_at;

The subtle part is a race: a model callback can land at the exact moment the reconciler decides a job is dead. So the reconciler re-locks the row and re-checks the heartbeat inside the transaction — a late callback that just bumped updated_at wins, and the job gets a reprieve.

from datetime import timedelta

def sweep_abandoned():
    for job_id in candidate_ids():             # cheap pre-filter outside the lock
        with tx():
            job   = select_for_update(job_id)  # re-lock the row
            fresh = job_heartbeat(job_id)       # re-read INSIDE the tx
            idle  = now() - fresh
            if idle > timedelta(minutes=60) and job.status in ("pending", "running"):
                transition_to_canceled(job)     # SAME path as the user's Cancel button

Notice the last line: the reconciler and the user-facing “Cancel” button call the same transition function. There is exactly one place in the codebase where a job becomes canceled, whether a person or a watchdog pulls the trigger — so the state machine can't quietly grow two subtly different flavors of “canceled.”

The gotchas that ate an afternoon

SQS-through-Celery has a special genre of bug: the thing you configured isn't the thing that runs. You set a value, you deploy, you feel good, and the system serenely ignores you.

The one that cost the most coffee: visibility timeout in the wrong place is completely inert.

# WRONG — kombu silently ignores visibility_timeout nested per-queue here:
BROKER_TRANSPORT_OPTIONS = {
    "predefined_queues": {
        "jobs": {"url": JOBS_URL, "visibility_timeout": 3600},  # ← ignored. gone. void.
    },
}

# RIGHT — it's a transport-wide option (and set it on the AWS queue too):
BROKER_TRANSPORT_OPTIONS = {
    "visibility_timeout": 3600,        # this one is actually honored
    "wait_time_seconds": 20,
    "predefined_queues": {"jobs": {"url": JOBS_URL}},
}

And the rest of the afternoon's greatest hits:

• The library's default wait-time wins. kombu defaults to a 10-second receive wait regardless of your queue's setting; you have to set it on the consumer too (above).
• Route to an unknown queue and it crash-loops. The UndefinedQueueException from earlier — hence the defensive alias.
• The env var lived in the task definition, not Terraform. I “changed” a value in code that was actually pinned by a JSON task-def. You can stare at the right file for a long time convinced it's set.
• IAM is per-queue-ARN, not a prefix. A brand-new queue needs a brand-new grant — the old policy silently didn't cover it, and SQS “access denied” looks a lot like “queue empty.”
• Wait-time doubles as a requeue delay. The transport reuses your receive wait-time as the visibility delay on requeue — one more reason not to leave it at a surprising default.

{
  "Effect": "Allow",
  "Action": [
    "sqs:SendMessage", "sqs:ReceiveMessage", "sqs:DeleteMessage",
    "sqs:GetQueueAttributes", "sqs:ChangeMessageVisibility"
  ],
  "Resource": [
    "arn:aws:sqs:REGION:ACCOUNT:jobs",
    "arn:aws:sqs:REGION:ACCOUNT:jobs_fast"   // the NEW lane — easy to forget
  ]
}

Most “it's slow” bugs aren't one bottleneck — they're several small serializations stacked up. Measure each layer, because fixing four out of five feels exactly like fixing none.

How I actually worked it

The issue showed up with a thread full of confident claims about what was wrong. My first move was to trust nothing and re-check every claim against the real code— and three of them were wrong (the IAM was per-ARN, not prefixed; two “prerequisite” changes were already merged; the consumer wasn't short-polling the way the thread insisted). Chasing the stated problem would have burned a day fixing things that were already fine.

Then the boring discipline that keeps a multi-repo change from becoming a multi-repo incident:

• Deploy in dependency order — infrastructure first (the new queue + scaling + protection), then the consumer that reads it, then the producer that writes to it. Each stage verified before the next, so nothing ever points at something that doesn't exist yet.
• Graceful fallback on everything, so each PR is independently deployable and independently reversible.
• Read the primary source. When I wasn't sure how the transport behaved, I read its actual source rather than trusting my own pull-request description — and ended up correcting two of my own claims in review. Humbling, but cheaper than shipping them.

Reference reading that earned its keep: the AWS SQS guide on short vs long polling (the 20-second max, and why wait-time is not task duration), and the transport library's own SQS source for the defaults nobody writes down.

What I deliberately didn't do

• No FIFO queue. We don't need strict ordering, and FIFO's throughput ceiling would've just swapped one bottleneck for a fancier one.
• No prefetch > 1. Pulling several long messages onto one worker just ages them against the visibility timeout while it works the first — a wonderful way to reprocess a 25-minute job you already finished.
• No blind visibility-timeout bump to 12 hours. Tempting to set it huge and forget it, but that turns one stuck job into a 12-hour ghost. The heartbeat reconciler is the honest version of that.
• No bigger instances. The work was I/O-bound and serialized; a faster CPU would have sat there waiting on the API just as long, only more expensively.

In short

The queue wasn't slow because it was small. It was slow because a quick job shared a line with hour-long ones, the fleet scaled on a gauge that never moved, and exactly one model call could happen at a time. The fix was mostly about letting parallelism that already existed actually happen: a second lane, a backlog-based scaler, a protection job so scaling in doesn't torch live work, a concurrency budget you can compute, and a watchdog that reclaims what people abandon. No cleverness, no bigger box — just matching each tool to the failure in front of it, and keeping the boring parts boring.