Disposing async engines on shutdown and in forked workers

Call await engine.dispose() while the event loop is still running when a process exits, and engine.dispose(close=False) in a freshly forked child so it abandons rather than closes the sockets it inherited — a pooled connection shared across a fork corrupts the protocol stream for both processes. This guide belongs to configuring async engines and connection pools.

Quick Answer

Two disposal calls, for two different situations. Getting them the wrong way round is worse than forgetting them.

Before — no disposal, and an engine inherited across a fork:

# app.py
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("postgresql+asyncpg://shop:secret@db/shop")
app = FastAPI()
# $ gunicorn -k uvicorn.workers.UvicornWorker --preload --workers 4 app:app
# Warm-up code that queried the database before the fork leaves four workers
# sharing one socket:
#   asyncpg.exceptions.ProtocolViolationError
#   ssl.SSLError: [SSL] decryption failed or bad record mac

After — disposal on shutdown, and a fresh pool in each worker:

from contextlib import asynccontextmanager

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

engine = create_async_engine("postgresql+asyncpg://shop:secret@db/shop", pool_size=10)
Session = async_sessionmaker(engine, expire_on_commit=False)


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Runs once per worker process, after any fork.
    engine.dispose(close=False)      # abandon anything inherited; do not close it
    yield
    await engine.dispose()           # close this worker's own pool, loop still alive


app = FastAPI(lifespan=lifespan)

engine.dispose(close=False) is synchronous and safe to call before anything has connected — it simply replaces the pool. await engine.dispose() at shutdown closes this worker's connections properly, while the event loop is still running, which is what stops the garbage-collector warnings described in fixing garbage collector non-checked-in connection warnings.

Execution Context & Async Workflow Integration

A pooled connection is a TCP socket, and os.fork() duplicates file descriptors. The child gets a descriptor referring to the same connection: the same sequence numbers, the same TLS session state, the same PostgreSQL backend. Neither process knows the other exists, so two statements can be written to one wire and two readers can consume one response stream.

The symptoms are memorable. With TLS, the session state diverges immediately and one side fails with decryption failed or bad record mac. Without TLS, asyncpg reports ProtocolViolationError, or — worst of all — one process receives rows belonging to the other's query. None of it is reproducible in a single-process development run, which is why this bug reaches production.

dispose(close=False) exists for exactly this. It discards the pool and replaces it with a fresh one, without closing the connections the old pool held, so the parent's sockets are left alone. The child then opens its own connections on first use. Calling plain dispose() in the child would close descriptors the parent is still using, which breaks the parent instead.

Which process models are affected comes down to one question: had the engine connected before the fork?

Gunicorn with --preload imports the application in the master and forks workers, so a module-level engine is created once and inherited. Nothing is broken until something connects pre-fork — a health check, a warm-up query, a migration check — and then everything is. Calling engine.dispose(close=False) at the start of each worker's lifespan makes it safe regardless.

Gunicorn or uvicorn without preload re-imports per worker, so each gets its own engine and nothing is shared.

Celery's pre-fork pool forks worker processes from a parent that has already imported the application. The worker_process_init signal runs in the child, which is where the engine must be discarded:

from celery.signals import worker_process_init

from shop.db import engine


@worker_process_init.connect
def _reset_engine(**_kwargs) -> None:
    engine.dispose(close=False)

multiprocessing with the fork start method has the same problem for any pool the parent touched; with spawn, the child re-imports and is unaffected.

There is a second, async-specific reason to dispose: an asyncpg connection is bound to the event loop that created it. A process that runs asyncio.run() more than once — a CLI with several commands, a test suite with function-scoped loops — must dispose the engine before the loop closes, or the next loop inherits connections it cannot use. That failure, and the NullPool answer for per-invocation loops, is covered in using NullPool for serverless and AWS Lambda.

Resolving Warnings, Errors & Common Mistakes

The script case is worth its own snippet, because putting the disposal after asyncio.run() looks tidier and does not work:

import asyncio

from shop.db import engine
from shop.jobs import rebuild_search_index


async def main() -> None:
    try:
        await rebuild_search_index()
    finally:
        await engine.dispose()      # inside the loop


asyncio.run(main())
# NOT: asyncio.run(main()); asyncio.run(engine.dispose())  — a second loop,
# and the connections belong to the first one.

Tests hit the loop-binding problem most often, because pytest-asyncio's default event loop scope is per function while an engine fixture is usually per session. Two arrangements work: a session-scoped engine with a session-scoped event loop, or a function-scoped engine disposed in its own teardown. Mixing the two produces got Future attached to a different loop on the second test that touches the database:

import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine


@pytest_asyncio.fixture(scope="session")
async def engine(postgres_url):
    engine = create_async_engine(postgres_url)
    yield engine
    await engine.dispose()

With that fixture, the event loop must also be session-scoped — configured with asyncio_default_fixture_loop_scope = "session" in the pytest settings — which is the arrangement described in using pytest-asyncio fixtures with AsyncSession.

