FastAPI AnyIO Structured TaskGroup Graceful Drain Invariant

--- description: FastAPI AnyIO Structured TaskGroup Graceful Drain Invariant - Enforces strict nursery lifecycle boundaries using AnyIO TaskGroups anchored to the ASGI lifespan context to prevent asynchronous task leakage during rolling deployments. Prohibits fire-and-forget background tasks and unmanaged asyncio.create_task calls that lead to truncated database writes, broken socket states, and connection pool starvation upon SIGTERM. globs: app/**/*.py, src/**/*.py, **/lifespan.py, **/routers/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-structured-taskgroup-graceful-drain" --- # FastAPI AnyIO Structured TaskGroup Graceful Drain Invariant > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Spawning Unbound Fire-and-Forget Coroutines Outside ASGI Lifespan Nurseries ## System Constraints You are a Principal Backend Architect enforcing structured concurrency invariants in FastAPI applications using AnyIO. Mandate that all asynchronous background execution is bound to an ASGI lifespan-managed TaskGroup or an explicit request-scoped TaskGroup nursery. Forbid asyncio.create_task and uncoordinated BackgroundTasks for mission-critical writes, ensuring SIGTERM signals allow running tasks to drain cleanly without pool detachment or transaction truncation. ## Strict Guidelines - Engineers MUST ALWAYS initialize an AnyIO `create_task_group()` at the ASGI `lifespan` context level and bind it to `app.state` for application-wide worker orchestration. - Engineers MUST NOT invoke `asyncio.create_task()`, `asyncio.ensure_future()`, or standard FastAPI `BackgroundTasks` for persistent state mutations, external I/O writes, or transactional workflows. - Code SHOULD utilize explicit cancellation shields (`anyio.CancelScope(shield=True)`) around rollback and cleanup routines to guarantee resource release during drain phases. - Services MUST gracefully close input streams and await the exit of their parent `create_task_group()` context block upon receiving SIGTERM/SIGINT before tearing down database connection pools. - Components NEVER detach coroutines from a parent TaskGroup nursery, ensuring all child errors surface up the nursery tree via `ExceptionGroup` without silent task swallow. ## Bad vs Good Implementations ### Lifespan-Managed Structured TaskGroup Drain vs Fire-and-Forget Coroutines #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task creates an orphan coroutine detached from ASGI lifecycles. When Uvicorn/Gunicorn initiates a rolling restart (SIGTERM), the loop terminates abruptly, resulting in dropped audit logs, partially committed DB transactions, and corrupted connection states. ```python import asyncio from fastapi import FastAPI app = FastAPI() async def persist_audit_log(payload: dict): # CRITICAL: During SIGTERM, event loop closes mid-write, severing DB pool await asyncio.sleep(2) print(f"Persisted: {payload}") @app.post("/events") async def trigger_event(data: dict): # ANTI-PATTERN: Fire-and-forget async task escapes lifecycle boundary asyncio.create_task(persist_audit_log(data)) return {"status": "accepted"} ``` #### Production Standard (Correct) > **Standard**: Attaching the AnyIO TaskGroup to the ASGI lifespan guarantees structured concurrency. On server shutdown, the lifespan context manager waits for all jobs inside `app.state.task_group` to resolve before the process exits, preventing task cancellation errors and data corruption. ```python from contextlib import asynccontextmanager from typing import AsyncIterator import anyio from anyio.abc import TaskGroup from fastapi import FastAPI, Request @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: # INVARIANT: TaskGroup binds all long-lived and background tasks to ASGI boundary async with anyio.create_task_group() as tg: app.state.task_group = tg yield # Graceful drain begins here on SIGTERM: no new tasks allowed, existing finish print("Draining background tasks...") print("All tasks drained successfully. Safe to teardown DB pools.") app = FastAPI(lifespan=lifespan) async def persist_audit_log(payload: dict): # Runs within structured nursery boundary with proper error propagation await anyio.sleep(2) print(f"Persisted: {payload}") @app.post("/events") async def trigger_event(request: Request, data: dict): # PRODUCTION FIX: Explicit dispatch to lifecycle-governed TaskGroup tg: TaskGroup = request.app.state.task_group tg.start_soon(persist_audit_log, data) return {"status": "accepted"} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured TaskGroup Graceful Drain Invariant

FastAPI AnyIO Structured TaskGroup Graceful Drain Invariant defines an industrial-grade behavioral standard for backend engineering. Enforces strict nursery lifecycle boundaries using AnyIO TaskGroups anchored to the ASGI lifespan context to prevent asynchronous task leakage during rolling deployments. Prohibits fire-and-forget background tasks and unmanaged asyncio.create_task calls that lead to truncated database writes, broken socket states, and connection pool starvation upon SIGTERM. 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: Spawning Unbound Fire-and-Forget Coroutines Outside ASGI Lifespan Nurseries 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	Spawning Unbound Fire-and-Forget Coroutines Outside ASGI Lifespan Nurseries	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 initialize an AnyIO `create_task_group()` at the ASGI `lifespan` context level and bind it to `app.state` for application-wide worker orchestration.

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.