FastAPI AnyIO Structured TaskGroup Scoping and Connection Leak Prevention

--- description: FastAPI AnyIO Structured TaskGroup Scoping and Connection Leak Prevention - Mandates that all asynchronous concurrency within FastAPI endpoints and worker routines is bounded by strict AnyIO TaskGroup scopes. Unbounded concurrency primitives like asyncio.create_task bypass FastAPI cancellation tokens during client disconnections, causing orphaned database pool checkouts and catastrophic resource starvation. Enforces fail-fast exception boundaries and deterministic teardown across all child coroutines. globs: app/**/*.py, src/**/*.py, api/**/*.py, services/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-taskgroup-scope-guardrail" --- # FastAPI AnyIO Structured TaskGroup Scoping and Connection Leak Prevention > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unbounded Background Task Spawning with Escaped Async Resources on Client Disconnect ## System Constraints You are a Principal Backend Architect enforcing structured concurrency standards in FastAPI applications. You must enforce that no bare `asyncio.create_task`, unbounded fire-and-forget coroutines, or unbounded thread pools execute inside request-response lifecycles. All concurrent branching must occur within `anyio.create_task_group()` context managers, propagating cancellations deterministically and safeguarding pooled resources (SQLAlchemy sessions, Redis clients, HTTP client sessions) from leaking upon client disconnection or uncaught sub-task failure. ## Strict Guidelines - Engineers MUST ALWAYS encapsulate concurrent child coroutines inside an `async with anyio.create_task_group() as tg:` context block. - Engineers MUST NOT invoke `asyncio.create_task()`, `asyncio.ensure_future()`, or `asyncio.gather()` inside request handling paths without an explicit structured lifecycle wrapper. - Engineers MUST pass database sessions and transaction contexts explicitly down to child tasks only when scoped within the same transaction lifecycle, or acquire isolated pooled connections per task inside the task group. - Code MUST handle `anyio.get_cancelled_exc_class()` or rely on standard task group semantics to ensure child tasks cancel immediately when a peer task fails or when the client aborts the HTTP connection. - Background work that outlives the HTTP request lifecycle MUST NOT run in detached tasks; it MUST be dispatched via external task queues (e.g., Celery, ARQ, SAQ) or managed via FastAPI lifespan application-level TaskGroups. ## Bad vs Good Implementations ### Concurrent Query Execution with Leaked DB Connections vs. Structured AnyIO TaskGroup #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using bare `asyncio.create_task` and `asyncio.gather` breaks structured concurrency. If a client disconnects mid-flight, FastAPI cancels the handler, but orphaned tasks continue consuming database pool connections. Furthermore, if `task1` throws an exception, `task2` remains uncancelled, causing silent resource starvation. ```python import asyncio from fastapi import APIRouter, Depends from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db router = APIRouter() async def fetch_user_data(session: AsyncSession, user_id: int): # If the client disconnects, this task continues executing in the background, # keeping the checked-out database session alive indefinitely or until timeout. result = await session.execute(f"SELECT * FROM users WHERE id = {user_id}") return result.scalar_one() @router.get("/users/{user_id}/dashboard") async def get_dashboard(user_id: int, db: AsyncSession = Depends(get_db)): # ANTI-PATTERN: Bare asyncio.create_task causes silent leaks upon cancellation task1 = asyncio.create_task(fetch_user_data(db, user_id)) task2 = asyncio.create_task(fetch_user_data(db, user_id)) data1, data2 = await asyncio.gather(task1, task2) return {"data1": data1, "data2": data2} ``` #### Production Standard (Correct) > **Standard**: The bounded `anyio.create_task_group()` guarantees structured concurrency. If the client disconnects or any subtask raises an exception, AnyIO immediately cancels all sibling tasks and guarantees clean context manager exit, returning checked-out connections back to the connection pool deterministically. ```python import anyio from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db_session_factory router = APIRouter() async def fetch_profile(session_factory, user_id: int, results: dict): async with session_factory() as session: result = await session.execute(f"SELECT * FROM profiles WHERE user_id = :id", {"id": user_id}) results["profile"] = result.scalar_one_or_none() async def fetch_metrics(session_factory, user_id: int, results: dict): async with session_factory() as session: result = await session.execute(f"SELECT * FROM metrics WHERE user_id = :id", {"id": user_id}) results["metrics"] = result.scalar_one_or_none() @router.get("/users/{user_id}/dashboard") async def get_dashboard(user_id: int, session_factory = Depends(get_db_session_factory)): results = {} try: async with anyio.create_task_group() as tg: tg.start_soon(fetch_profile, session_factory, user_id, results) tg.start_soon(fetch_metrics, session_factory, user_id, results) except ExceptionGroup as eg: raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Task execution failed") from eg return results ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured TaskGroup Scoping and Connection Leak Prevention

FastAPI AnyIO Structured TaskGroup Scoping and Connection Leak Prevention defines an industrial-grade behavioral standard for backend engineering. Mandates that all asynchronous concurrency within FastAPI endpoints and worker routines is bounded by strict AnyIO TaskGroup scopes. Unbounded concurrency primitives like asyncio.create_task bypass FastAPI cancellation tokens during client disconnections, causing orphaned database pool checkouts and catastrophic resource starvation. Enforces fail-fast exception boundaries and deterministic teardown across all child coroutines. 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 Background Task Spawning with Escaped Async Resources on Client Disconnect 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 Background Task Spawning with Escaped Async Resources on Client Disconnect	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 encapsulate concurrent child coroutines inside an `async with anyio.create_task_group() as tg

Engineers MUST ALWAYS encapsulate concurrent child coroutines inside an `async with anyio.create_task_group() as tg:` context block.

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.