Concurrency on a worker fleet

A queue solves the problems of the previous chapter and none of the problems of the first five. The endpoint answers in milliseconds, no request is ever long, deploys stop destroying work, and imports survive a file of any size. What the migration does not touch is the rate at which the provider is called, because that was never a property of where the code ran.

The tell is the division from chapter 5: measured throughput times measured service time gives the requests genuinely outstanding. Ten worker processes draining a queue at 12 items per second with an 800 ms service time is 9.6 requests in flight — one per worker, near enough — while the account sits at 720 requests per minute against a deployment rated for twenty thousand. Low CPU and no rejections complete the picture: nothing is being throttled and nothing is working hard, because only ten calls exist at a time.

Two divisions carry that paragraph. Twelve items a second, each holding a slot for 800 ms, is twelve multiplied by eight tenths of a second, which is 9.6 requests genuinely outstanding — against ten worker processes, one apiece. And 12 items a second is 720 a minute, which against a deployment rated at twenty thousand a minute is a few per cent of what the account will serve. Neither number needs a profiler, and together they say something a profiler would not: the constraint is not inside any single worker, because each worker is doing one thing at a time and doing it at the provider's own speed. Scaling the fleet is the obvious response and the expensive one, since ten processes issuing one request each is the same defect as one process issuing one request at a time, bought ten times over. The whole failure is uneventful — nothing errors, nothing is refused, and the only symptom is that a migration which fixed three real problems left the fourth exactly where it was.

Concurrency lives inside the job#

The code that produces this contains no visible sequential loop over the provider, which is why it survives review:

async def summarise_import(ctx: Context, import_id: str) -> None:
    rows = await store.rows_for(import_id)
    for row in rows:                                   # 812 rows, one at a time
        summary = await summarise(ctx["client"], row)
        await store.write_summary(row.id, summary)

The await inside the loop yields to the event loop, so this worker is perfectly well-behaved asyncio, and it still issues one request at a time. Ten such workers offer ten concurrent requests, and the queue drains at a rate set by how many processes are running.

Worker frameworks give you a per-process job concurrency setting — max_jobs in arq, --concurrency in RQ or Celery — controlling how many jobs a single worker process runs at once. Raising it from the default is the first thing anyone tries, and on this workload it changes nothing, because there is one import in the queue and one job cannot be run twice. The knob controls jobs in flight; the workload's parallelism lives inside a job. Which knob matters is a consequence of job granularity, and that is a design decision made when the queue was introduced.

One job per row is the shape most queue documentation encourages, and it makes concurrency a fleet-level setting: ten workers at max_jobs=20 give 200 requests in flight, tunable without changing application code. The cost is that "this import is finished" stops being something the queue can tell you, so you maintain a counter — a Redis DECR on a key seeded with the row count, or a COUNT(*) WHERE summary IS NULL against the import — and something has to notice imports that stall at 811 of 812 because one row's job was lost.

Chunking is the compromise that survives contact with large files. Seventeen messages instead of 812 keeps queue overhead and completion tracking cheap, a retry costs at most fifty rows of duplicated work, and each chunk job runs its own bounded fan-out internally, using the same semaphore-and-client structure a standalone batch process would:

async def summarise_chunk(ctx: Context, import_id: str, row_ids: list[str]) -> None:
    rows = await store.rows_by_id(row_ids)
    gate: asyncio.Semaphore = ctx["gate"]              # shared across jobs in this process

    async def one(row: Row) -> None:
        async with gate:
            summary = await summarise(ctx["client"], row)
        await store.write_summary(row.id, summary)     # idempotent on row.id

    async with asyncio.TaskGroup() as tg:
        for row in rows:
            tg.create_task(one(row))

The gate living in ctx rather than inside the function is the detail that makes this correct. A semaphore constructed per job bounds each job independently, so twenty concurrent chunk jobs at a bound of ten present two hundred requests, and the number in the constructor stops describing anything. A process-wide gate, created once when the worker starts, bounds the process. Writing each summary as it completes rather than collecting them keeps the streamed per-row progress the page depends on, and makes a chunk retry cheap in the sense that matters: rows already written are skipped by an idempotent upsert rather than re-summarised.

The chunk size itself is chosen against two clocks rather than against queue overhead. An 812-row import at 50 rows a chunk is 17 messages, which keeps queue traffic and completion tracking cheap. A 50-row chunk running at ten calls in flight is five batches of ten at 800 ms apiece, about four seconds of work — well inside a 30-second shutdown grace, so a deploy landing mid-chunk lets it finish and the message is never redelivered. A 500-row chunk is ten times that, around forty seconds, which is outside the grace period and gets killed part-done every time a deploy overlaps it. Retry cost moves in the same direction: a lost chunk redoes at most 50 rows, and rows already written are skipped by the idempotent upsert, so what a retry actually costs is bounded by the chunk size rather than by the import size.

A bound spread across a fleet that resizes itself#

Every gate in this topic has been a Python object in one process, and the consequence arrives properly here: ten workers each holding a process-wide bound of ten present a hundred concurrent requests, and an autoscaler adding ten more during a busy afternoon presents two hundred, without any configuration having changed. The per-process number is a fleet number divided by a replica count that is not constant.

A shared counter is the usual first build, and the thing to get right is atomicity, since a read-modify-write across the network races under exactly the load that makes the limit matter. A Lua script executed by Redis, or an INCR with a per-window key and an EXPIRE, gives a rate limit every replica agrees on:

ALLOW = """
local current = redis.call('INCR', KEYS[1])
if current == 1 then redis.call('EXPIRE', KEYS[1], ARGV[2]) end
return current <= tonumber(ARGV[1])
"""


