FastAPI Async Generator SSE Disconnect Detection and Upstream Backpressure Invariant

--- description: FastAPI Async Generator SSE Disconnect Detection and Upstream Backpressure Invariant - Requires all FastAPI Server-Sent Events (SSE) and StreamingResponse async generators to actively poll ASGI disconnect state via `request.is_disconnected()` or monitor underlying transport signals. Failing to detect disconnects causes orphaned upstream database/API subscriptions, memory leaks, and worker event-loop starvation when clients drop connections midway. globs: app/**/*.py, src/**/*.py, api/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-sse-client-disconnect-backpressure" --- # FastAPI Async Generator SSE Disconnect Detection and Upstream Backpressure Invariant > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Orphaned Infinite Async Generators Consuming Upstream Streams After Client HTTP Disconnection ## System Constraints You are a Principal Backend Architect enforcing robust ASGI disconnect detection and backpressure propagation for FastAPI and Starlette streaming endpoints. You mandate that any long-running or continuous async generator serving SSE or chunked responses must actively check for client disconnects, clean up upstream iterators/queues upon termination, and never assume Starlette will abort downstream generators automatically without an intervening yield. ## Strict Guidelines - Engineers MUST ALWAYS pass the `fastapi.Request` object into streaming generator functions to enable ASGI disconnect polling. - Engineers MUST ALWAYS check `await request.is_disconnected()` on every iteration or within bounded polling intervals during long waits. - Code SHOULD utilize an `asyncio.TaskGroup` or race `asyncio.wait` between the upstream message consumer and `request.receive()` when upstream generation is push-based or blocking. - Services MUST clean up upstream subscriptions, database cursors, and message queues in a `finally` block or context manager when the generator is closed or cancelled. - Engineers MUST NOT rely exclusively on `yield` throwing `GeneratorExit` or `asyncio.CancelledError`, as network buffers or buffering proxies (e.g., NGINX) can delay generator write errors indefinitely. ## Bad vs Good Implementations ### SSE Async Generator Disconnect Polling & Resource Cleanup #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: The generator runs indefinitely without inspecting the ASGI disconnect state. If a client closes their browser tab, the async task continues fetching expensive upstream telemetry and consuming event loop ticks. ```python import asyncio from fastapi import FastAPI from fastapi.responses import StreamingResponse app = FastAPI() async def unbounded_event_stream(): # ANTI-PATTERN: Ignores client connection state. # If client drops, the generator keeps consuming from upstream broker indefinitely # until a TCP write buffer fills and triggers a broken pipe. while True: data = await fetch_expensive_upstream_telemetry() yield f"data: {data}\n\n" await asyncio.sleep(1) @app.get("/stream") async def stream_telemetry(): return StreamingResponse(unbounded_event_stream(), media_type="text/event-stream") ``` #### Production Standard (Correct) > **Standard**: The generator receives the ASGI Request, proactively polls `request.is_disconnected()`, yields heartbeat keep-alives to detect dead sockets, handles `asyncio.CancelledError`, and cleanly closes upstream resources in a `finally` block. ```python import asyncio import logging from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse app = FastAPI() logger = logging.getLogger(__name__) async def bounded_event_stream(request: Request): upstream_subscription = await subscribe_upstream_telemetry() try: while True: # INVARIANT: Check ASGI disconnect status before scheduling upstream operations if await request.is_disconnected(): logger.info("Client disconnected; aborting SSE generator upstream ingestion.") break try: # Enforce timeout or race with disconnect to prevent infinite hangs on upstream data = await asyncio.wait_for(upstream_subscription.next_event(), timeout=1.0) yield f"data: {data}\n\n" except asyncio.TimeoutError: # Send SSE keep-alive comment to flush pipeline and trigger disconnect if broken yield ": keepalive\n\n" except asyncio.CancelledError: logger.info("Streaming task cancelled by ASGI server.") raise finally: # Clean up upstream resource deterministically await upstream_subscription.unsubscribe() @app.get("/stream") async def stream_telemetry(request: Request): return StreamingResponse( bounded_event_stream(request), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "Connection": "keep-alive"} ) ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Async Generator SSE Disconnect Detection and Upstream Backpressure Invariant

FastAPI Async Generator SSE Disconnect Detection and Upstream Backpressure Invariant defines an industrial-grade behavioral standard for backend engineering. Requires all FastAPI Server-Sent Events (SSE) and StreamingResponse async generators to actively poll ASGI disconnect state via `request.is_disconnected()` or monitor underlying transport signals. Failing to detect disconnects causes orphaned upstream database/API subscriptions, memory leaks, and worker event-loop starvation when clients drop connections midway. 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 Infinite Async Generators Consuming Upstream Streams After Client HTTP Disconnection 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 Infinite Async Generators Consuming Upstream Streams After Client HTTP Disconnection	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 pass the `fastapi.Request` object into streaming generator functions to enable ASGI disconnect polling.

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.