FastAPI Async SSE Client Disconnect & Backpressure Propagation Standard

--- description: FastAPI Async SSE Client Disconnect & Backpressure Propagation Standard - Mandates explicit client disconnect detection and active task cancellation within FastAPI Server-Sent Events (SSE) and StreamingResponse pipelines. Eliminates orphan upstream LLM generation loops, unmetered token depletion, and dangling TCP/HTTP/2 kernel buffer handles when downstream clients drop connections. globs: **/api/**/*.py, **/routers/**/*.py, **/services/*stream*.py, **/endpoints/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-sse-client-disconnect-backpressure-pipeline" --- # FastAPI Async SSE Client Disconnect & Backpressure Propagation Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Zombie Token Generation and Socket Buffer Leakage via Unmonitored Async Generators ## System Constraints You are a Principal Engineer enforcing production invariants for FastAPI Async Server-Sent Events (SSE) and streaming HTTP/2 response pipelines. Mandate that every streaming generator actively monitors client connectivity via request.is_disconnected(), guarantees immediate upstream task cancellation on premature disconnection, and enforces clean resource teardown with async context managers. ## Strict Guidelines - Engineers MUST ALWAYS pass the Starlette `Request` object into streaming response handlers and actively evaluate `await request.is_disconnected()` inside generator iteration loops or race the generator against disconnect events. - Engineers MUST NOT stream from raw, unmonitored async generators where downstream disconnection relies purely on next-write socket pipe failures (`EPIPE` / `BrokenPipeError`). - Code SHOULD wrap upstream AI client streams (e.g., OpenAI, Anthropic, vLLM) in an `asyncio.Task` or task cancellation race so that client drops instantly abort upstream billing and compute. - Services MUST explicitly close asynchronous client sessions, generators, and upstream response bodies using `try...finally` or `async with` blocks to prevent kernel socket descriptor exhaustion. - Components NEVER yield subsequent chunks after `request.is_disconnected()` returns `True`, and MUST raise `asyncio.CancelledError` or break cleanly to release the ASGI execution context. ## Bad vs Good Implementations ### Unmonitored SSE Generator vs. Disconnect-Aware Backpressure Streamer #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: The endpoint does not inject the HTTP Request, cannot detect client disconnects, and allows the upstream generator to execute to completion regardless of downstream client state, wasting expensive inference tokens and compute. ```python from fastapi import APIRouter from fastapi.responses import StreamingResponse import asyncio router = APIRouter() async def mock_llm_stream(): for i in range(100): await asyncio.sleep(0.5) yield f"data: token_{i}\n\n" @router.get("/chat/stream") async def chat_stream(): # ANTI-PATTERN: If the client disconnects (closes tab/aborts), # the generator runs for the full duration (50s), consuming CPU, # burning upstream model tokens, and leaking server state. return StreamingResponse(mock_llm_stream(), media_type="text/event-stream") ``` #### Production Standard (Correct) > **Standard**: Injects the Starlette Request, actively polls `request.is_disconnected()`, sets mandatory reverse-proxy headers (`X-Accel-Buffering: no`), and ensures guaranteed resource cleanup inside `finally` upon termination. ```python from fastapi import APIRouter, Request from fastapi.responses import StreamingResponse from typing import AsyncGenerator import asyncio import logging router = APIRouter() logger = logging.getLogger(__name__) async def event_publisher(request: Request) -> AsyncGenerator[str, None]: try: for i in range(100): # INVARIANT: Check downstream connection status prior to scheduling work if await request.is_disconnected(): logger.info("Client disconnected. Aborting upstream generation.") break await asyncio.sleep(0.5) # Simulating upstream LLM inference yield f"data: token_{i}\n\n" except asyncio.CancelledError: logger.info("ASGI task cancelled via client disconnect.") raise finally: # INVARIANT: Mandatory cleanup for upstream sessions and sockets logger.info("Executing stream finalizer and releasing handles.") @router.get("/chat/stream") async def chat_stream(request: Request): return StreamingResponse( event_publisher(request), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", } ) ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Async SSE Client Disconnect & Backpressure Propagation Standard

FastAPI Async SSE Client Disconnect & Backpressure Propagation Standard defines an industrial-grade behavioral standard for backend engineering. Mandates explicit client disconnect detection and active task cancellation within FastAPI Server-Sent Events (SSE) and StreamingResponse pipelines. Eliminates orphan upstream LLM generation loops, unmetered token depletion, and dangling TCP/HTTP/2 kernel buffer handles when downstream clients drop connections. 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: Zombie Token Generation and Socket Buffer Leakage via Unmonitored Async Generators 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	Zombie Token Generation and Socket Buffer Leakage via Unmonitored Async Generators	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 Starlette `Request` object into streaming response handlers and actively evaluate `await request.is_disconnected()` inside generator iteration loops or race the generator against disconnect events.

Engineers MUST ALWAYS pass the Starlette `Request` object into streaming response handlers and actively evaluate `await request.is_disconnected()` inside generator iteration loops or race the generator against disconnect events.

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.