Batching and Async Queues

Also known as: batch inference, job queue, deferred processing

Move work nobody is waiting on off the request path and into batches that are cheaper and easier to control.

8 min read

The problem

Bulk work gets forced through the synchronous request path because that is how the first version was written. Classifying fifty thousand records, enriching a whole catalog, or re-embedding a corpus then hits rate limits, ties up web workers, times out at the load balancer, and pays full interactive pricing for work whose results nobody will look at for hours. A single retry storm can also starve the traffic that genuinely is interactive.

The pattern

Split synchronous from asynchronous work explicitly. Anything with a user waiting stays on the request path. Everything else goes to a queue: batched into provider batch APIs where available for a substantial discount, processed with bounded concurrency and proper backoff, checkpointed so a failure resumes rather than restarts, and reported through a job status the caller can poll.

Use when

Bulk operations over many items where results are not needed immediately
Your provider offers a batch API at a meaningful discount
Rate limits are the binding constraint on throughput
Long-running work currently blocks web workers or times out

Avoid when

A user is waiting — never queue interactive work
Volume is low enough that a simple loop is sufficient
Results are needed within seconds, which batch APIs do not guarantee

Tradeoffs

GainBatch pricing is typically around half of interactive, on large volumes
CostLatency measured in minutes to hours; unusable for anything interactive
GainThe interactive path is protected from bulk work and rate-limit contention
CostQueue infrastructure, job status, and retry semantics to build and operate
GainCheckpointing makes a failed bulk job resumable instead of restartable
CostPartial completion is a state the caller must understand and handle

The dividing line is simple and worth enforcing: is a human waiting for this right now? If not, it does not belong on the request path — and the moment it moves off, cheaper options open up.

Use the provider's batch API where the deadline allows

Batch endpoints trade latency for price and for much higher effective rate limits. Submit the whole workload, poll for completion, and reconcile results by custom id. The important detail is that partial failure is normal: some items in a batch will error, and the job must handle a mixed result rather than assuming all-or-nothing.

async def run_batch(items: list[Item], job_id: str) -> BatchResult:
    requests = [
        {"custom_id": item.id, "params": {"model": MODEL, "messages": prompt_for(item)}}
        for item in items
    ]

    batch = await client.messages.batches.create(requests=requests)
    await jobs.update(job_id, status="submitted", provider_batch_id=batch.id, total=len(items))

    while (status := await client.messages.batches.retrieve(batch.id)).processing_status != "ended":
        await jobs.update(job_id, completed=status.request_counts.succeeded)
        await asyncio.sleep(30)

    succeeded, failed = {}, []
    async for result in client.messages.batches.results(batch.id):
        if result.result.type == "succeeded":
            succeeded[result.custom_id] = result.result.message
        else:
            failed.append((result.custom_id, result.result.error))   # normal, not exceptional

    return BatchResult(succeeded=succeeded, failed=failed)

When you need a live queue instead

Batch APIs have deadlines measured in hours, which does not suit work that should finish in minutes. For that, run a worker pool against the normal API with bounded concurrency, adaptive backoff on rate limits, and per-item checkpointing so a crash resumes where it stopped.

async def process_queue(job_id: str, items: list[Item], concurrency: int = 10) -> None:
    done = await checkpoints.completed_ids(job_id)      # resume, do not restart
    todo = [i for i in items if i.id not in done]
    sem = asyncio.Semaphore(concurrency)

    async def one(item: Item) -> None:
        async with sem:
            for attempt in range(5):
                try:
                    result = await call_model(prompt_for(item))
                    await checkpoints.record(job_id, item.id, result)
                    return
                except RateLimited as e:
                    delay = e.retry_after or (2 ** attempt) + random.random()
                    await asyncio.sleep(delay)          # jitter avoids synchronized retries
                except PermanentError as e:
                    await checkpoints.record_failure(job_id, item.id, str(e))
                    return
            await checkpoints.record_failure(job_id, item.id, "rate limit retries exhausted")

    await asyncio.gather(*(one(i) for i in todo))
    await jobs.update(job_id, status="complete")

warning

Always add jitter to backoff. Without it, every worker that hit a rate limit retries at the same instant, and the synchronized burst re-triggers the limit — a self-inflicted thundering herd that turns a brief throttle into a long one.

Give the caller a real job status

Asynchronous work needs a status surface, or users will re-submit jobs already running. Return a job id immediately, expose progress including failures, and make partial results retrievable before the job finishes — for most bulk work, 90% of the results now beats 100% of them in an hour.

The test is whether a human is waiting; if not, it leaves the request path
Use the provider batch API when hours of latency are acceptable
Bound concurrency, back off with jitter, and checkpoint per item
Treat partial failure as normal and report it, item by item
Expose job status and partial results so callers do not resubmit

tip

Keep separate rate-limit budgets for interactive and bulk traffic, or a large job will consume the shared quota and degrade the experience of every live user while it runs.

Related patterns

Parallel Tool Fan-Out Decomposed Extraction Circuit Breaker and Fallback Chain