async def acquire_request_slot(redis: Redis, window_seconds: int = 1) -> bool:
    key = f"provider:rpm:{int(time.time()) // window_seconds}"
    return bool(await redis.eval(ALLOW, 1, key, RATE_PER_WINDOW, window_seconds * 2))

A one-second window rather than a one-minute window is worth the extra keys, because a fixed minute window lets the whole allowance be spent in the first two seconds and then rejects everything for fifty-eight — the same burst that a semaphore sized from a published rate produces, reproduced by your own limiter. Sub-second windows smooth it further at the cost of more Redis traffic; one second is a reasonable place to stop.

Putting the window at one second rather than one minute is the burst argument from the first chapter of this topic arriving from the other direction. A fixed one-minute counter permits the entire allowance to be spent in whatever fraction of the minute the fleet needs, and then refuses everything until the counter resets — so a limiter installed to smooth traffic reproduces exactly the burst it was meant to prevent. A one-second window caps any burst at a sixtieth of the allowance, which is small enough that a provider metering over short windows never sees a spike. The cost is sixty times as many keys, each expiring on its own, and that is cheap enough that the reason to stop at one second is not expense but diminishing returns.

Queue depth is the obvious signal to autoscale on, and it is the one every platform makes easiest to wire up: backlog above a threshold, add workers. Against a provider allowance that does not grow, that control loop has a failure mode built into it. A backlog appears because the account is already saturated, the autoscaler adds workers, the additional workers raise offered concurrency past the fleet bound, the extra requests are refused, throughput falls, the backlog grows, and the autoscaler adds more. The system converges on maximum replicas, maximum spend and minimum completed work. The fix is a ceiling on the autoscaler rather than a different signal: maximum replicas is the fleet-wide in-flight bound divided by the per-process bound, and above that number a worker cannot obtain capacity to use. Queue depth remains the right metric below that ceiling. Depth pinned high with the replica count at its ceiling means the account is the constraint, which is a quota conversation rather than a scaling one, and depth at zero with workers idle means the producer is the constraint.

The reason that loop actively makes things worse, rather than merely failing to help, is the property established when the bound was being tuned: a refused request still consumes the account's request allowance. Every worker added past the fleet bound issues requests that cannot be served, those requests spend budget that would otherwise have carried real work, and completed items per minute therefore falls rather than levelling off. Falling throughput lengthens the queue, a queue-depth rule reads a longer queue as demand for more workers, and the loop tightens on itself. A ceiling cuts it at the one point where the control law and the constraint meet, which is why the number to cap at is the fleet-wide in-flight bound divided by the per-process bound and not a guess about how much traffic there is.

Bulk and interactive traffic on one allowance#

All of this becomes urgent when work moves onto shared workers, because the imports now draw from the same account as the chat endpoint, continuously, at whatever rate the fleet can manage. A batch is a perfectly efficient consumer of any quota left unattended, and every request it makes is one the interactive path may be refused.

The cleanest resolution is to stop sharing. Most providers will issue a second API key, a second project, or a separate provisioned deployment, and partitioning quota — say 80% to interactive, 20% to bulk — makes the two workloads incapable of harming each other. The cost is that the partition is static, so the bulk share sits idle overnight when interactive traffic is near zero, which is exactly when a batch would most like to run fast.

Where quota cannot be partitioned, the bulk side yields dynamically by reading the provider's own accounting. The rate-limit headers on every response report what remains in the current window, and a worker that pauses when remaining capacity falls below a reserve keeps the reserve available for user traffic:

async def yield_to_interactive(response: httpx.Response) -> None:
    remaining = int(response.headers.get("x-ratelimit-remaining-requests", "999999"))
    if remaining < INTERACTIVE_RESERVE:                # e.g. 30% of the window's allowance
        await asyncio.sleep(float(response.headers.get("x-ratelimit-reset-requests", "1")))

This is cooperative rather than enforced, so it works while you control every bulk client and fails the moment a new job appears that nobody told about the convention — which is the argument for the metering proxy, where the rule lives in one place and applies to callers who have never heard of it.

Where bulk work is genuinely going is off the synchronous API altogether. Most major providers now offer an asynchronous batch endpoint that accepts a file of requests, returns results within a stated window measured in hours, charges roughly half the per-token price, and consumes a separate quota pool from real-time traffic. For an overnight import of 40,000 catalogue entries with no user waiting on it, that is a better answer than any concurrency setting in this topic: half the cost, no rate-limit arithmetic, and no contention with the chat endpoint. It is the wrong answer for a merchandiser watching rows appear on a screen, which is why both paths tend to exist in the same codebase.

A queue, then, relocates the waiting and hands you a new place to put the concurrency bound. Where that bound used to be one semaphore in one process, it is now a fleet number that has to be divided by a replica count nobody guarantees, enforced by a shared counter or a proxy if the replica count moves, and capped at the autoscaler so a backlog cannot summon workers the account cannot feed.

The shape that survives contact with production has four numbers in it, and each comes from something measured rather than from a default. The fleet-wide in-flight bound comes from the tighter of the account's published limits at the observed latency, then adjusted upward by a sweep until completed work per minute stops improving. The per-process bound is that number divided by the replica count, or delegated to a shared limiter if the replica count moves. The transport's max_connections is set at or above the per-process bound, so the pool never becomes an accidental second gate. And the job granularity is chosen so a retry costs one chunk rather than one import, and so a chunk finishes inside the deploy grace period. The check that says whether all four are right is the same division throughout: measured throughput multiplied by measured service time should equal the concurrency configured. When it does, the provider is the constraint and the remaining levers are quota, prompt size and the batch endpoint. When it does not, the gap is inside your own process, and the six causes tabulated in chapter 5 are where it hides.