6 · Async, Jobs & Event Loops

This section carries two production bugs. Both are the kind that pass every local test and fail in deployment, and both come from the same root: an event loop is not an implementation detail you can ignore.

Why async at all

A text-to-SQL request is almost entirely waiting:

model call     ~1500 ms   waiting on network
database query  ~200 ms   waiting on I/O
answer call     ~800 ms   waiting on network
actual CPU        ~5 ms

With threads you would need one per concurrent request, each with its own stack. With asyncio, one thread interleaves hundreds of requests, because every one of those waits is an await that yields control.

Parallelism inside the pipeline

Answer generation (slow, network) and chart generation (fast, local) are independent:

answer_task = asyncio.create_task(generate_answer(ctx, provider))
generate_chart(ctx)          # runs while the model call is in flight
await answer_task

create_task schedules immediately; the synchronous chart work fills the gap; then we await. Saves roughly the chart’s duration on every query.

Trap 1 — create_task and weak references

# Awaited rather than fire-and-forget: a bare asyncio.create_task is only
# weakly referenced by the event loop and can be garbage collected before it runs.
await write_cache(ctx, ttl=settings.query_cache_ttl_seconds)

The event loop holds tasks weakly. If nothing keeps a strong reference, a fire-and-forget task can be collected before it completes. Cache writes silently vanished under load — the worst kind of bug, because nothing errors.

Two valid fixes:

await write_cache(...)                     # 1. just await it
 
_running: set[asyncio.Task] = set()        # 2. keep a strong reference
handle = asyncio.create_task(...)
_running.add(handle)
handle.add_done_callback(_running.discard)

The pipeline uses the first; src/workers/dispatch.py uses the second, because there the whole point is not to block.

Trap 2 — connections belong to a loop

The one that broke production.

Celery tasks are synchronous functions; everything else here is async. The bridge:

# src/workers/runner.py
def run_async(coro):
    """Run an async coroutine to completion from a synchronous Celery task."""
    return asyncio.run(coro)

asyncio.run() creates a fresh event loop, runs the coroutine, and closes it. The original code used asyncio.get_event_loop(), which is deprecated since 3.10, raises on 3.12+ when no loop is running in the current thread, and — when it did return one — returned a shared loop, so a failing task could leave it broken for every later task in that worker.

So far so good. Then the first production schema sync failed:

got Future <Future pending cb=[BaseProtocol._on_waiter_completed()]>
attached to a different loop

What actually happened

API process
├── main event loop  ──── opened Postgres connections into the shared pool
└── job thread
    └── asyncio.run() → NEW loop  ──── grabbed a pooled connection
                                       created on the OTHER loop  ✗

A SQLAlchemy engine caches pooled connections bound to the loop that opened them, and an asyncpg connection cannot move between loops.

The fix

@asynccontextmanager
async def worker_session() -> AsyncIterator[AsyncSession]:
    engine = create_async_engine(settings.database_url, poolclass=NullPool, echo=False)
    factory = async_sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
    try:
        async with factory() as session:
            yield session
    finally:
        await engine.dispose()

NullPool means every connection is opened on the loop that asks for it and closed when released, so nothing is ever shared across loops. Disposing the engine releases it before the loop is torn down.

The cost is a fresh connect per job. That is the right trade for work that runs for seconds and is measured in jobs per hour, not requests per second — the API keeps its pooled engine.

Inline jobs — and why a thread

Render’s free plan has no background workers, so jobs can run inside the API process:

def dispatch(task, *args) -> str:
    settings = get_settings()
    if not settings.inline_workers:
        return task.delay(*args).id            # normal: Celery broker
 
    task_id = str(uuid.uuid4())
    handle = asyncio.create_task(asyncio.to_thread(_run_inline, task, args, task_id))
    _running.add(handle)
    handle.add_done_callback(_running.discard)
    return task_id

Why asyncio.to_thread and not await?

Every task ends in run_async() → asyncio.run(), which raises RuntimeError if a loop is already running in the calling thread — and inside a FastAPI request, one always is. A worker thread has no loop, so asyncio.run() is legal there.

Why not Celery’s task_always_eager?

It runs the task synchronously in the caller’s thread, which hits the same asyncio.run() problem and blocks the HTTP response until the job finishes — turning an async job endpoint into a synchronous one.

The API contract is identical either way: a job ID comes back immediately and GET /v1/jobs/{id} reports progress. tests/test_dispatch.py pins both properties — that dispatch returns before the job finishes, and that asyncio.run() inside the task does not raise.

Queue routing — silence is the failure mode

DEFAULT_QUEUE = "celery"
QUEUES = ("schema", "insight", DEFAULT_QUEUE)
 
task_routes = {
    "src.workers.schema_worker.*":  {"queue": "schema"},
    "src.workers.insight_worker.*": {"queue": "insight"},
}

A worker started without -Q consumes only the default queue. With these routes, such a worker accepts a task, returns a job ID, and then never runs anything. Nothing errors — POST /v1/sources/{id}/sync just returns a job that stays pending forever.

Since schema sync must succeed before any question can be asked, the deployment would have looked healthy and been unusable.

The command must consume every queue:

celery -A src.workers.celery_app worker -Q schema,insight,celery

tests/test_celery_routing.py fails the build if a task routes somewhere no worker listens — and it reads the real deploy files, so dropping -Q from docker-compose.yml or render.yaml breaks CI rather than production.

Interview questions this section answers

“What’s the difference between concurrency and parallelism here?” The pipeline is concurrent — one thread interleaving many awaits. It is not parallel; there is one CPU doing the coordination. Actual parallelism comes from separate worker processes.

“How do you call blocking code from async?” asyncio.to_thread() (or a thread-pool executor). Calling it directly blocks the whole loop, stalling every other request in that process.

“How do you call async code from sync?” asyncio.run() — but only from a thread with no running loop. That constraint is exactly what shaped the inline dispatcher.

“Why NullPool for workers but a pool for the API?” The API has one long-lived loop, so pooling is a pure win. Workers get a fresh loop per task, so pooled connections would be bound to a dead loop. Different lifecycles, different pooling.