← Back to the section

When Kubernetes sends a SIGTERM signal, the application receives a command: "finish your work cleanly." HTTP requests can still be served to completion, the database can be closed correctly — but what to do with background tasks that run in a loop?

This is the most dangerous spot during shutdown. If you kill a task midway, the transaction rolls back, but the side effect — a sent HTTP request or a message to Kafka — has already happened. Let's break down piece by piece how to handle this.

Why a sudden stop of an asyncio task is dangerous

Imagine a background task that processes an order queue every half a second. Inside it makes a database request and sends an event to an external service.

If SIGTERM arrives in the middle — between the send and the commit to the database — you get an inconsistency: the external service thinks the order is processed, while the database doesn't. On the next start the task will process the same order again.

The solution: let the task finish the current iteration, but not start a new one.

How to correctly stop an asyncio.Task

In FastAPI, background tasks are created via asyncio.create_task() and started in the lifespan function — the application lifecycle handler.

Here's a typical example with a correct stop:

from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncio
import structlog

log = structlog.get_logger()

async def order_sync_worker(ready: dict) -> None:
    while ready["accepting"]:
        try:
            await _process_pending_orders()
            await asyncio.sleep(0.5)
        except asyncio.CancelledError:
            log.info("order_sync_worker.cancelled")
            await _process_pending_orders()   # finish the last iteration
            raise
        except Exception:
            log.exception("order_sync_worker.error")
            await asyncio.sleep(2)

@asynccontextmanager
async def lifespan(app: FastAPI):
    ready = {"accepting": True}
    app.state.ready = ready

    task = asyncio.create_task(order_sync_worker(ready), name="order-sync")
    log.info("order_sync_worker.started")

    yield

    log.info("shutdown.begin")
    ready["accepting"] = False
    task.cancel()
    await asyncio.gather(task, return_exceptions=True)
    log.info("shutdown.tasks_done")

What happens on SIGTERM:

  1. uvicorn receives SIGTERM and enters the lifespan shutdown phase.
  2. ready["accepting"] = False — the task won't start a new iteration after the current one.
  3. task.cancel() — a CancelledError exception is thrown into the task.
  4. The task catches CancelledError, finishes the current chunk of work, then does raise — propagates the exception upward.
  5. await asyncio.gather(task, return_exceptions=True) — we wait for the task to actually finish.

A common mistake is to write task.cancel() and not wait: the task keeps working for some more time, but the application already considers itself stopped. You always need an await after the cancel.

Another mistake is to catch CancelledError and swallow it (not do raise). Then the task stays active and gather will wait forever.

APScheduler: don't kill a job midway

If the background logic is implemented via APScheduler, the principle is the same — you need to let the current run finish:

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from contextlib import asynccontextmanager
from fastapi import FastAPI
import structlog

log = structlog.get_logger()

async def sync_products() -> None:
    async with async_session() as session:
        pending = await session.execute(
            select(Product).where(Product.synced_at.is_(None)).limit(20)
        )
        for product in pending.scalars():
            await _push_to_warehouse(product)
            product.synced_at = datetime.now(timezone.utc)
        await session.commit()

@asynccontextmanager
async def lifespan(app: FastAPI):
    scheduler = AsyncIOScheduler()
    scheduler.add_job(sync_products, "interval", seconds=1, id="product-sync")
    scheduler.start()
    log.info("scheduler.started")

    yield

    log.info("shutdown.scheduler.begin")
    scheduler.shutdown(wait=True)   # wait for the current run
    log.info("shutdown.scheduler.done")

scheduler.shutdown(wait=True) — APScheduler waits for the executing job to finish. New runs don't happen in the meantime. If you write wait=False, the scheduler kills the job immediately — right in the middle of the loop that saves to the database.

Outbox relay: a readiness flag instead of while True

The relay is a task that reads events from a queue table (the outbox) and sends them to Kafka or another broker. It runs in an infinite loop.

The bad version:

# can't be stopped without a hard interruption midway
async def relay_loop() -> None:
    while True:
        await _publish_batch()
        await asyncio.sleep(0.1)

The correct one — via a readiness flag, as in the order_sync_worker example above:

