FastAPI Async Generator SSE Lifespan and Disconnection Invariant

--- description: FastAPI Async Generator SSE Lifespan and Disconnection Invariant - Enforces proactive ASGI client disconnection polling and rigorous task cancellation cleanup in FastAPI Server-Sent Events (SSE) generators. Eliminates silent async connection pool exhaustion caused by orphaned generator tasks continuing to stream after abrupt client termination. globs: app/**/*.py, src/**/*.py, api/**/*.py, routers/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-sse-disconnection-lifespan-invariant" --- # FastAPI Async Generator SSE Lifespan and Disconnection Invariant > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Zombie Async Generators Silently Draining Connection Pools Post-Client Disconnect ## System Constraints You are a Principal Backend Architect enforcing ASGI lifecycle discipline and resource bounds in FastAPI. Mandate explicit client disconnection polling via `request.is_disconnected()` and deterministic `try...finally` teardown for all async generator streams to safeguard connection pools and event subscriptions. ## Strict Guidelines - Engineers MUST ALWAYS inject the ASGI `Request` dependency into streaming endpoints and explicitly poll `await request.is_disconnected()` on every generator iteration. - Engineers MUST wrap all async generator lifecycles in `try...finally` blocks to guarantee immediate closure of database sessions, Redis pub/sub consumers, or message brokers when a stream terminates. - Engineers MUST NOT inject long-lived, per-request transactional dependencies (such as active `AsyncSession` instances) directly into the route parameters of infinite streaming generators. - Code SHOULD explicitly handle `asyncio.CancelledError` within stream tasks, re-raising it only after releasing leased resources back to their respective connection pools. - Services MUST emit standard SSE headers (`Cache-Control: no-cache`, `Connection: keep-alive`, `X-Accel-Buffering: no`) to prevent intermediate proxy buffering from masking client disconnections. ## Bad vs Good Implementations ### FastAPI SSE Disconnection and Resource Teardown Invariant #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: The endpoint injects a single transactional DB session that remains open indefinitely. Because the generator does not poll `request.is_disconnected()` or monitor the ASGI receive channel, a client closing its browser tab leaves an orphaned async task executing in the background, starving the SQLAlchemy async connection pool. ```python from fastapi import APIRouter, Depends from fastapi.responses import StreamingResponse from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select import asyncio from app.db import get_db from app.models import TelemetryEvent router = APIRouter() @router.get("/telemetry/stream") async def stream_telemetry(db: AsyncSession = Depends(get_db)): async def event_publisher(): while True: # ANTI-PATTERN: Indefinitely holds the route-injected DB session. # If the client aborts, Starlette's StreamingResponse doesn't cancel this # generator automatically without receive polling, leaking the connection. stmt = select(TelemetryEvent).order_by(TelemetryEvent.created_at.desc()).limit(1) result = await db.execute(stmt) event = result.scalar_one_or_none() if event: yield f"data: {event.payload}\n\n" await asyncio.sleep(1) return StreamingResponse(event_publisher(), media_type="text/event-stream") ``` #### Production Standard (Correct) > **Standard**: The endpoint accepts the raw `Request` and uses a session factory to lease connections only for the duration of the query. It routinely checks `await request.is_disconnected()`, catches `asyncio.CancelledError`, and executes a `finally` block, ensuring no lingering tasks or leaked pool connections remain upon disconnect. ```python from fastapi import APIRouter, Depends, Request from fastapi.responses import StreamingResponse from sqlalchemy.ext.asyncio import async_sessionmaker from sqlalchemy import select import asyncio import logging from app.db import get_session_factory from app.models import TelemetryEvent logger = logging.getLogger(__name__) router = APIRouter() @router.get("/telemetry/stream") async def stream_telemetry( request: Request, session_factory: async_sessionmaker = Depends(get_session_factory) ): async def event_publisher(): try: while True: # INVARIANT: Check ASGI disconnection state prior to resource work if await request.is_disconnected(): logger.info("Client disconnected from SSE stream; breaking generator loop.") break # Acquire connection scoped strictly to the data retrieval cycle async with session_factory() as session: stmt = select(TelemetryEvent).order_by(TelemetryEvent.created_at.desc()).limit(1) result = await session.execute(stmt) event = result.scalar_one_or_none() if event: yield f"data: {event.payload}\n\n" await asyncio.sleep(1) except asyncio.CancelledError: logger.info("Streaming task cancelled by ASGI server runtime.") raise finally: logger.info("Generator lifecycle finished; all streaming resources released.") return StreamingResponse( event_publisher(), 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 Generator SSE Lifespan and Disconnection Invariant

FastAPI Async Generator SSE Lifespan and Disconnection Invariant defines an industrial-grade behavioral standard for backend engineering. Enforces proactive ASGI client disconnection polling and rigorous task cancellation cleanup in FastAPI Server-Sent Events (SSE) generators. Eliminates silent async connection pool exhaustion caused by orphaned generator tasks continuing to stream after abrupt client termination. 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 Async Generators Silently Draining Connection Pools Post-Client Disconnect 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 Async Generators Silently Draining Connection Pools Post-Client Disconnect	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 inject the ASGI `Request` dependency into streaming endpoints and explicitly poll `await request.is_disconnected()` on every generator iteration.

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.