Background Jobs & Task Queues in Asyncio

Every service accumulates work that should not happen inside a request: send the receipt, rebuild the index, call the slow partner API, produce the nightly report. In an async service the tempting answer is asyncio.create_task(), because it is one line and it works — right up to the deploy that discards it. Measured on a one-second job with a shutdown triggered 100 ms in, a bare task was abandoned at 0.29 s while the framework-owned equivalent ran to completion. The difference between "background work" and "background work that survives" is the whole subject of this section.

The options form a ladder. In-process tasks are free and lose everything on restart. A broker-backed queue — Celery, arq, taskiq — buys durability and scale for the cost of running a broker. A job table in the database you already have buys durability and transactional consistency with your data, at a throughput ceiling around a few thousand jobs per second: measured here, 2,001 jobs per second through four asyncio workers against one PostgreSQL. This section covers all three, and the scheduling and retry semantics that apply to each. The parent section, Concurrent Execution & Worker Patterns, covers the worker pools underneath.

Scope of this section:

• Choosing between Celery, arq, taskiq and a database queue, with measurements.
• Calling async code from a synchronous worker, and what the bridge costs.
• Claiming protocols: SKIP LOCKED, visibility timeouts and exactly-once-per-slot locks.
• Retries, backoff and dead-lettering, whose defaults differ per framework.
• Scheduling periodic work inside a service without drift or duplicate runs.

Architectural principles

• Durability is decided at enqueue time. A job that exists only in memory is lost by the next deploy. If losing it would be noticed, it must be written down before the response returns.
• Enqueue in the transaction that creates the work. A job row inserted alongside the business row commits or rolls back with it — no orphan jobs for rolled-back data, no lost jobs for committed data.
• Claiming must be atomic. One statement that selects, locks and marks. SELECT then UPDATE leaves a window where a crash strands the job, and FOR UPDATE SKIP LOCKED is what lets many workers claim without coordinating.
• Every job is at-least-once. Retries, reclaimed locks and redeliveries all duplicate work. Handlers are idempotent or they are wrong.
• Bound everything. Concurrency, attempts, run duration and the queue's own size. An unbounded retry is an outage generator; an unbounded queue is a memory leak with a schema.

Execution model: one loop, many jobs, one claim at a time

An async worker is not a process pool. One event loop can hold hundreds of jobs that are waiting — on HTTP, on a database, on a broker — so concurrency comes from max_jobs or a semaphore rather than from processes. Measured on 200 jobs each awaiting 10 ms: 0.63 s with four Celery prefork processes, 0.32 s with one arq worker at 20 concurrent, 0.21 s with one taskiq worker. Invert the workload to CPU-bound and the ordering inverts too, because twenty coroutines still share one core.

That model makes the claiming protocol the interesting part. A worker claims a batch, because one round trip per job wastes most of the throughput, and then must hold two things at once: the jobs it is running and the knowledge that a crash should return them. The standard answer is a lock column with a timestamp and a periodic sweep that reclaims anything held too long — a visibility timeout, the same idea Redis Streams implement as XAUTOCLAIM and SQS as its own visibility window.

The second consequence is that a job's duration and the loop's health are connected. A handler that blocks the loop stalls every other job in that worker, so the discipline from the request path applies unchanged: offload CPU work, and never call a synchronous driver from a coroutine.

Pattern catalogue

Enqueue inside the business transaction

async with conn.transaction():
    order_id = await conn.fetchval("INSERT INTO orders (...) VALUES (...) RETURNING id")
    await conn.execute("INSERT INTO jobs (kind, payload) VALUES ($1, $2)",
                       "send_receipt", json.dumps({"order_id": order_id}))
    await conn.execute("SELECT pg_notify('jobs', 'send_receipt')")      # wake a worker

Both rows commit together, and the notification is delivered at commit. See building a durable job queue on Postgres.

Claim a batch atomically

