FastAPI Lifespan AnyIO TaskGroup Cancellation Invariant

--- description: FastAPI Lifespan AnyIO TaskGroup Cancellation Invariant - Enforces that all persistent connection pools, background consumer coroutines, and polling loops in FastAPI applications are strictly bound to an AnyIO TaskGroup governed by the application's lifespan context. This prevents orphaned background tasks and hung worker threads when receiving SIGTERM signals during Kubernetes pod eviction and graceful shutdown sequences. globs: **/lifespan.py, **/main.py, **/app.py, **/dependencies.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-lifespan-anyio-taskgroup-cancellation-invariant" --- # FastAPI Lifespan AnyIO TaskGroup Cancellation Invariant > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Orphaning Unbounded Background Coroutines via asyncio.create_task Outside Lifespan TaskGroups ## System Constraints You are a Principal Software Architect enforcing the FastAPI Lifespan AnyIO TaskGroup Cancellation Invariant. You must reject any fire-and-forget background tasks scheduled via asyncio.create_task inside lifespan handlers, startup hooks, or dependency injections that lack bounded cancellation scopes. Mandate the usage of asynccontextmanager with anyio.create_task_group() to supervise worker lifecycles, ensuring explicit structured concurrency across application teardown. ## Strict Guidelines - Engineers MUST ALWAYS scope asynchronous background loops and persistent consumers inside an 'anyio.create_task_group()' managed within the '@asynccontextmanager' lifespan. - Engineers MUST NOT invoke 'asyncio.create_task()' or 'loop.create_task()' without storing a reference and explicitly cancelling and awaiting the resulting task prior to the lifespan context exit. - Code MUST handle 'anyio.get_cancelled_exc_class()' or 'asyncio.CancelledError' gracefully within long-running workers to ensure prompt release of network sockets, database connections, and file handles. - Services MUST pass active task group references via 'app.state' or explicit yield contexts rather than storing them in global unbound variables. - Lifespan handlers MUST cancel all spawned worker coroutines using 'tg.cancel_scope.cancel()' when exiting the yield block if the workers are intended to terminate cleanly on SIGTERM. ## Bad vs Good Implementations ### Lifespan Background Coroutine Management via AnyIO TaskGroup #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task inside the lifespan without structured supervision leaves the event_consumer coroutine running unbound. When uvicorn initiates a graceful shutdown following a Kubernetes SIGTERM, the orphaned task prevents the loop from closing cleanly or abruptly terminates with unhandled CancelledError mid-execution, leaking connections. ```python import asyncio from contextlib import asynccontextmanager from fastapi import FastAPI async def event_consumer(): while True: await asyncio.sleep(1) @asynccontextmanager async def lifespan(app: FastAPI): # ANTI-PATTERN: Fire-and-forget task without bound scope. # Kubernetes SIGTERM will cancel the lifespan, but this coroutine # remains orphaned or causes uvicorn to hang indefinitely. asyncio.create_task(event_consumer()) yield app = FastAPI(lifespan=lifespan) ``` #### Production Standard (Correct) > **Standard**: The AnyIO TaskGroup establishes a structured concurrency perimeter. Starting the task with tg.start ensures synchronization before traffic arrives. Upon pod termination, the finally block explicitly cancels the TaskGroup scope, awaiting worker unwinding and ensuring resource cleanup before uvicorn releases process execution. ```python from contextlib import asynccontextmanager from fastapi import FastAPI import anyio from anyio.abc import TaskStatus async def event_consumer(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED) -> None: task_status.started() try: while True: await anyio.sleep(1) except anyio.get_cancelled_exc_class(): # Invariant: Graceful cleanup executed upon task group cancellation await anyio.to_thread.run_sync(print, "Flushing event buffer...") raise @asynccontextmanager async def lifespan(app: FastAPI): async with anyio.create_task_group() as tg: # Ensure consumer starts before accepting traffic await tg.start(event_consumer) app.state.task_group = tg try: yield finally: # Invariant: Explicit teardown of all scoped background coroutines tg.cancel_scope.cancel() app = FastAPI(lifespan=lifespan) ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Lifespan AnyIO TaskGroup Cancellation Invariant

FastAPI Lifespan AnyIO TaskGroup Cancellation Invariant defines an industrial-grade behavioral standard for backend engineering. Enforces that all persistent connection pools, background consumer coroutines, and polling loops in FastAPI applications are strictly bound to an AnyIO TaskGroup governed by the application's lifespan context. This prevents orphaned background tasks and hung worker threads when receiving SIGTERM signals during Kubernetes pod eviction and graceful shutdown sequences. 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: Orphaning Unbounded Background Coroutines via asyncio.create_task Outside Lifespan TaskGroups 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	Orphaning Unbounded Background Coroutines via asyncio.create_task Outside Lifespan TaskGroups	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 scope asynchronous background loops and persistent consumers inside an 'anyio.create_task_group()' managed within the '@asynccontextmanager' lifespan.

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.