FastAPI AnyIO Structured Concurrency & Deterministic Cancellation Guardrail

--- description: FastAPI AnyIO Structured Concurrency & Deterministic Cancellation Guardrail - Enforces AnyIO structured concurrency task groups and deterministic cancellation cascades across all asynchronous FastAPI endpoints. Guarantees that client disconnects immediately abort nested async I/O workloads and execute shielded cleanups without leaving orphaned database transactions or zombie worker tasks. 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-cancellation-propagation-guardrail" --- # FastAPI AnyIO Structured Concurrency & Deterministic Cancellation Guardrail > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Swallowing cancellation exceptions with broad try/except blocks and spawning unmanaged asyncio.create_task background jobs within request lifecycles. ## System Constraints You are a Principal Backend Architect enforcing deterministic cancellation cascades and structured concurrency in FastAPI services using AnyIO. Client disconnects must cleanly terminate in-flight child tasks and safely clean up resources without leaking coroutines or stranding database locks. ## Strict Guidelines - Engineers MUST ALWAYS manage concurrent sub-tasks within request lifecycles using AnyIO TaskGroups (`anyio.create_task_group()`) rather than unstructured primitives like `asyncio.create_task` or `asyncio.gather`. - Engineers MUST NOT swallow `anyio.get_cancelled_exc_class()` or `asyncio.CancelledError` inside broad `except Exception:` blocks; cancellation exceptions must always be allowed to propagate upwards. - Code MUST wrap critical cleanup routines (such as transaction rollbacks, lock releases, and connection check-ins) in `with anyio.CancelScope(shield=True):` to prevent abortion during in-flight teardown. - Services MUST monitor `request.is_disconnected()` in long-polling or Server-Sent Events (SSE) loops and break explicitly to trigger the ASGI cancellation cascade. - Components NEVER start detached fire-and-forget tasks using database sessions bound to the incoming request context; decoupled background processing MUST be dispatched to a dedicated background worker queue (e.g., Celery, ARQ, SAQ). ## Bad vs Good Implementations ### Unstructured fire-and-forget tasks with bare exception handling vs. AnyIO TaskGroup with shielded cleanup #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Spawning detached tasks via `asyncio.create_task` bypasses structured concurrency: when a client disconnects, Uvicorn cancels the main endpoint coroutine, but the detached audit log continues running on an invalidated session. Furthermore, if cancellation triggers during cleanup, an unshielded `session.rollback()` is immediately cancelled, stranding locks in the database connection pool. ```python import asyncio from fastapi import APIRouter, Depends from sqlalchemy.ext.asyncio import AsyncSession from app.db import get_session router = APIRouter() async def write_audit_log(session: AsyncSession, payload: dict): session.add(AuditLog(**payload)) await session.commit() @router.post("/process") async def process_data(payload: dict, session: AsyncSession = Depends(get_session)): try: # ANTI-PATTERN: Fire-and-forget task detaches from client disconnect cascade # If client disconnects, task continues running on a closed/recycled session asyncio.create_task(write_audit_log(session, payload)) result = await perform_heavy_query(session, payload) return {"status": "ok", "data": result} except Exception as exc: # ANTI-PATTERN: Catches CancelledError on Python <3.8 or swallowed if BaseException caught # Fails to shield cleanup: rollback will be aborted if cancellation hits here await session.rollback() return {"status": "error", "detail": str(exc)} ``` #### Production Standard (Correct) > **Standard**: Using `anyio.create_task_group()` guarantees that if the client disconnects, the parent task cancellation instantly cancels `audit_subtask` and `perform_heavy_query`. The cancellation exception is explicitly allowed to re-raise, while `anyio.CancelScope(shield=True)` ensures `session.rollback()` runs to completion without interruption. ```python import anyio from fastapi import APIRouter, Depends, Request from sqlalchemy.ext.asyncio import AsyncSession from app.db import get_session router = APIRouter() async def audit_subtask(session: AsyncSession, payload: dict): session.add(AuditLog(**payload)) await session.flush() @router.post("/process") async def process_data( request: Request, payload: dict, session: AsyncSession = Depends(get_session) ): try: # INVARIANT: TaskGroup binds child lifetimes strictly to the request lifecycle async with anyio.create_task_group() as tg: tg.start_soon(audit_subtask, session, payload) result = await perform_heavy_query(session, payload) await session.commit() return {"status": "ok", "data": result} except anyio.get_cancelled_exc_class(): # INVARIANT: Shield teardown so cancellation cannot abort the rollback with anyio.CancelScope(shield=True): await session.rollback() raise except Exception: with anyio.CancelScope(shield=True): await session.rollback() raise ```

Architectural Specification & Invariant Rationale

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

FastAPI AnyIO Structured Concurrency & Deterministic Cancellation Guardrail defines an industrial-grade behavioral standard for backend engineering. Enforces AnyIO structured concurrency task groups and deterministic cancellation cascades across all asynchronous FastAPI endpoints. Guarantees that client disconnects immediately abort nested async I/O workloads and execute shielded cleanups without leaving orphaned database transactions or zombie worker tasks. 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: Swallowing cancellation exceptions with broad try/except blocks and spawning unmanaged asyncio.create_task background jobs within request lifecycles. 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	Swallowing cancellation exceptions with broad try/except blocks and spawning unmanaged asyncio.create_task background jobs within request lifecycles.	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 concurrent sub-tasks within request lifecycles using AnyIO TaskGroups (`anyio.create_task_group()`) rather than unstructured primitives like `asyncio.create_task` or `asyncio.gather`.

Engineers MUST ALWAYS manage concurrent sub-tasks within request lifecycles using AnyIO TaskGroups (`anyio.create_task_group()`) rather than unstructured primitives like `asyncio.create_task` or `asyncio.gather`.

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.