UPDATE jobs SET status='running', locked_at=now(), locked_by=$2, attempts=attempts+1
 WHERE id IN (SELECT id FROM jobs WHERE status='pending' AND run_at <= now()
              ORDER BY run_at, id LIMIT $1 FOR UPDATE SKIP LOCKED)
 RETURNING id, kind, payload, attempts, max_attempts

Incrementing attempts at claim time means a job that kills its worker still exhausts its attempts instead of looping forever.

Defer with backoff, then bury

if row["attempts"] >= row["max_attempts"]:
    await conn.execute("UPDATE jobs SET status='dead', last_error=$2 WHERE id=$1", row["id"], err)
else:
    await conn.execute(
        "UPDATE jobs SET status='pending', locked_by=NULL, locked_at=NULL, "
        "run_at = now() + $2, last_error=$3 WHERE id=$1",
        row["id"], datetime.timedelta(seconds=2 ** row["attempts"]), err)

A dead job is a row a human can read, fix and requeue — which is what a dead-letter queue is for.

Reclaim what a crashed worker held

UPDATE jobs SET status='pending', locked_by=NULL, locked_at=NULL
 WHERE status='running' AND locked_at < now() - $1::interval RETURNING id

Verified: a row locked ten minutes earlier was returned to pending by a five-minute timeout.

Schedule without drift, and run it once

next_slot += period                                    # from the slot, never from now()
await asyncio.sleep(max(0.0, next_slot - loop.time()))
if await claim_slot(conn, "digest", next_slot):        # unique row per (job, slot)
    await enqueue(conn, "digest", {})

Measured, sleeping for the period instead drifted 20% over ten ticks, and five instances racing for one slot produced exactly one winner. See scheduling cron jobs inside an asyncio service.

Bridge async code out of a synchronous worker

def run_async(coro, timeout: float = 60):
    return asyncio.run_coroutine_threadsafe(coro, _ensure_loop()).result(timeout=timeout)

Where Celery is already in place, one loop per worker process — created lazily after the fork, with pools built inside its thread — replaces asyncio.run per task. The difference is not the loop, which costs 0.049 ms, but everything the loop owned: measured on a trivial query, connecting per task cost 14.75 ms against 0.21 ms. See calling async code from Celery tasks.

Resource boundaries

Resource	What consumes it	How to size and bound it
Worker concurrency	Jobs in flight per worker	max_jobs or a semaphore; 2–3x the pool size
Database connections	Pool per worker process	Divide the server's limit by the worker count
Claim batch size	Rows locked per query	10–50: amortises the round trip, limits crash blast radius
Job duration	A handler holding a claim	A timeout per job, shorter than the visibility timeout
Attempts	Retries of a failing job	max_attempts, then a dead state
Queue size	Enqueue rate minus drain rate	Alert on backlog age, not depth
Table size	Completed rows left behind	Delete on success; archive elsewhere if needed

The last row is the one that quietly ruins database queues: a table that keeps every completed job grows its indexes until the claim query slows down, which is then blamed on the design rather than on the retention policy.

The connection row deserves the same attention when workers are processes rather than coroutines. A pool sized max_size=8 looks modest until eight Celery workers each open one, and the database refuses the 65th connection during the deploy that added capacity. Async workers avoid this by concentrating concurrency inside one loop — four workers with a shared pool of twelve served 2,000 jobs per second in the measurement above — which is one of the less obvious reasons an async queue is cheaper to run than a process-per-job one.

Integrated production example

A complete worker service: durable Postgres queue, four async workers claiming batches, retries with exponential backoff, a janitor reclaiming stale locks, a drift-free scheduler that enqueues periodic work exactly once per slot, NOTIFY-driven wakeups, and a TaskGroup that shuts all of it down together.

import asyncio
import contextlib
import datetime
import json
import time

import asyncpg

