FastAPI Lifespan-Managed Async Task Pipeline and Drain Invariant

--- description: FastAPI Lifespan-Managed Async Task Pipeline and Drain Invariant - Enforces deterministic, structured async worker lifecycles managed strictly through FastAPI lifespan context managers. Prevents orphaned background tasks, silent unhandled task exceptions, and thread/database pool exhaustion during client disconnection cascades by prohibiting ad-hoc asyncio.create_task and legacy Starlette BackgroundTasks for unbounded processing. globs: **/*.py, app/**/*.py, src/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-lifespan-task-supervisor-invariant" --- # FastAPI Lifespan-Managed Async Task Pipeline and Drain Invariant > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Ad-hoc asyncio.create_task spawning and unmonitored BackgroundTasks leaking resources on client disconnect ## System Constraints You are a Principal Backend Architect enforcing structured concurrency and lifespan-bounded task management in FastAPI. You must reject any fire-and-forget task execution using raw asyncio.create_task or starlette.background.BackgroundTasks for critical work. Require explicitly bounded asyncio.Queue pipelines supervised within the @asynccontextmanager lifespan, complete with drain semantics, cancellation shielding on critical paths, and graceful worker shutdown to protect DB pools. ## Strict Guidelines - Engineers MUST ALWAYS manage worker loops inside a top-level `@asynccontextmanager` FastAPI lifespan using `asyncio.TaskGroup` (Python 3.11+) or an explicit supervisor task list. - Engineers MUST NOT invoke `asyncio.create_task` or `FastAPI.BackgroundTasks` inside route handlers for I/O operations or database writes; jobs must be dispatched via bounded `asyncio.Queue` instances. - Services MUST implement explicit queue draining and worker task cancellation timeouts during lifespan shutdown to prevent database pool exhaustion and stuck processes. - Engineers MUST attach `add_done_callback` error-logging hooks to any standalone supervisor tasks to ensure unhandled worker exceptions fail loudly instead of being silently swallowed. - Route endpoints MUST handle client disconnection gracefully using `request.is_disconnected()` checks before enqueueing high-cost pipeline payloads. ## Bad vs Good Implementations ### Orphaned create_task vs. Lifespan Supervised Queue Pipeline #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Spawning raw asyncio.create_task inside request handlers causes orphaned execution on client disconnect or server shutdown. If the server receives SIGTERM, in-flight tasks are forcefully killed mid-transaction, leaving database connections leaked or data partially committed. ```python import asyncio from fastapi import FastAPI, Request app = FastAPI() async def process_audit_log(payload: dict): # Anti-Pattern: No bounded concurrency, no shutdown drain. # Client disconnect or SIGTERM leaves connection leaks in pool. await asyncio.sleep(2) print(f"Processed: {payload}") @app.post("/events") async def ingest_event(payload: dict): # CRITICAL BUG: Fire-and-forget task without lifespan tracking asyncio.create_task(process_audit_log(payload)) return {"status": "accepted"} ``` #### Production Standard (Correct) > **Standard**: Tasks are bounded by an in-memory queue initiated and torn down within FastAPI's lifespan. During SIGTERM, pending tasks are cleanly joined, cancellation is safely caught, client disconnections prior to queue push are respected, and backpressure is enforced via QueueFull checks. ```python import asyncio from contextlib import asynccontextmanager from typing import AsyncGenerator from fastapi import FastAPI, Request, HTTPException, status class AuditWorkerPipeline: def __init__(self, maxsize: int = 1000): self.queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=maxsize) self.worker_task: asyncio.Task | None = None async def _worker_loop(self): while True: item = await self.queue.get() try: # Explicit task execution bounded by pipeline lifecycle await asyncio.sleep(0.1) except asyncio.CancelledError: break except Exception as exc: # Catch-all to prevent unhandled worker death print(f"Worker pipeline error: {exc}") finally: self.queue.task_done() def start(self): self.worker_task = asyncio.create_task(self._worker_loop()) async def stop(self): # Graceful drain and structured cancellation await self.queue.join() if self.worker_task: self.worker_task.cancel() await asyncio.gather(self.worker_task, return_exceptions=True) pipeline = AuditWorkerPipeline() @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: pipeline.start() yield await pipeline.stop() app = FastAPI(lifespan=lifespan) @app.post("/events", status_code=status.HTTP_202_ACCEPTED) async def ingest_event(payload: dict, request: Request): if await request.is_disconnected(): raise HTTPException(status_code=499, detail="Client Disconnected") try: pipeline.queue.put_nowait(payload) except asyncio.QueueFull: raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Worker queue saturated") return {"status": "queued"} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Lifespan-Managed Async Task Pipeline and Drain Invariant

FastAPI Lifespan-Managed Async Task Pipeline and Drain Invariant defines an industrial-grade behavioral standard for backend engineering. Enforces deterministic, structured async worker lifecycles managed strictly through FastAPI lifespan context managers. Prevents orphaned background tasks, silent unhandled task exceptions, and thread/database pool exhaustion during client disconnection cascades by prohibiting ad-hoc asyncio.create_task and legacy Starlette BackgroundTasks for unbounded processing. 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: Ad-hoc asyncio.create_task spawning and unmonitored BackgroundTasks leaking resources on 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	Ad-hoc asyncio.create_task spawning and unmonitored BackgroundTasks leaking resources on 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 manage worker loops inside a top-level `@asynccontextmanager` FastAPI lifespan using `asyncio.TaskGroup` (Python 3.11+) or an explicit supervisor task list.

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.