FastAPI AnyIO Structured Concurrency TaskGroup Lifecycle Guardrail

--- description: FastAPI AnyIO Structured Concurrency TaskGroup Lifecycle Guardrail - Eliminates orphaned async background tasks, thread pool starvation, and leaked socket descriptors triggered by premature client disconnections. Mandates the use of AnyIO TaskGroup cancellation scopes bound directly to the FastAPI request lifecycle instead of unmonitored asyncio fire-and-forget tasks or unbounded BackgroundTasks. Guarantees deterministic cooperative cancellation cascades, resource deallocation, and shielded critical sections during connection drops. 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-structured-concurrency-lifecycle" --- # FastAPI AnyIO Structured Concurrency TaskGroup Lifecycle Guardrail > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Spawning detached coroutines via asyncio.create_task or unbounded background tasks that outlive client disconnection and leak resources. ## System Constraints You are a Principal Software Architect enforcing structured concurrency invariants in FastAPI applications via AnyIO. You must reject raw asyncio.create_task invocations and unmonitored FastAPI BackgroundTasks that decouple subtask lifecycles from the HTTP request context. Mandate anyio.create_task_group() across all asynchronous concurrent operations to ensure child coroutines cleanly unwind on client disconnection, while strictly isolating transactional teardowns in shielded cancellation scopes. ## Strict Guidelines - Engineers MUST ALWAYS encapsulate concurrent asynchronous operations within an `anyio.create_task_group()` context block tied to the request scope. - Engineers MUST NOT invoke `asyncio.create_task()`, `asyncio.ensure_future()`, or fire-and-forget background threads for operations logically bound to the HTTP transaction. - Handlers MUST NOT suppress or catch-and-discard `anyio.get_cancelled_exc_class()` or `asyncio.CancelledError`; cancellation exceptions must propagate to trigger orderly task group teardowns. - Engineers MUST wrap post-cancellation finalization (e.g., transactional rollback, audit logging, socket cleanups) inside `with anyio.CancelScope(shield=True):` to prevent aborting critical teardown code. - Streaming endpoints and long-polling loops MUST periodically await cooperative cancellation checkpoints or poll `await request.is_disconnected()` prior to initiating expensive compute or database I/O. - Services MUST replace FastAPI's default `BackgroundTasks` with explicit worker queue dispatchers (e.g., Celery, ARQ, SAQ) whenever asynchronous work must deterministically survive client disconnections. ## Bad vs Good Implementations ### Unbounded asyncio.create_task vs Structured AnyIO TaskGroup with Shielded Cleanup #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task detaches child execution from the request context. If the HTTP client disconnects prematurely, run_analytics runs indefinitely as a zombie task. Additionally, generic exception handling fails to handle or propagates cancellation improperly, risking descriptor and connection pool leaks. ```python import asyncio from fastapi import FastAPI, Request, HTTPException from app.services import fetch_metrics, run_analytics, log_transaction app = FastAPI() @app.post("/analytics/process") async def process_data(request: Request): payload = await request.json() # ANTI-PATTERN: Fire-and-forget task decoupled from client lifecycle. # If the client disconnects or times out, this task continues executing in the background, # leaking database connections and memory, while subsequent failures are silent. asyncio.create_task(run_analytics(payload)) try: metrics = await fetch_metrics(payload["id"]) await log_transaction(payload["id"], status="success") return {"status": "complete", "metrics": metrics} except Exception: # Fails to catch CancelledError properly; swallows critical asyncio control signals raise HTTPException(status_code=500, detail="Processing failed") ``` #### Production Standard (Correct) > **Standard**: Uses anyio.create_task_group to bind child tasks to the request scope. Client disconnection immediately cancels run_analytics and fetch_metrics without orphaned resources. Mandatory post-execution cleanups and audit records run within shielded cancellation scopes so they execute safely even if cancellation is active. ```python import anyio from fastapi import FastAPI, Request, HTTPException from app.services import fetch_metrics, run_analytics, log_transaction app = FastAPI() @app.post("/analytics/process") async def process_data(request: Request): payload = await request.json() metrics = None try: # INVARIANT: TaskGroup binds child coroutine lifecycle to incoming HTTP request. # If client disconnects, ASGI triggers cancellation across all tasks in this group. async with anyio.create_task_group() as tg: tg.start_soon(run_analytics, payload) metrics = await fetch_metrics(payload["id"]) except anyio.get_cancelled_exc_class(): # Cooperative cancellation caught; execute shielded finalization with anyio.CancelScope(shield=True): await log_transaction(payload["id"], status="cancelled") raise except Exception as exc: with anyio.CancelScope(shield=True): await log_transaction(payload["id"], status="failed") raise HTTPException(status_code=500, detail=f"Processing failed: {str(exc)}") from exc with anyio.CancelScope(shield=True): await log_transaction(payload["id"], status="success") return {"status": "complete", "metrics": metrics} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured Concurrency TaskGroup Lifecycle Guardrail

FastAPI AnyIO Structured Concurrency TaskGroup Lifecycle Guardrail defines an industrial-grade behavioral standard for backend engineering. Eliminates orphaned async background tasks, thread pool starvation, and leaked socket descriptors triggered by premature client disconnections. Mandates the use of AnyIO TaskGroup cancellation scopes bound directly to the FastAPI request lifecycle instead of unmonitored asyncio fire-and-forget tasks or unbounded BackgroundTasks. Guarantees deterministic cooperative cancellation cascades, resource deallocation, and shielded critical sections during connection drops. 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 via asyncio.create_task or unbounded background tasks that outlive client disconnection and leak resources. 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 via asyncio.create_task or unbounded background tasks that outlive client disconnection and leak resources.	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 asynchronous operations within an `anyio.create_task_group()` context block tied to the request scope.

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.