FastAPI AnyIO TaskGroup Structured Concurrency Teardown Invariant

--- description: FastAPI AnyIO TaskGroup Structured Concurrency Teardown Invariant - Enforces strict structured concurrency teardown semantics using anyio.create_task_group() within async FastAPI streaming endpoints and background job runners. Guarantees deterministic cancellation propagation, unhandled exception containment, and immediate file/socket resource cleanup upon client disconnects or upstream failures. globs: app/**/*.py, src/**/*.py, routers/**/*.py, services/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-taskgroup-teardown-invariant" --- # FastAPI AnyIO TaskGroup Structured Concurrency Teardown Invariant > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Fire-and-Forget asyncio.create_task Inside Streaming Generators Causing Orphaned Sockets and Zombie Coroutines ## System Constraints You are a Principal Backend Architect enforcing strict structured concurrency and resource teardown invariants in FastAPI applications powered by AnyIO. You must eliminate all unmanaged fire-and-forget coroutine dispatches (e.g., asyncio.create_task, BackgroundTasks with streaming responses) and ensure every concurrent background stream or sub-coroutine is scoped deterministically within an AnyIO TaskGroup or CancellationScope with guaranteed cleanup blocks. ## Strict Guidelines - Engineers MUST ALWAYS bound concurrent operations and asynchronous worker coroutines inside 'async with anyio.create_task_group() as tg:' or 'anyio.CancelScope()'. - Engineers MUST NOT spawn detached coroutines via 'asyncio.create_task()' or 'asyncio.ensure_future()' within request-response or streaming endpoints. - Streaming endpoints (SSE/WebSocket/Chunked HTTP) MUST monitor 'request.is_disconnected()' and explicitly trigger task group cancellation when disconnects occur. - Code MUST employ 'finally' blocks or asynchronous context managers ('async with') inside the TaskGroup boundary to guarantee immediate socket closure and cleanup. - Components NEVER swallow or suppress 'anyio.get_cancelled_exc_class()' (or 'asyncio.CancelledError') without re-raising, unless performing explicit clean termination in a shielded finalizer scope. ## Bad vs Good Implementations ### FastAPI Streaming Endpoint Concurrency Lifecycle #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task inside or directly adjacent to a streaming generator decouples the background work from the client connection lifecycle. When the client closes the HTTP socket, the background coroutine continues to run in the global loop as a zombie, exhausting system file descriptors and database connection pools. ```python import asyncio from fastapi import FastAPI from fastapi.responses import StreamingResponse app = FastAPI() async def stream_metrics(): # ANTI-PATTERN: Detached task leaks on client disconnect # Network sockets and memory buffers remain orphaned indefinitely asyncio.create_task(run_background_telemetry_collector()) for i in range(100): yield f"data: metric-{i}\n\n" await asyncio.sleep(1) @app.get("/metrics") async def get_metrics(): return StreamingResponse(stream_metrics(), media_type="text/event-stream") ``` #### Production Standard (Correct) > **Standard**: The AnyIO TaskGroup deterministically binds the producer coroutine to the request scope. If the client disconnects, tg.cancel_scope.cancel() signals cancellation to all child tasks. The memory stream closes cleanly, and the TaskGroup context manager guarantees all child tasks have completed teardown before returning. ```python import anyio from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse from collections.abc import AsyncGenerator app = FastAPI() async def telemetry_producer(channel: anyio.streams.memory.MemoryObjectSendStream[str]) -> None: async with channel: while True: await anyio.sleep(1) await channel.send("telemetry_tick") async def stream_generator(request: Request) -> AsyncGenerator[str, None]: send_stream, receive_stream = anyio.create_memory_object_stream[str](max_buffer_size=16) # INVARIANT: TaskGroup strictly encapsulates background work within request lifecycle async with anyio.create_task_group() as tg: tg.start_soon(telemetry_producer, send_stream) async with receive_stream: async for item in receive_stream: if await request.is_disconnected(): tg.cancel_scope.cancel() break yield f"data: {item}\n\n" # Execution blocks until telemetry_producer terminates; zero zombie leaks ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO TaskGroup Structured Concurrency Teardown Invariant

FastAPI AnyIO TaskGroup Structured Concurrency Teardown Invariant defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency teardown semantics using anyio.create_task_group() within async FastAPI streaming endpoints and background job runners. Guarantees deterministic cancellation propagation, unhandled exception containment, and immediate file/socket resource cleanup upon client disconnects or upstream failures. 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: Fire-and-Forget asyncio.create_task Inside Streaming Generators Causing Orphaned Sockets and Zombie Coroutines 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	Fire-and-Forget asyncio.create_task Inside Streaming Generators Causing Orphaned Sockets and Zombie Coroutines	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 and asynchronous worker coroutines inside 'async with anyio.create_task_group() as tg

Engineers MUST ALWAYS bound concurrent operations and asynchronous worker coroutines inside 'async with anyio.create_task_group() as tg:' or 'anyio.CancelScope()'.

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.