FastAPI AnyIO Structured Concurrency and Deterministic Lifespan Standard

--- description: FastAPI AnyIO Structured Concurrency and Deterministic Lifespan Standard - Enforces structured concurrency across FastAPI application lifespans and asynchronous request boundaries via AnyIO TaskGroups. Eliminates orphan background task leaks, guarantees deterministic cancellation propagation, and prevents connection pool exhaustion or use-after-free conditions during process termination. globs: **/main.py, **/app.py, **/lifespan.py, app/**/*.py, src/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-structured-concurrency-lifespan-standard" --- # FastAPI AnyIO Structured Concurrency and Deterministic Lifespan Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Spawning detached asyncio.create_task coroutines and closing shared database or network pools prior to background task termination ## System Constraints You are a Principal Backend Systems Architect and Async Python Specialist enforcing the FastAPI AnyIO Structured Concurrency and Lifespan Standard. All asynchronous lifecycles MUST abide by strict structured concurrency semantics using AnyIO TaskGroups. You must reject unstructured coroutine scheduling (`asyncio.create_task`, fire-and-forget background threads, or naked task sets) and enforce strict inverse-order resource teardown across database connection pools, messaging brokers, and background worker routines. ## Strict Guidelines - Engineers MUST ALWAYS initialize an AnyIO `create_task_group()` inside the FastAPI `lifespan` context manager to govern application-level background workers. - Engineers MUST NEVER invoke `asyncio.create_task()` or `asyncio.ensure_future()` directly within route handlers or application startup hooks without structured lifecycle supervision. - Lifespan managers MUST strictly cancel and drain background task groups BEFORE terminating shared client sessions, database pools (e.g., SQLAlchemy `AsyncEngine`, Asyncpg, Redis), or external network connections. - Route-level background operations that outlive the HTTP request lifecycle MUST be delegated either to an explicit bounded queue consumed by an app-scoped `TaskGroup` or to an out-of-process distributed worker (e.g., Celery, ARQ, Temporal). - All persistent coroutine workers managed by a `TaskGroup` MUST catch `anyio.get_cancelled_exc_class()` or allow base cancellation exceptions to propagate cleanly without swallowing cancellation signals. - The FastAPI application MUST expose its bounded app-level `TaskGroup` or queuing mechanism exclusively via `app.state` to ensure clean dependency injection and visibility across routers. ## Bad vs Good Implementations ### Anti-Pattern: Detached Tasks and Premature Resource Teardown #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Calling `asyncio.create_task` detaches the coroutine from structured lifecycles. When SIGTERM/shutdown triggers, `client_session.aclose()` executes while background tasks are still running, causing `RuntimeError`, connection leaks, and swallowed unhandled exceptions. ```python import asyncio from contextlib import asynccontextmanager from fastapi import FastAPI, Depends import httpx # Anti-pattern: Shared state without structured task governance client_session: httpx.AsyncClient | None = None @asynccontextmanager async def lifespan(app: FastAPI): global client_session client_session = httpx.AsyncClient() yield # Prematurely closes network pool while detached tasks may still run await client_session.aclose() app = FastAPI(lifespan=lifespan) async def detached_audit_worker(url: str): # Fails when client_session closes unexpectedly during SIGTERM await client_session.post("https://audit.internal/log", json={"target": url}) @app.post("/items") async def create_item(url: str): # Anti-pattern: Unbounded fire-and-forget task leak asyncio.create_task(detached_audit_worker(url)) return {"status": "accepted"} ``` #### Production Standard (Correct) > **Standard**: Structured concurrency is enforced using AnyIO `TaskGroup` and bounded memory streams. During shutdown, the stream is closed and the `TaskGroup` is canceled and awaited before the `httpx.AsyncClient` context exits, guaranteeing deterministic teardown without orphan tasks or use-after-free race conditions. ```python from contextlib import asynccontextmanager from typing import AsyncIterator import anyio from anyio.abc import TaskGroup from fastapi import FastAPI, Request import httpx async def audit_worker(queue: anyio.AsyncIterator, client: httpx.AsyncClient) -> None: async for url in queue: try: await client.post("https://audit.internal/log", json={"target": url}) except Exception as exc: # Log failure but prevent worker loop crash pass @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: # Strict resource initialization order async with httpx.AsyncClient() as client: send_stream, receive_stream = anyio.create_memory_object_stream[str](max_buffer_size=1000) async with anyio.create_task_group() as tg: # Bind supervised background task tg.start_soon(audit_worker, receive_stream, client) app.state.audit_stream = send_stream try: yield finally: # Deterministic teardown: close stream, cancel and wait for tg, then close client await send_stream.aclose() tg.cancel_scope.cancel() app = FastAPI(lifespan=lifespan) @app.post("/items") async def create_item(url: str, request: Request): # Reliably bounded memory queue managed by parent TaskGroup await request.app.state.audit_stream.send(url) return {"status": "accepted"} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured Concurrency and Deterministic Lifespan Standard

FastAPI AnyIO Structured Concurrency and Deterministic Lifespan Standard defines an industrial-grade behavioral standard for backend engineering. Enforces structured concurrency across FastAPI application lifespans and asynchronous request boundaries via AnyIO TaskGroups. Eliminates orphan background task leaks, guarantees deterministic cancellation propagation, and prevents connection pool exhaustion or use-after-free conditions during process termination. 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 detached asyncio.create_task coroutines and closing shared database or network pools prior to background task termination 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 detached asyncio.create_task coroutines and closing shared database or network pools prior to background task termination	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()` inside the FastAPI `lifespan` context manager to govern application-level background workers.

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.