async def outbox_relay(ready: dict) -> None:
    while ready["accepting"]:
        try:
            published = await _publish_outbox_batch()
            if published == 0:
                await asyncio.sleep(0.5)
        except asyncio.CancelledError:
            log.info("outbox_relay.cancelled, finishing current batch")
            await _publish_outbox_batch()   # finish the last batch
            raise
        except Exception:
            log.exception("outbox_relay.error")
            await asyncio.sleep(2)

async def _publish_outbox_batch() -> int:
    async with async_session() as session:
        async with session.begin():
            rows = await session.execute(
                select(OutboxEvent)
                .where(OutboxEvent.published_at.is_(None))
                .order_by(OutboxEvent.id)
                .limit(50)
                .with_for_update(skip_locked=True)
            )
            events = rows.scalars().all()
            if not events:
                return 0

            for event in events:
                await producer.send_and_wait(
                    event.topic,
                    key=event.partition_key.encode(),
                    value=event.payload.encode(),
                )
                event.published_at = datetime.now(timezone.utc)

            return len(events)

.with_for_update(skip_locked=True) — pessimistic row locking. While one pod is processing a batch, another pod won't take the same rows. On SIGTERM the current batch is finished, and the unlocked rows will be picked up by another pod.

A long HTTP call inside a task — a danger zone

Sometimes a background task makes an external HTTP call with retries — for example, charging a card:

# dangerous: SIGTERM during a retry → the payment went through, the DB isn't updated
async def charge_customer(customer_id: UUID, amount: Decimal) -> None:
    result = await payment_client.charge(customer_id, amount)  # 200 OK
    await session.commit()  # if SIGTERM here — inconsistency

The safe approach — write the intent into the outbox and exit. The relay will send it when the time comes:

async def enqueue_charge(
    session: AsyncSession,
    customer_id: UUID,
    amount: Decimal,
    idempotency_key: str,
) -> None:
    session.add(OutboxEvent(
        topic="payments.charge-requested",
        partition_key=str(customer_id),
        payload=json.dumps({
            "customer_id": str(customer_id),
            "amount": str(amount),
            "idempotency_key": idempotency_key,
        }),
    ))
    await session.flush()

Now SIGTERM is safe at any moment: either the row is written to the outbox (the relay will send it) or the transaction rolled back (there's no row). The idempotency_key protects against a double send on retries.

The order of closing resources

An important detail: the database connection pool (engine.dispose()) needs to be closed after all tasks have finished, not before. Tasks may still run database queries while finishing up — they need the pool.

@asynccontextmanager
async def lifespan(app: FastAPI):
    ready = {"accepting": True}
    app.state.ready = ready

    relay_task = asyncio.create_task(outbox_relay(ready), name="outbox-relay")
    sync_task = asyncio.create_task(order_sync_worker(ready), name="order-sync")

    yield

    log.info("shutdown.begin")

    ready["accepting"] = False
    relay_task.cancel()
    sync_task.cancel()

    await asyncio.gather(relay_task, sync_task, return_exceptions=True)
    log.info("shutdown.tasks_done")

    await producer.stop()        # flush the buffer and close the Kafka producer
    await consumer.stop()        # commit offsets and close

    await engine.dispose()       # close the SQLAlchemy pool — last
    log.info("shutdown.complete")

In short

  • task.cancel() without await — the task hangs. Always use await asyncio.gather(task, return_exceptions=True).
  • CancelledError needs to be caught, the critical section finished (commit, send), then raise.
  • APScheduler — always scheduler.shutdown(wait=True). wait=False kills the job midway through an iteration.
  • The relay loop must check the readiness flag (while ready["accepting"]), not be while True.
  • A long HTTP call with retries — move it into the outbox, don't do it inline in a background task.
  • engine.dispose() is the very last step. Tasks still use the database while finishing up.
  • Budgets and observability — the total 60-second budget and shutdown metrics.
  • Database and persistence — engine.dispose(), transactions on shutdown.
  • HTTP drain — uvicorn graceful, preStop sleep.
  • In-flight idempotency — protection against double processing on retry.
  • Kafka on shutdown — consumer.stop(), producer.stop(), manual commit.