FastAPI AnyIO SSE TaskGroup Structured Concurrency & Socket Disconnect Guardrail

--- description: FastAPI AnyIO SSE TaskGroup Structured Concurrency & Socket Disconnect Guardrail - Enforces strict structured concurrency using AnyIO task groups and cancel scopes for all FastAPI Server-Sent Events (SSE) and long-lived streaming endpoints. Mandates bidirectional client-disconnect listening and immediate cancellation of upstream database cursors, LLM inference streams, and background workers upon socket termination to prevent resource exhaustion and connection leaks. globs: **/api/**/*.py, **/routers/**/*.py, **/endpoints/**/*.py, **/streaming/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-sse-taskgroup-cancellation-invariant" --- # FastAPI AnyIO SSE TaskGroup Structured Concurrency & Socket Disconnect Guardrail > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Orphaned Streaming Tasks and Unbounded Database Cursor Leaks on Silent SSE Disconnects ## System Constraints You are a Principal Software Architect enforcing production-grade structured concurrency invariants in Python/FastAPI backends. All Server-Sent Events (SSE) and streaming generators MUST be bound to AnyIO TaskGroups or CancelScopes monitored against `request.is_disconnected()`. Unbounded generators, fire-and-forget background tasks during streaming, and untracked database/RPC iterators are critical architecture violations that cause thread/connection starvation. ## Strict Guidelines - Engineers MUST ALWAYS monitor `request.is_disconnected()` in an independent, concurrent task within an `anyio.create_task_group()` to signal downstream cancel scopes immediately upon client disconnect. - Engineers MUST NOT use unshielded infinite generators or native `asyncio.create_task()` inside SSE endpoints without explicit lifecycle ownership tied to the streaming HTTP connection. - Streaming responses MUST wrap external downstream resources (database cursors, async Redis pub/sub iterators, LLM API client sessions) in deterministic `try...finally` or asynchronous context managers to release pool leases upon cancellation. - Services MUST yield raw string or EventSource-compliant packets using an intermediary memory object stream (`anyio.create_memory_object_stream()`) rather than executing blocking work inside the generator body. - Components NEVER assume `StreamingResponse` will automatically terminate deep background coroutines if the client terminates the TCP socket without an active AnyIO cancel scope cancellation trigger. ## Bad vs Good Implementations ### FastAPI SSE Generator Resource Leak vs. AnyIO TaskGroup Structured Concurrency #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: When an SSE client drops the socket (e.g., closing browser tab), `StreamingResponse` only halts when the generator attempts to yield to a broken pipe. Heavy upstream processing, database cursors, and network calls continue executing in the background, exhausting database connection pools and CPU threads. ```python import asyncio from fastapi import APIRouter, Request from fastapi.responses import StreamingResponse from db import get_db_session router = APIRouter() @router.get("/events") async def stream_events(request: Request): # ANTI-PATTERN: If client disconnects, cursor and loop run indefinitely until next yield async def event_generator(): async with get_db_session() as session: cursor = await session.stream_large_dataset() async for row in cursor: # No active check for socket termination during heavy processing await asyncio.sleep(1) yield f"data: {row.json()}\n\n" return StreamingResponse(event_generator(), media_type="text/event-stream") ``` #### Production Standard (Correct) > **Standard**: Uses an AnyIO TaskGroup with an explicit memory object stream. A dedicated task monitors `request.is_disconnected()` and cancels the scope immediately upon TCP drop, instantly halting the producer task and deterministic context managers to release database cursors and pool connections. ```python import anyio from fastapi import APIRouter, Request from fastapi.responses import StreamingResponse from db import get_db_session router = APIRouter() @router.get("/events") async def stream_events(request: Request): send_channel, receive_channel = anyio.create_memory_object_stream(max_buffer_size=10) async def monitor_disconnect(cancel_scope: anyio.CancelScope): while not cancel_scope.cancel_called: if await request.is_disconnected(): cancel_scope.cancel() break await anyio.sleep(0.5) async def produce_events(cancel_scope: anyio.CancelScope): async with send_channel: try: async with get_db_session() as session: async with session.stream_large_dataset() as cursor: async for row in cursor: if cancel_scope.cancel_called: break await send_channel.send(f"data: {row.json()}\n\n") await anyio.sleep(1) except anyio.get_cancelled_exc_class(): # Downstream async database cursor instantly closed upon cancellation raise async def stream_orchestrator(): async with anyio.create_task_group() as tg: tg.start_soon(monitor_disconnect, tg.cancel_scope) tg.start_soon(produce_events, tg.cancel_scope) async with receive_channel: async for message in receive_channel: yield message tg.cancel_scope.cancel() return StreamingResponse(stream_orchestrator(), media_type="text/event-stream") ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO SSE TaskGroup Structured Concurrency & Socket Disconnect Guardrail

FastAPI AnyIO SSE TaskGroup Structured Concurrency & Socket Disconnect Guardrail defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency using AnyIO task groups and cancel scopes for all FastAPI Server-Sent Events (SSE) and long-lived streaming endpoints. Mandates bidirectional client-disconnect listening and immediate cancellation of upstream database cursors, LLM inference streams, and background workers upon socket termination to prevent resource exhaustion and connection leaks. 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: Orphaned Streaming Tasks and Unbounded Database Cursor Leaks on Silent SSE 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	Orphaned Streaming Tasks and Unbounded Database Cursor Leaks on Silent SSE 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 monitor `request.is_disconnected()` in an independent, concurrent task within an `anyio.create_task_group()` to signal downstream cancel scopes immediately upon client disconnect.

Engineers MUST ALWAYS monitor `request.is_disconnected()` in an independent, concurrent task within an `anyio.create_task_group()` to signal downstream cancel scopes immediately upon client disconnect.

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.