FastAPI AnyIO Structured Concurrency Session Poison Shielding

--- description: FastAPI AnyIO Structured Concurrency Session Poison Shielding - Enforces strict structured concurrency boundaries around pooled database sessions when executing concurrent child tasks via AnyIO TaskGroups or asyncio.gather in FastAPI. Prevents unhandled task cancellations and exceptions from stranding uncommitted, dirty, or transactionally poisoned connections in shared connection pools (such as SQLAlchemy AsyncEngine or asyncpg). Mandates explicit error boundaries, shielded rollbacks, and per-task isolated transactional units. globs: app/**/*.py, src/**/*.py, services/**/*.py, api/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-taskgroup-poison-shielding" --- # FastAPI AnyIO Structured Concurrency Session Poison Shielding > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Sharing a Single Database Session Across AnyIO TaskGroups Without Shielded Teardown and Isolation ## System Constraints You are a Principal Software Architect enforcing production-grade structured concurrency and resource integrity in FastAPI services using AnyIO and asynchronous database clients (SQLAlchemy AsyncSession, asyncpg). You must mandate strict transactional isolation, child task shielding, and deterministic session cleanup to prevent poisoned connections from leaking back into shared connection pools upon task failure or cancellation. ## Strict Guidelines - Engineers MUST NEVER pass a single AsyncSession or pooled database connection across multiple concurrent tasks inside an AnyIO create_task_group() or asyncio.gather(). - Each concurrent child task MUST obtain its own distinct AsyncSession from the sessionmaker or pool to prevent race conditions on underlying protocol buffers. - Engineers MUST wrap session cleanup (rollback or close) in anyio.CancelScope(shield=True) within finally blocks to ensure teardown runs even if the parent TaskGroup is cancelled. - Services MUST explicitly call session.rollback() before releasing connections to the pool in exception handlers, preventing uncommitted transactional state from bleeding into subsequent requests. - Components NEVER execute raw transactional queries inside an unhandled TaskGroup without wrapping the group in a top-level try/except/finally resource management context. ## Bad vs Good Implementations ### Poisoned Shared Session in AnyIO TaskGroup vs Shielded Per-Task Sessions #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Sharing a single SQLAlchemy AsyncSession or raw asyncpg connection across multiple tasks inside an AnyIO TaskGroup causes protocol race conditions. When one task raises an exception, the task group cancels all other tasks, leaving the shared transaction uncommitted and unrolled-back. Returning this dirty session to the pool poisons future requests reusing that socket. ```python from fastapi import APIRouter, Depends from sqlalchemy.ext.asyncio import AsyncSession import anyio from app.db import get_session router = APIRouter() async def worker(session: AsyncSession, item_id: int): # Concurrent writes on the same AsyncSession cause protocol corruption await session.execute("UPDATE items SET processed = true WHERE id = :id", {"id": item_id}) if item_id == 13: raise ValueError("Simulated crash") @router.post("/batch-process") async def process_batch(session: AsyncSession = Depends(get_session)): # Anti-pattern: Sharing session across task group leaves uncommitted transaction # on failure and corrupts the connection in the pool. async with anyio.create_task_group() as tg: for item_id in range(20): tg.start_soon(worker, session, item_id) await session.commit() return {"status": "ok"} ``` #### Production Standard (Correct) > **Standard**: Every concurrent child task spawns its own AsyncSession from the factory. AnyIO cancel shielding guarantees that session.rollback() runs even if the parent task group triggers a cancellation cascade, completely isolating failures and preserving connection pool integrity. ```python from fastapi import APIRouter, Depends from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker import anyio from app.db import get_session_factory router = APIRouter() async def isolated_worker(session_factory: async_sessionmaker[AsyncSession], item_id: int): # Each concurrent task manages its own isolated connection lifecycle async with session_factory() as session: try: async with session.begin(): await session.execute("UPDATE items SET processed = true WHERE id = :id", {"id": item_id}) if item_id == 13: raise ValueError("Simulated crash") except Exception: # Shield rollback from AnyIO cancellation to ensure pool hygiene with anyio.CancelScope(shield=True): await session.rollback() raise @router.post("/batch-process") async def process_batch(session_factory: async_sessionmaker[AsyncSession] = Depends(get_session_factory)): async with anyio.create_task_group() as tg: for item_id in range(20): tg.start_soon(isolated_worker, session_factory, item_id) return {"status": "ok"} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured Concurrency Session Poison Shielding

FastAPI AnyIO Structured Concurrency Session Poison Shielding defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency boundaries around pooled database sessions when executing concurrent child tasks via AnyIO TaskGroups or asyncio.gather in FastAPI. Prevents unhandled task cancellations and exceptions from stranding uncommitted, dirty, or transactionally poisoned connections in shared connection pools (such as SQLAlchemy AsyncEngine or asyncpg). Mandates explicit error boundaries, shielded rollbacks, and per-task isolated transactional units. 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: Sharing a Single Database Session Across AnyIO TaskGroups Without Shielded Teardown and Isolation 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	Sharing a Single Database Session Across AnyIO TaskGroups Without Shielded Teardown and Isolation	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 NEVER pass a single AsyncSession or pooled database connection across multiple concurrent tasks inside an AnyIO create_task_group() or asyncio.gather().

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.