FastAPI ASGI Streaming Backpressure & Cooperative Task Cancellation Guardrail

--- description: FastAPI ASGI Streaming Backpressure & Cooperative Task Cancellation Guardrail - Enforces strict cooperative cancellation and backpressure propagation between ASGI HTTP client disconnect events and underlying async generator pipelines. Prevents catastrophic worker thread starvation, uncollected generator frames, and phantom compute leaks by polling request disconnect status and handling asyncio.CancelledError during chunk yielding. globs: app/**/*.py, src/**/*.py, services/**/*.py, routers/**/*.py, api/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-asgi-streaming-backpressure-cancellation-guardrail" --- # FastAPI ASGI Streaming Backpressure & Cooperative Task Cancellation Guardrail > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Silent Async Generator Execution Post-Client Disconnect Leading to Worker Thread Pool Starvation ## System Constraints You are a Principal Backend Systems Architect enforcing rock-solid ASGI streaming backpressure and task cancellation invariants in FastAPI. You must mandate that every async generator piped into a StreamingResponse continuously checks request.is_disconnected() or intercepts asyncio.CancelledError to terminate upstream compute, database cursors, and network sockets immediately upon client disconnect. ## Strict Guidelines - Engineers MUST ALWAYS pass the raw `fastapi.Request` object into generator dependencies that feed `StreamingResponse` instances. - Engineers MUST NOT yield tokens or chunks within an unmonitored `while` or `for` loop without actively probing `await request.is_disconnected()` or handling `asyncio.CancelledError`. - Code SHOULD encapsulate streaming logic inside `try...finally` or asynchronous context managers to ensure immediate resource cleanup (e.g., closing database cursors, terminating LLM client connections, releasing locks) upon client abort. - Services MUST NEVER offload streaming generator iteration to unshielded background threads via `anyio.to_thread.run_sync` without an explicit cooperative cancellation token or timeout. - Components NEVER swallow `asyncio.CancelledError` or catch generic `BaseException` without re-raising `asyncio.CancelledError` to allow the ASGI server event loop to tear down the task hierarchy cleanly. ## Bad vs Good Implementations ### Cooperative Client Disconnect Check and Generator Cancellation in FastAPI StreamingResponse #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: The generator does not accept or poll the `Request` object. When a client aborts the connection, FastAPI and Uvicorn catch the broken socket downstream, but the generator continues iterating in the background until completion, draining CPU and memory. ```python from fastapi import APIRouter from fastapi.responses import StreamingResponse import asyncio router = APIRouter() async def simulate_slow_heavy_stream(): for i in range(1000): # ANTI-PATTERN: Client disconnect is ignored. Loop continues executing # heavy compute and consuming memory even if the client closes the connection. await asyncio.sleep(0.5) yield f"data: chunk_{i}\n\n" @router.get("/events") async def stream_events(): return StreamingResponse(simulate_slow_heavy_stream(), media_type="text/event-stream") ``` #### Production Standard (Correct) > **Standard**: The generator accepts the FastAPI `Request` object and proactively monitors `await request.is_disconnected()`. It propagates `asyncio.CancelledError` and enforces cleanup within a `finally` block, ensuring no leaked background tasks. ```python from fastapi import APIRouter, Request from fastapi.responses import StreamingResponse import asyncio import logging logger = logging.getLogger(__name__) router = APIRouter() async def simulate_guarded_stream(request: Request): try: for i in range(1000): # Check disconnect state before performing expensive compute/fetch if await request.is_disconnected(): logger.info("Client disconnected; aborting generator pipeline") break await asyncio.sleep(0.5) yield f"data: chunk_{i}\n\n" except asyncio.CancelledError: logger.info("Streaming task cancelled by ASGI server runtime") raise finally: # Guaranteed cleanup of downstream handles logger.info("Releasing streaming resources and closing handles") @router.get("/events") async def stream_events(request: Request): return StreamingResponse( simulate_guarded_stream(request), media_type="text/event-stream" ) ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI ASGI Streaming Backpressure & Cooperative Task Cancellation Guardrail

FastAPI ASGI Streaming Backpressure & Cooperative Task Cancellation Guardrail defines an industrial-grade behavioral standard for backend engineering. Enforces strict cooperative cancellation and backpressure propagation between ASGI HTTP client disconnect events and underlying async generator pipelines. Prevents catastrophic worker thread starvation, uncollected generator frames, and phantom compute leaks by polling request disconnect status and handling asyncio.CancelledError during chunk yielding. 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: Silent Async Generator Execution Post-Client Disconnect Leading to Worker Thread Pool Starvation 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	Silent Async Generator Execution Post-Client Disconnect Leading to Worker Thread Pool Starvation	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 raw `fastapi.Request` object into generator dependencies that feed `StreamingResponse` instances.

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.