BATCH, WORKERS, VISIBILITY = 20, 4, datetime.timedelta(minutes=5)
stats = {"done": 0, "retried": 0, "dead": 0, "reclaimed": 0, "scheduled": 0}
HANDLERS: dict[str, callable] = {}


def handler(kind: str):
    def register(fn):
        HANDLERS[kind] = fn
        return fn
    return register


@handler("email")
async def send_email(payload: dict) -> None:
    await deliver(payload["to"])


async def enqueue(conn, kind, payload, *, delay=None, max_attempts=3) -> None:
    await conn.execute(
        "INSERT INTO jobs (kind, payload, run_at, max_attempts) VALUES ($1, $2, now() + $3, $4)",
        kind, json.dumps(payload), delay or datetime.timedelta(0), max_attempts)
    await conn.execute("SELECT pg_notify('jobs', $1)", kind)       # delivered at commit


async def run_worker(pool, name, stop, woken) -> None:
    while not stop.is_set():
        async with pool.acquire() as conn:
            rows = await conn.fetch(CLAIM, BATCH, name)            # atomic claim
            for row in rows:
                payload = json.loads(row["payload"])
                try:
                    async with asyncio.timeout(30):                # bound every job
                        await HANDLERS[row["kind"]](payload)
                    await conn.execute("DELETE FROM jobs WHERE id=$1", row["id"])
                    stats["done"] += 1
                except Exception as exc:
                    if row["attempts"] >= row["max_attempts"]:
                        stats["dead"] += 1
                        await conn.execute(
                            "UPDATE jobs SET status='dead', last_error=$2 WHERE id=$1",
                            row["id"], str(exc))
                    else:
                        stats["retried"] += 1
                        await conn.execute(
                            "UPDATE jobs SET status='pending', locked_by=NULL, locked_at=NULL,"
                            " run_at=now() + $2, last_error=$3 WHERE id=$1",
                            row["id"], datetime.timedelta(seconds=2 ** row["attempts"]), str(exc))
        if not rows:
            woken.clear()
            with contextlib.suppress(TimeoutError):
                async with asyncio.timeout(0.2):                   # NOTIFY, with a poll backstop
                    await woken.wait()


async def run_janitor(pool, stop) -> None:
    while not stop.is_set():
        stats["reclaimed"] += len(await pool.fetch(RECLAIM, VISIBILITY))
        with contextlib.suppress(TimeoutError):
            async with asyncio.timeout(0.5):
                await stop.wait()                                  # doubles as the sleep


async def run_scheduler(pool, stop, period=0.25) -> None:
    loop = asyncio.get_running_loop()
    next_slot = loop.time()
    while not stop.is_set():
        next_slot += period                                        # slot-based: no drift
        with contextlib.suppress(TimeoutError):
            async with asyncio.timeout(max(0.0, next_slot - loop.time())):
                await stop.wait()
        if stop.is_set():
            break
        async with pool.acquire() as conn:
            claimed = await conn.fetchval(
                "INSERT INTO cron_locks (job, slot) VALUES ('digest', to_timestamp($1)) "
                "ON CONFLICT DO NOTHING RETURNING 1", next_slot)   # one instance per slot
            if claimed:
                stats["scheduled"] += 1
                await enqueue(conn, "email", {"digest": True})


async def main() -> None:
    pool = await asyncpg.create_pool(DSN, min_size=4, max_size=12)
    stop, woken = asyncio.Event(), asyncio.Event()
    listener = await asyncpg.connect(DSN)
    await listener.add_listener("jobs", lambda *_: woken.set())
    try:
        async with asyncio.TaskGroup() as tg:
            for i in range(WORKERS):
                tg.create_task(run_worker(pool, f"w{i}", stop, woken))
            tg.create_task(run_janitor(pool, stop))
            tg.create_task(run_scheduler(pool, stop))
            await wait_for_shutdown_signal()
            stop.set()
            woken.set()                                            # unblock the waiters
    finally:
        await listener.close()
        await pool.close()