Advanced: Shutting Down Cleanly Under a Process Manager

Disposal on shutdown only helps if shutdown actually reaches the code. Under a process manager, that depends on signals, and the common failure is a worker killed before its lifespan shutdown runs — which leaves connections for the server to time out and, worse, transactions to roll back on their own schedule.

Three things make the shutdown path reliable.

A grace period long enough for in-flight work. Kubernetes sends SIGTERM, waits terminationGracePeriodSeconds, then sends SIGKILL. Uvicorn and gunicorn translate SIGTERM into a graceful shutdown: stop accepting connections, finish in-flight requests, run the lifespan shutdown. If the grace period is shorter than the longest request, the disposal is skipped. Thirty seconds is a common starting point; the requirement is "longer than your slowest request plus disposal".

Disposal that cannot hang. await engine.dispose() closes each pooled connection, and a connection whose server is unreachable can block. Bounding it keeps shutdown predictable:

import asyncio
import logging

log = logging.getLogger("shutdown")


async def dispose_with_timeout(engine, seconds: float = 5.0) -> None:
    try:
        async with asyncio.timeout(seconds):
            await engine.dispose()
    except TimeoutError:
        log.warning("engine disposal timed out; abandoning the pool")
        engine.dispose(close=False)     # give up on closing, but drop the pool

Background work finished before the engine goes. Anything the application started — a TaskGroup, a queue consumer, a scheduled refresh — is still holding sessions. Disposing the engine underneath it produces a burst of confusing errors in the logs on every deploy. Cancel and await those tasks first, then dispose:

from contextlib import asynccontextmanager

from fastapi import FastAPI

from shop.db import engine


@asynccontextmanager
async def lifespan(app: FastAPI):
    engine.dispose(close=False)
    tasks = start_background_tasks()
    try:
        yield
    finally:
        for task in tasks:
            task.cancel()
        for task in tasks:
            try:
                await task
            except asyncio.CancelledError:
                pass
        await dispose_with_timeout(engine)

The ordering is the point: stop the things that use the engine, then dispose the engine. Reversing it is how a clean deploy produces a page of InterfaceError: connection is closed in the logs, which then gets misdiagnosed as a database problem.

One last check worth having in the shutdown path: assert that nothing is still checked out. engine.sync_engine.pool.checkedout() should be zero before disposal, and logging it when it is not tells you which deploys are leaking sessions — the leak-hunting technique from detecting idle-in-transaction sessions from async code.

A Checklist for Every Entry Point

Engine lifecycle is one of those concerns that is correct in the main application and wrong in the three smaller entry points nobody thought about. A short checklist, applied to each one, is enough.

The web application. Engine created at import or in the lifespan; dispose(close=False) at the start of the lifespan so a preloading server is safe; await engine.dispose() at the end, after background tasks are cancelled; grace period longer than the slowest request.

Workers. For Celery's pre-fork pool, worker_process_init calls dispose(close=False); for an async worker, the same lifespan pattern as the web application. A worker that processes long jobs also wants a longer statement_timeout and a smaller pool than the API, as in configuring connect_args and server_settings.

Scripts and management commands. Disposal in a finally block inside the coroutine. A script that runs for seconds does not need a pool at all: poolclass=NullPool removes the question.

Tests. One engine per event loop, disposed in the fixture teardown, and an assertion that nothing is left checked out after each test.

Migrations. NullPool and an explicit await connectable.dispose(), which the async env.py template already does.

Two invariants make the checklist auditable rather than aspirational. First, create_async_engine should appear in exactly one module outside tests — a grep that returns one hit means every entry point shares the same construction and the same disposal helper. Second, a process should never fork after connecting; if the deployment uses --preload, the dispose(close=False) call in the lifespan makes that safe whether or not anything connected.

Finally, verify the behaviour rather than trusting it. A test that forks and queries from both processes reproduces the shared-socket bug in a second, and fails loudly if the child does not reset the pool:

import os

import pytest


@pytest.mark.asyncio
async def test_child_process_does_not_share_connections(engine):
    async with engine.connect() as conn:          # ensure the pool has a connection
        await conn.execute(text("SELECT 1"))

    pid = os.fork()
    if pid == 0:                                   # child
        os._exit(0 if _child_can_query(engine) else 1)
    _, status = os.waitpid(pid, 0)
    assert os.waitstatus_to_exitcode(status) == 0

Where _child_can_query calls engine.dispose(close=False), runs its own asyncio.run() with a query, and returns whether it succeeded. Removing the dispose(close=False) line should make that test fail — which is the proof that the guard is doing something.

Frequently Asked Questions

Related

• Configuring Async Engines and Connection Pools — The parent guide: engine and pool parameters end to end.
• Setting up an async engine from scratch — Creating the engine and wiring the lifespan.
• Using NullPool for serverless and AWS Lambda — Connections and event loops per invocation.
• Fixing garbage collector non-checked-in connection warnings — What a missing disposal looks like in the logs.