FastAPI Uvicorn Multiplexed Stream Backpressure Guardrail

--- description: FastAPI Uvicorn Multiplexed Stream Backpressure Guardrail - Enforces strict producer-consumer backpressure mechanisms and bounded concurrency queues for HTTP/2 multiplexed streams and Server-Sent Events (SSE) in FastAPI/Uvicorn deployments. Prevents unconstrained heap ballooning, event-loop starvation, and OOM terminations caused by fast upstream generators paired with slow network consumers. globs: **/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-uvicorn-stream-backpressure-guardrail" --- # FastAPI Uvicorn Multiplexed Stream Backpressure Guardrail > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unbounded Async Generator Consumption and Zero-Backpressure Task Spawning on Multiplexed SSE/Streaming Endpoints ## System Constraints You are a Principal Backend Architect enforcing strict asyncio event-loop backpressure and memory constraints on FastAPI and Uvicorn streaming endpoints. You mandate bounded queues, client-disconnect checks, and flow-controlled chunk emission across all HTTP/2 and SSE streams. ## Strict Guidelines - Engineers MUST ALWAYS enforce bounded buffer sizes on `asyncio.Queue` (e.g., `maxsize=32` to `maxsize=128`) when piping upstream messages to `StreamingResponse` or SSE streams. - Engineers MUST NOT spawn detached `asyncio.create_task` calls per incoming event or client chunk without passing through an explicit concurrency semaphore and bounded memory pool. - Engineers MUST verify client connectivity status before and during heavy stream yields using `await request.is_disconnected()` to terminate abandoned upstream processing immediately. - Streaming generators MUST employ cooperative yields using `await asyncio.sleep(0)` or explicitly await socket write availability to prevent monolithic CPU bursts from starving the single-threaded event loop. - Engineers MUST configure Uvicorn worker limits (`--limit-concurrency`, `--timeout-keep-alive`) and apply global admission control (e.g., token-bucket or semaphore middleware) to prevent multiplexed stream descriptor exhaustion. ## Bad vs Good Implementations ### Multiplexed SSE Producer with Bounded Queue and Disconnect Detection #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: The generator executes without assessing downstream socket flow or client disconnections. If an HTTP/2 client throttles its TCP window or stalls consumption, Uvicorn's internal buffer balloons in the Python heap, leading to memory saturation and eventual OOM crash. ```python import asyncio from fastapi import FastAPI from fastapi.responses import StreamingResponse app = FastAPI() async def unbounded_event_generator(): # ANTI-PATTERN: Infinite unbounded generator pushing chunks without checking client status # or using unconstrained task accumulation, leading to memory bloat when client reads slowly. while True: data = await fetch_large_telemetry_payload() yield f"data: {data}\n\n" # Zero backpressure: if network socket is congested, Uvicorn buffers heap frames infinitely @app.get("/stream") async def stream_data(): return StreamingResponse(unbounded_event_generator(), media_type="text/event-stream") ``` #### Production Standard (Correct) > **Standard**: Implements an explicit bounded queue with `maxsize=64`, proactively monitors `request.is_disconnected()` to reclaim resources immediately on disconnect, bounds active streams via a global `Semaphore`, and ensures tasks are gracefully cancelled in a `finally` block. ```python import asyncio from typing import AsyncGenerator from fastapi import FastAPI, Request, HTTPException, status from fastapi.responses import StreamingResponse app = FastAPI() GLOBAL_STREAM_SEMAPHORE = asyncio.Semaphore(500) # Constrain concurrent active streams async def bounded_event_generator(request: Request) -> AsyncGenerator[str, None]: queue: asyncio.Queue[str] = asyncio.Queue(maxsize=64) # Strict bounded backpressure async def upstream_producer(): try: while not await request.is_disconnected(): item = await fetch_telemetry_batch() # Puts block when consumer/network stalls, preventing heap ballooning await queue.put(f"data: {item}\n\n") await asyncio.sleep(0) # Cooperative event loop yield finally: await queue.put("__DONE__") producer_task = asyncio.create_task(upstream_producer()) try: while True: if await request.is_disconnected(): break msg = await queue.get() if msg == "__DONE__": break yield msg queue.task_done() finally: producer_task.cancel() await asyncio.gather(producer_task, return_exceptions=True) @app.get("/stream") async def stream_data(request: Request): if GLOBAL_STREAM_SEMAPHORE.locked(): raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Max streams exceeded") async def guarded_stream(): async with GLOBAL_STREAM_SEMAPHORE: async for chunk in bounded_event_generator(request): yield chunk return StreamingResponse(guarded_stream(), media_type="text/event-stream") ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Uvicorn Multiplexed Stream Backpressure Guardrail

FastAPI Uvicorn Multiplexed Stream Backpressure Guardrail defines an industrial-grade behavioral standard for backend engineering. Enforces strict producer-consumer backpressure mechanisms and bounded concurrency queues for HTTP/2 multiplexed streams and Server-Sent Events (SSE) in FastAPI/Uvicorn deployments. Prevents unconstrained heap ballooning, event-loop starvation, and OOM terminations caused by fast upstream generators paired with slow network consumers. 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: Unbounded Async Generator Consumption and Zero-Backpressure Task Spawning on Multiplexed SSE/Streaming Endpoints 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	Unbounded Async Generator Consumption and Zero-Backpressure Task Spawning on Multiplexed SSE/Streaming Endpoints	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 enforce bounded buffer sizes on `asyncio.Queue` (e.g., `maxsize=32` to `maxsize=128`) when piping upstream messages to `StreamingResponse` or SSE streams.

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.