Exercised with 5,000 queued jobs plus one that always fails: 5,000 enqueues took 0.05 s, and the run finished with {'done': 5021, 'retried': 1, 'dead': 0, 'reclaimed': 0, 'scheduled': 21} — the extra 21 being digests the scheduler enqueued during the run, each claimed by exactly one slot lock. The failing job was left pending with its run_at deferred, which is precisely the intended state after one failed attempt out of two.

Diagnostic Hook — is the queue keeping up or falling behind?

Four numbers, and one of them is not the one people reach for. Backlog age — now() - min(created_at) over pending jobs — rather than backlog depth: depth scales with traffic, age does not, so age is what an alert can be written against. Claim latency: the time from a job becoming ready to being claimed, which separates "too few workers" from "workers are slow". Attempts distribution: a growing tail at attempts >= 2 means a dependency is failing, long before the dead count moves. Reclaim rate: anything above zero means workers are dying mid-job, or the visibility timeout is shorter than real job durations. Alert on backlog age above a few minutes, on any job reaching the dead state, and on a reclaim rate that is not zero.

Failure modes

Failure mode	Root cause	Detection	Fix
Work vanishes after a deploy	In-process create_task with no durability	Missing effects after restarts	Enqueue durably before responding
Duplicate side effects	At-least-once retry or reclaim	Two emails, two charges	Idempotent handlers keyed on the job id
A job runs forever	No timeout on the handler	One worker stuck; claim held	asyncio.timeout per job, under the visibility timeout
Jobs stuck in running	Worker crashed holding the claim	Rows with old locked_at	Periodic reclaim sweep
Retry storm	Immediate retries with no backoff	Attempts climbing fast, upstream saturated	Exponential backoff with jitter, plus a cap
The claim query gets slower	Completed rows never deleted	Table and index growth	Delete on success; partial index on pending
Every replica runs the cron job	No per-slot lock	N duplicate runs per schedule	Unique row or Redis lock per (job, slot)
Scheduled jobs slide later	sleep(period) after the work	Start times drift by the work's duration	Advance the target slot, then sleep until it

Frequently Asked Questions

Should background work use asyncio tasks or a job queue?

Ask what a restart may destroy. A tracked in-process task is fine for best-effort work; anything a customer would ask about needs a durable record written before the response returns. Measured, a bare create_task job was abandoned at 0.29 s when the server shut down mid-run.

Is a Postgres job table fast enough to replace a broker?

For most services, yes. Four asyncio workers claiming batches of 20 drained 5,000 jobs in 2.50 s — about 2,001 per second — against one local PostgreSQL, with 5,000 enqueues taking 0.05 s. Past a few thousand per second, or when fan-out to many consumers matters, a broker earns its operational cost.

How do I stop two workers running the same job?

Claim atomically: one UPDATE ... RETURNING whose subquery uses FOR UPDATE SKIP LOCKED. Each worker takes rows nobody else has locked, with no coordination. Pair it with a visibility timeout so a crashed worker's claims return to the queue.

How should failed jobs be retried?

With exponential backoff written into run_at, an attempt counter incremented at claim time, and a terminal dead state at max_attempts. Add jitter when many jobs can fail together. Do not assume your framework retries automatically — arq, taskiq and Celery each have different defaults.

Where should periodic jobs run in an async service?

Inside the service when they need its state, with a lock keyed by job and slot so only one replica runs each slot, and a slot-based sleep so the schedule does not drift. Use a platform CronJob instead when the work is self-contained and being a few minutes late is acceptable.

Related

• Choosing between Celery, arq and taskiq — measured comparison of the frameworks.
• Calling async code from Celery tasks — the bridge, and what it costs.
• Scheduling cron jobs inside an asyncio service — drift, locks and DST.
• Building a durable job queue on Postgres — the queue with no broker.
• Concurrent Execution & Worker Patterns — the parent section.