FastAPI AnyIO TaskGroup Connection-Drop Shielding & Cleanup Invariant

--- description: FastAPI AnyIO TaskGroup Connection-Drop Shielding & Cleanup Invariant - Enforces AnyIO structured concurrency task groups and shielded cleanup scopes across all streaming and long-lived FastAPI endpoints. Guarantees complete transactional rollbacks, external resource deallocation, and state auditing when clients abruptly terminate HTTP connections, completely preventing unhandled coroutine leaks and orphaned background 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-connection-drop-shielding" --- # FastAPI AnyIO TaskGroup Connection-Drop Shielding & Cleanup Invariant > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unshielded Cleanup Routines and Unstructured asyncio.create_task Leaks on Abrupt Client Disconnects ## System Constraints You are a Principal Backend Architect enforcing strict structured concurrency and transactional safety in FastAPI and AnyIO applications. Client disconnects (ASGI cancellation) must propagate cleanly through structured task groups, while cleanup, rollback, and auditing tasks must execute within shielded AnyIO cancellation scopes without leaving zombie coroutines or hung worker threads. ## Strict Guidelines - Engineers MUST ALWAYS use `anyio.create_task_group()` instead of bare `asyncio.create_task()` or `asyncio.gather()` for spawning concurrent subtasks within endpoint lifecycles. - Engineers MUST wrap all critical resource cleanups, database rollbacks, and audit logging in `with anyio.CancelScope(shield=True):` inside `finally` blocks when handling streaming endpoints or cancelable async workflows. - Engineers MUST NOT swallow `anyio.get_cancelled_exc_class()` or `asyncio.CancelledError`; cancellations must be allowed to propagate to ASGI servers (e.g., Uvicorn) to recycle socket descriptors. - Code SHOULD regularly check `await request.is_disconnected()` in high-latency or long-running streaming generators to trigger early exit before starting expensive processing steps. - Services MUST NEVER fire detached background tasks without an explicit, centrally managed background worker queue (e.g., Celery, ARQ, SAQ) or an application-lifespan-scoped AnyIO task group. - Components NEVER execute raw async database write transactions inside an unshielded cancellation context during connection tear-down. ## Bad vs Good Implementations ### Streaming Generator Connection Teardown and Resource Shielding #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task causes orphaned coroutines that outlive the client request lifecycle. Performing async cleanup in an unshielded 'finally' block causes the ASGI cancellation exception to interrupt the rollback and GPU release mid-flight, leaking hardware resources and corrupting connection pools. ```python import asyncio from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse app = FastAPI() @app.post("/stream/generate") async def stream_generation(request: Request): # ANTI-PATTERN: Spawns fire-and-forget task without structured scope # If the client disconnects, task continues running indefinitely (orphaned coroutine leak) asyncio.create_task(audit_log_start(request)) async def event_generator(): try: for i in range(100): await asyncio.sleep(0.5) yield f"data: chunk_{i}\n\n" finally: # ANTI-PATTERN: When client disconnects, CancelledError is raised immediately # during this await, terminating the DB rollback mid-execution await db_session.rollback() await release_gpu_slot() return StreamingResponse(event_generator(), media_type="text/event-stream") ``` #### Production Standard (Correct) > **Standard**: Using anyio.create_task_group guarantees all auxiliary coroutines are bound to the generator lifecycle. Wrapping cleanup routines within anyio.CancelScope(shield=True) ensures database rollbacks and GPU releases complete fully even when client socket cancellation aborts the primary response stream. ```python import anyio from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse app = FastAPI() @app.post("/stream/generate") async def stream_generation(request: Request): async def event_generator(): # Invariant: Structured concurrency using AnyIO TaskGroup async with anyio.create_task_group() as tg: # Register subtasks within the structured group tg.start_soon(audit_log_start, request.client.host) try: for i in range(100): # Proactive disconnect detection if await request.is_disconnected(): raise anyio.get_cancelled_exc_class()() await anyio.sleep(0.5) yield f"data: chunk_{i}\n\n" finally: # Invariant: Shield critical resource cleanup from ASGI cancellation with anyio.CancelScope(shield=True): await db_session.rollback() await release_gpu_slot() # Structured TaskGroup automatically awaits tg child tasks before exiting return StreamingResponse(event_generator(), media_type="text/event-stream") ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO TaskGroup Connection-Drop Shielding & Cleanup Invariant

FastAPI AnyIO TaskGroup Connection-Drop Shielding & Cleanup Invariant defines an industrial-grade behavioral standard for backend engineering. Enforces AnyIO structured concurrency task groups and shielded cleanup scopes across all streaming and long-lived FastAPI endpoints. Guarantees complete transactional rollbacks, external resource deallocation, and state auditing when clients abruptly terminate HTTP connections, completely preventing unhandled coroutine leaks and orphaned background 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: Unshielded Cleanup Routines and Unstructured asyncio.create_task Leaks on Abrupt Client Disconnects 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	Unshielded Cleanup Routines and Unstructured asyncio.create_task Leaks on Abrupt Client Disconnects	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 use `anyio.create_task_group()` instead of bare `asyncio.create_task()` or `asyncio.gather()` for spawning concurrent subtasks within endpoint lifecycles.

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.