FastAPI Lifespan TaskGroup ExceptionGroup Drain Invariant

--- description: FastAPI Lifespan TaskGroup ExceptionGroup Drain Invariant - Guarantees that background workers, message consumers, and persistent coroutines managed within FastAPI lifespan contexts utilize Python 3.12+ asyncio.TaskGroup with graceful cancellation and comprehensive ExceptionGroup unwinding. Prevents silent coroutine abandonment, zombie connection leaks, and SIGTERM worker deadlocks during graceful shutdown. Enforces clean context drain before the ASGI application server finalizes process termination. globs: **/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-lifespan-taskgroup-drain-invariant" --- # FastAPI Lifespan TaskGroup ExceptionGroup Drain Invariant > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Fire-and-Forget asyncio.create_task Inside FastAPI Lifespan Without TaskGroup Drain or ExceptionGroup Handlers ## System Constraints You are a Principal Software Architect enforcing production-grade concurrency invariants in FastAPI applications running Python 3.12+. You strictly prohibit fire-and-forget background tasks launched via asyncio.create_task within application lifespan hooks without deterministic cancellation and drain protocols. Mandate the use of structured concurrency via asyncio.TaskGroup, deterministic SIGTERM/SIGINT teardown via asyncio.Event flags, and rigorous ExceptionGroup (PEP 654) unwinding using 'except*' syntax to guarantee all background tasks terminate cleanly before ASGI worker teardown. ## Strict Guidelines - Engineers MUST ALWAYS encapsulate concurrent background tasks managed during the application lifespan within an explicit asyncio.TaskGroup or an equivalent structured concurrency supervisor. - Engineers MUST ALWAYS signal background workers to stop using a coordinated asyncio.Event or threading.Event prior to exiting the lifespan context manager. - Engineers MUST ALWAYS handle task failures during shutdown using Python 3.12+ 'except* ExceptionGroup' syntax to inspect and log individual sub-exceptions without masking termination signals. - Engineers MUST NOT invoke raw 'asyncio.create_task()' inside the lifespan generator without retaining strong task references and explicit awaitable cancellation barriers in the post-yield cleanup phase. - Services MUST enforce a deterministic timeout via 'asyncio.timeout()' around the lifespan task drain to prevent hung external I/O operations from blocking ASGI server worker cycling. ## Bad vs Good Implementations ### FastAPI Lifespan Background Task Drain with TaskGroup and except* #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Launching background tasks using untracked asyncio.create_task() violates structured concurrency. The task reference can be prematurely garbage collected, unhandled task exceptions escape unmonitored until loop garbage collection, and ASGI server shutdown (e.g., Uvicorn/Gunicorn SIGTERM) will terminate the process before the task can finalize inflight I/O or flush database buffers. ```python import asyncio from contextlib import asynccontextmanager from fastapi import FastAPI async def event_consumer(): while True: await asyncio.sleep(1) # If this raises an unhandled error, it is silenced or crashes the loop arbitrarily @asynccontextmanager async def lifespan(app: FastAPI): # Anti-pattern: Detached coroutine, no reference retention, no structured drain asyncio.create_task(event_consumer()) yield # Teardown exits immediately while event_consumer is abruptly killed or left dangling app = FastAPI(lifespan=lifespan) ``` #### Production Standard (Correct) > **Standard**: Leverages Python 3.12+ asyncio.TaskGroup and except* syntax. Background tasks are coordinated via a shared asyncio.Event shutdown flag, guaranteed a bounded drain window via asyncio.timeout(), and cleanly collected upon context exit. Any compound failures are surfaced as typed ExceptionGroups, preventing zombie tasks and ensuring deterministic ASGI server process exits. ```python import asyncio import logging from contextlib import asynccontextmanager from fastapi import FastAPI logger = logging.getLogger(__name__) async def event_consumer(shutdown_event: asyncio.Event) -> None: try: while not shutdown_event.is_set(): await asyncio.sleep(0.5) except asyncio.CancelledError: logger.info("Consumer received cancellation, flushing local buffers.") raise @asynccontextmanager async def lifespan(app: FastAPI): shutdown_event = asyncio.Event() # Launch supervisor in a controlled TaskGroup try: async with asyncio.TaskGroup() as tg: consumer_task = tg.create_task(event_consumer(shutdown_event)) yield # Teardown sequence: signal stop and cancel if unresponsive shutdown_event.set() try: async with asyncio.timeout(5.0): await asyncio.sleep(0) except* TimeoutError: logger.warning("Drain timed out; cancelling consumer task.") consumer_task.cancel() except* asyncio.CancelledError: logger.info("All lifespan tasks cancelled cleanly.") except* Exception as eg: logger.error("Unhandled exception(s) during lifespan execution: %s", eg.exceptions) raise app = FastAPI(lifespan=lifespan) ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Lifespan TaskGroup ExceptionGroup Drain Invariant

FastAPI Lifespan TaskGroup ExceptionGroup Drain Invariant defines an industrial-grade behavioral standard for backend engineering. Guarantees that background workers, message consumers, and persistent coroutines managed within FastAPI lifespan contexts utilize Python 3.12+ asyncio.TaskGroup with graceful cancellation and comprehensive ExceptionGroup unwinding. Prevents silent coroutine abandonment, zombie connection leaks, and SIGTERM worker deadlocks during graceful shutdown. Enforces clean context drain before the ASGI application server finalizes process 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: Fire-and-Forget asyncio.create_task Inside FastAPI Lifespan Without TaskGroup Drain or ExceptionGroup Handlers 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	Fire-and-Forget asyncio.create_task Inside FastAPI Lifespan Without TaskGroup Drain or ExceptionGroup Handlers	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 encapsulate concurrent background tasks managed during the application lifespan within an explicit asyncio.TaskGroup or an equivalent structured concurrency supervisor.

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.