FastAPI AnyIO Structured TaskGroup Exception Barrier Standard

--- description: FastAPI AnyIO Structured TaskGroup Exception Barrier Standard - Enforces strict structured concurrency in FastAPI services by replacing unbounded fire-and-forget background tasks with AnyIO TaskGroups. This eliminates silent async task failures, guarantees deterministic cancellation propagation, and prevents socket descriptor and connection pool leaks under peak concurrency. Every spawned concurrent task must be explicitly bounded to a managed lifecycle scope with an aggregate exception barrier. globs: app/**/*.py, src/**/*.py, services/**/*.py, api/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-taskgroup-exception-barrier" --- # FastAPI AnyIO Structured TaskGroup Exception Barrier Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Spawning detached coroutines with asyncio.create_task or unbounded Starlette BackgroundTasks that drop unhandled exceptions and leak network sockets ## System Constraints You are a Principal Software Architect enforcing zero-leak structured concurrency across FastAPI asynchronous services. You must forbid detached coroutines (such as asyncio.create_task or unchecked BackgroundTasks) and mandate AnyIO TaskGroups with explicit ExceptionGroup barriers. Ensure deterministic task lifecycles, graceful teardown, and non-leaking connection management under all failure modes. ## Strict Guidelines - Engineers MUST ALWAYS bound concurrent operations within an 'anyio.create_task_group()' context manager or an explicitly managed service nursery. - Engineers MUST NOT invoke 'asyncio.create_task()' or 'asyncio.ensure_future()' directly within FastAPI request scopes. - Engineers MUST catch multi-task failures using Python 3.11+ 'except* Exception' or AnyIO's exception hierarchy to prevent unhandled exception suppression. - Services MUST guarantee that all child tasks are cancelled and awaited before exiting the parent scope, ensuring open socket descriptors and database sessions are finalized. - Code SHOULD inject dependencies and database sessions per task or pass explicitly bound transactional context rather than sharing mutable session instances across task group boundaries. - Components NEVER allow client disconnects to orphan dangling long-running workers without active cancellation tokens or timeout guards. ## Bad vs Good Implementations ### Bounded TaskGroup with Exception Barrier vs Detached Fire-and-Forget Coroutine #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task in request handlers creates detached coroutines that escape FastAPI's exception boundaries. If downstream network failures occur, the task fails silently or pollutes stderr without structured logging, leaving network sockets in CLOSE_WAIT and exhausting file descriptors under load. ```python import asyncio import httpx from fastapi import FastAPI, APIRouter router = APIRouter() async def push_audit_metric(event_id: str, payload: dict) -> None: async with httpx.AsyncClient() as client: # If this times out or fails, the exception is swallowed or printed as an unhandled warning await client.post("https://metrics.internal/v1/emit", json={"id": event_id, "data": payload}) @router.post("/items/{item_id}") async def create_item(item_id: str, payload: dict): # CRITICAL FLAW: Fire-and-forget task. Unhandled exceptions are silent, # and high concurrency leaks sockets when downstream slows or crashes. asyncio.create_task(push_audit_metric(item_id, payload)) return {"status": "accepted", "item_id": item_id} ``` #### Production Standard (Correct) > **Standard**: By utilizing 'anyio.create_task_group()' combined with 'except*', all child coroutines are explicitly supervised. Sockets are deterministically closed on exit, timeouts propagate cancellation immediately, and errors cannot escape silently or leak uncollected system resources. ```python import logging import anyio import httpx from fastapi import FastAPI, APIRouter, HTTPException, status logger = logging.getLogger(__name__) router = APIRouter() async def push_audit_metric(client: httpx.AsyncClient, event_id: str, payload: dict) -> None: with anyio.fail_after(2.0): response = await client.post("https://metrics.internal/v1/emit", json={"id": event_id, "data": payload}) response.raise_for_status() @router.post("/items/{item_id}") async def create_item(item_id: str, payload: dict): # Structured Concurrency: Task lifecycle is strictly bounded by the task group. try: async with httpx.AsyncClient() as client: async with anyio.create_task_group() as tg: tg.start_soon(push_audit_metric, client, item_id, payload) # Additional concurrent bounded tasks can be spawned here except* httpx.HTTPError as eg: logger.error("Telemetry delivery failed within bounded task group: %s", eg.exceptions) # Exception is bounded; we can choose to record fallback without leaking sockets except* anyio.TimeoutError: logger.warning("Telemetry dispatch timed out; child task cancelled deterministically") except* Exception as eg: logger.critical("Fatal unexpected error in task group: %s", eg.exceptions) raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Processing failed") return {"status": "accepted", "item_id": item_id} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured TaskGroup Exception Barrier Standard

FastAPI AnyIO Structured TaskGroup Exception Barrier Standard defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency in FastAPI services by replacing unbounded fire-and-forget background tasks with AnyIO TaskGroups. This eliminates silent async task failures, guarantees deterministic cancellation propagation, and prevents socket descriptor and connection pool leaks under peak concurrency. Every spawned concurrent task must be explicitly bounded to a managed lifecycle scope with an aggregate exception barrier. 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 coroutines with asyncio.create_task or unbounded Starlette BackgroundTasks that drop unhandled exceptions and leak network sockets 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 coroutines with asyncio.create_task or unbounded Starlette BackgroundTasks that drop unhandled exceptions and leak network sockets	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 bound concurrent operations within an 'anyio.create_task_group()' context manager or an explicitly managed service nursery.

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.