FastAPI ASGI Lifespan Structured TaskGroup Drain Boundary

--- description: FastAPI ASGI Lifespan Structured TaskGroup Drain Boundary - Enforces deterministic structured concurrency boundaries for asynchronous background work within FastAPI ASGI applications. Guarantees that any spawned background coroutines are tethered to an application-scoped or request-scoped TaskGroup that safely drains within lifespan shutdown before database and network connection pools close during Kubernetes SIGTERM events. globs: app/**/*.py, src/**/*.py, api/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-asgi-structured-concurrency-drain-boundary" --- # FastAPI ASGI Lifespan Structured TaskGroup Drain Boundary > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unbounded asyncio.create_task and BackgroundTasks Leaks Causing Database Connection Poisoning During SIGTERM ## System Constraints You are a Principal Engineer enforcing robust ASGI structured concurrency and deterministic shutdown boundaries in Python FastAPI services. Mandate that raw asyncio.create_task and fire-and-forget Starlette BackgroundTasks are strictly prohibited for work dependent on application resources (such as DB pools, HTTP clients, and message queues). Require explicit TaskGroup or AsyncExitStack management tied directly to the ASGI lifespan context to ensure graceful task draining on Kubernetes SIGTERM termination sequences before infrastructure clients disconnect. ## Strict Guidelines - Engineers MUST ALWAYS manage non-trivial background tasks using structured concurrency boundaries (e.g., anyio.create_task_group() or Python 3.11+ asyncio.TaskGroup) encapsulated within an ASGI lifespan context manager. - Engineers MUST NOT invoke asyncio.create_task() or Starlette BackgroundTasks for coroutines that require database connections, Redis clients, or shared HTTP connection pools. - Services MUST cancel and drain remaining background tasks with explicit timeouts prior to disposing of database engines (e.g., SQLAlchemy create_async_engine.dispose()) or closing connection pools. - Engineers MUST configure readiness probes in Kubernetes to flip to unhealthy immediately upon receiving SIGTERM, allowing in-flight requests and drained TaskGroups to finalize within terminationGracePeriodSeconds. - Components NEVER swallow asyncio.CancelledError inside background coroutines; tasks must cleanup resources and re-raise or allow CancelledError to propagate to the enclosing TaskGroup. ## Bad vs Good Implementations ### Unbounded Background Task Spawning vs. Lifespan TaskGroup Drain Boundary #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Spawning raw tasks via asyncio.create_task leaves coroutines unmanaged. When Kubernetes terminates the pod, Uvicorn exits and closes the SQLAlchemy engine immediately, aborting the coroutine mid-execution with connection errors and leaving orphaned state. ```python import asyncio from fastapi import FastAPI, Depends from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db") app = FastAPI() async def write_audit_log(user_id: str): # ANTI-PATTERN: Leaks beyond request and lifespan lifecycle. # During SIGTERM, engine is disposed while this task is still running, # causing InterfaceError, orphaned connections, and uncommitted data. async with AsyncSession(engine) as session: await session.execute("INSERT INTO audit_logs (user_id) VALUES (:id)", {"id": user_id}) await session.commit() @app.post("/users/{user_id}/action") async def trigger_action(user_id: str): asyncio.create_task(write_audit_log(user_id)) return {"status": "accepted"} ``` #### Production Standard (Correct) > **Standard**: The AnyIO TaskGroup binds the lifespan of all background jobs to the ASGI app lifespan. When Uvicorn initiates shutdown on SIGTERM, the lifespan block exits, waits for all active tasks to complete or cancel gracefully, and only then tears down the database engine. ```python import asyncio from contextlib import asynccontextmanager from typing import AsyncGenerator import anyio from anyio.abc import TaskGroup from fastapi import FastAPI, Request from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db", pool_pre_ping=True) @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: # SECURE: Explicit task group bound to ASGI lifespan boundary async with anyio.create_task_group() as tg: app.state.background_tg = tg yield # Graceful drain initiation: cancel and wait with timeout tg.cancel_scope.cancel() # Infrastructure is ONLY disposed after all tasks have drained await engine.dispose() app = FastAPI(lifespan=lifespan) async def write_audit_log(user_id: str): try: async with AsyncSession(engine) as session: await session.execute("INSERT INTO audit_logs (user_id) VALUES (:id)", {"id": user_id}) await session.commit() except anyio.get_cancelled_exc_class(): # Optional: handle graceful rollback if interrupted raise @app.post("/users/{user_id}/action") async def trigger_action(user_id: str, request: Request): tg: TaskGroup = request.app.state.background_tg tg.start_soon(write_audit_log, user_id) return {"status": "accepted"} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI ASGI Lifespan Structured TaskGroup Drain Boundary

FastAPI ASGI Lifespan Structured TaskGroup Drain Boundary defines an industrial-grade behavioral standard for backend engineering. Enforces deterministic structured concurrency boundaries for asynchronous background work within FastAPI ASGI applications. Guarantees that any spawned background coroutines are tethered to an application-scoped or request-scoped TaskGroup that safely drains within lifespan shutdown before database and network connection pools close during Kubernetes SIGTERM events. Modern software architectures require strict adherence to deterministic patterns to guarantee high availability, maintainability, and zero security vulnerabilities.

Framework Paradigm Shift & Drift MechanicsLLMs are trained on millions of open-source repositories containing outdated code, legacy idioms, and unpatched antipatterns. When coding assistants generate implementations for this stack, they default to common statistical tokens: Unbounded asyncio.create_task and BackgroundTasks Leaks Causing Database Connection Poisoning During SIGTERM Without an explicit system instruction file, developers spend hours debugging silent failures.

Failure Modes & Production Blast Radius

Hazard Analysis: What Breaks Without This Rule

The following failure modes have been cataloged from hundreds of hours of production code reviews where AI assistants produced non-compliant implementations:

Failure Mode	AI Symptom (The Anti-Pattern)	Severity	Production Impact & Remedy
AI Hallucination & Deprecation Drift	Unbounded asyncio.create_task and BackgroundTasks Leaks Causing Database Connection Poisoning During SIGTERM	Critical	Violating this industrial standard introduces regression bugs, security risks, and unexpected runtime failures in production. Enforce explicit behavioral system directives across all AI coding assistants in your repository.
Silent Architecture Inconsistency	AI assistants produce mixed paradigms across files, creating fragmented codebase conventions.	Medium	Increases cognitive load during code reviews and leads to maintenance debt across engineering teams. Pin project conventions in .cursor/rules, CLAUDE.md, and AGENTS.md so all team agents follow identical guardrails.

Architectural Directives in Detail

Step-by-Step Production Directives

Each rule constraint below is injected into the AI agent's system prompt to establish deterministic boundaries:

Engineers MUST ALWAYS manage non-trivial background tasks using structured concurrency boundaries (e.g., anyio.create_task_group() or Python 3.11+ asyncio.TaskGroup) encapsulated within an ASGI lifespan context manager.

Engineers MUST ALWAYS manage non-trivial background tasks using structured concurrency boundaries (e.g., anyio.create_task_group() or Python 3.11+ asyncio.TaskGroup) encapsulated within an ASGI lifespan context manager.

Rationale:AI models frequently generate unconstrained code here because statistical language distributions lack project-specific architectural context. Mandating this directive ensures that generated code adheres to verified industrial production requirements.