FastAPI AnyIO Structured Concurrency & Disconnect Cancellation Guardrail

--- description: FastAPI AnyIO Structured Concurrency & Disconnect Cancellation Guardrail - Enforces strict structured concurrency using AnyIO TaskGroups bound to incoming FastAPI request lifecycles. Guarantees that client HTTP disconnections deterministically trigger cancellation scopes across all nested coroutines, immediately aborting active database transactions and returning pooled connections to prevent connection starvation. globs: app/**/*.py, src/**/*.py, api/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-taskgroup-cancellation-propagation" --- # FastAPI AnyIO Structured Concurrency & Disconnect Cancellation Guardrail > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unbounded Background Tasks and Ignored Client Disconnections Leading to DB Pool Exhaustion ## System Constraints You are a Principal Backend Architect enforcing structured concurrency invariants in FastAPI applications. You must prevent orphaned background coroutines, zombie database transactions, and pool exhaustion by enforcing AnyIO TaskGroups and deterministic client disconnect cancellation propagation. ## Strict Guidelines - Engineers MUST NEVER use raw 'asyncio.create_task' or 'asyncio.gather' inside FastAPI endpoints to spawn unmonitored concurrent workloads. - Engineers MUST use 'anyio.create_task_group()' within an ASGI request scope when initiating concurrent asynchronous operations. - Code MUST monitor client connectivity via a concurrent polling watcher (e.g., inspecting 'await request.is_disconnected()') that cancels the enclosing 'anyio.CancelScope' immediately upon disconnect. - Database operations inside task groups MUST be managed within transactional context managers ('async with session.begin():') so AnyIO cancellation triggers automatic rollback and connection release. - Services MUST NOT suppress 'anyio.get_cancelled_exc_class()' or 'asyncio.CancelledError' with bare 'except Exception:' clauses; cancellation exceptions MUST be re-raised immediately. - Background tasks that outlive the HTTP request lifecycle MUST be offloaded to dedicated distributed task queues (e.g., Celery, ARQ, Temporal) instead of detached in-process coroutines. ## Bad vs Good Implementations ### Unstructured asyncio.create_task vs AnyIO TaskGroup with Disconnect Watcher #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task detaches the coroutine from the FastAPI request scope. If the client closes the connection, the task continues executing the 30-second query, holding the asyncpg/SQLAlchemy pool connection and eventually causing pool exhaustion under client churn. ```python import asyncio from fastapi import FastAPI, Depends, Request from sqlalchemy.ext.asyncio import AsyncSession from app.db import get_db_session app = FastAPI() async def heavy_query(session: AsyncSession, item_id: int): # Anti-pattern: Detached execution, continues even if client aborts return await session.execute(f"SELECT pg_sleep(30), id FROM items WHERE id = {item_id}") @app.get("/items/{item_id}") async def get_item(item_id: int, session: AsyncSession = Depends(get_db_session)): # Anti-pattern: asyncio.create_task creates orphan task # If client disconnects, task retains DB session and query runs to completion task = asyncio.create_task(heavy_query(session, item_id)) result = await task return {"data": result.scalars().all()} ``` #### Production Standard (Correct) > **Standard**: Binds both the worker and the disconnect watcher within an AnyIO TaskGroup. If the client disconnects, cancel_scope.cancel() immediately cancels fetch_item, rolling back the database transaction and returning the connection to the pool without leaking resources. ```python import anyio from fastapi import FastAPI, Depends, Request, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, text from app.db import get_db_session app = FastAPI() async def disconnect_watcher(request: Request, cancel_scope: anyio.CancelScope) -> None: """Continuously monitors client connection state and triggers cancellation.""" while not cancel_scope.cancel_called: if await request.is_disconnected(): cancel_scope.cancel() return await anyio.sleep(0.2) async def fetch_item(session: AsyncSession, item_id: int, results: dict) -> None: async with session.begin(): query = text("SELECT pg_sleep(30), id FROM items WHERE id = :id") res = await session.execute(query, {"id": item_id}) results["item"] = res.scalar_one_or_none() @app.get("/items/{item_id}") async def get_item(request: Request, item_id: int, session: AsyncSession = Depends(get_db_session)): results: dict = {} try: async with anyio.create_task_group() as tg: # Watch client disconnection concurrently tg.start_soon(disconnect_watcher, request, tg.cancel_scope) # Run workload bound to same task group tg.start_soon(fetch_item, session, item_id, results) # Cancel watcher once main work finishes tg.cancel_scope.cancel() except Exception as exc: # Verify if cancellation was initiated by client disconnect if await request.is_disconnected(): raise HTTPException(status_code=status.HTTP_499_CLIENT_CLOSED_REQUEST, detail="Client disconnected") from exc raise return {"data": results.get("item")} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured Concurrency & Disconnect Cancellation Guardrail

FastAPI AnyIO Structured Concurrency & Disconnect Cancellation Guardrail defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency using AnyIO TaskGroups bound to incoming FastAPI request lifecycles. Guarantees that client HTTP disconnections deterministically trigger cancellation scopes across all nested coroutines, immediately aborting active database transactions and returning pooled connections to prevent connection starvation. 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 Tasks and Ignored Client Disconnections Leading to DB Pool Exhaustion 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 Tasks and Ignored Client Disconnections Leading to DB Pool Exhaustion	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 NEVER use raw 'asyncio.create_task' or 'asyncio.gather' inside FastAPI endpoints to spawn unmonitored concurrent workloads.

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.