FastAPI Blocking Worker Threadpool Isolation & ContextVar Telemetry Propagation

--- description: FastAPI Blocking Worker Threadpool Isolation & ContextVar Telemetry Propagation - Enforces strict isolation of blocking synchronous dependencies onto dedicated, capacity-bounded worker pools to prevent event loop and default AnyIO threadpool starvation. Guarantees deterministic propagation of contextvars (e.g., OpenTelemetry spans, tenant IDs, correlation IDs) across asyncio-to-thread boundaries without state leaking or dropping. globs: **/api/**/*.py, **/routers/**/*.py, **/services/**/*.py, **/dependencies/**/*.py, **/core/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-threadpool-starvation-contextvar-propagation" --- # FastAPI Blocking Worker Threadpool Isolation & ContextVar Telemetry Propagation > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Executing blocking synchronous operations inside async def handlers or offloading to raw executors that drop ContextVar state and exhaust global worker threadpools. ## System Constraints You are a Principal Software Architect enforcing production-grade concurrency invariants in FastAPI. You must eliminate event-loop blocking and prevent default threadpool starvation while guaranteeing distributed trace and context integrity across asynchronous and synchronous boundaries. Reject any synchronous I/O or CPU-heavy task executed directly inside an `async def` route. Ensure all threadpool offloading explicitly propagates Python `contextvars` and uses isolated, bounded capacity limiters or custom executors instead of depleting the global worker threadpool. ## Strict Guidelines - Engineers MUST NEVER execute blocking synchronous I/O (e.g., `requests`, `time.sleep`, synchronous DB drivers, synchronous boto3) directly within an `async def` endpoint or dependency. - Engineers MUST NOT rely on unconstrained default AnyIO/Starlette threadpools (`fastapi.concurrency.run_in_threadpool` or implicit `def` route delegation) for high-latency or high-throughput external blocking services, which risk cascading global threadpool exhaustion. - Services MUST isolate distinct blocking domain operations (e.g., report generation, legacy API integrations, heavy cryptographic tasks) into dedicated, capacity-limited worker threadpools managed via `anyio.CapacityLimiter` or bounded `concurrent.futures.ThreadPoolExecutor` instances. - Engineers MUST ALWAYS propagate `contextvars.copy_context()` when delegating synchronous tasks to standard `ThreadPoolExecutor` or `asyncio.loop.run_in_executor`, ensuring OpenTelemetry trace context and request-scoped state are preserved. - Threaded tasks MUST NEVER mutate outer ContextVar states in a manner that creates race conditions or unhandled thread-local side effects; executions MUST run inside an isolated snapshot via `context.run(...)`. ## Bad vs Good Implementations ### Blocking I/O in Async Handler and ContextVar Dropping #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Invoking `time.sleep` and `requests.get` directly within an `async def` endpoint halts the single asyncio event loop thread, preventing all other concurrent requests from being serviced and causing severe latency spikes. Additionally, un-instrumented synchronous tasks lack threadpool isolation and context propagation. ```python import time import requests from fastapi import APIRouter, Depends from app.core.telemetry import correlation_id_var router = APIRouter() @router.get("/reports/{report_id}") async def get_report(report_id: str): # CRITICAL BUG 1: Blocking call freezes the entire asyncio event loop for this worker. time.sleep(2) # CRITICAL BUG 2: Synchronous HTTP request halts all concurrent async request handling. response = requests.get(f"https://legacy-billing.internal/reports/{report_id}") # Reading contextvar directly, but nested threads or unmanaged subroutines lose it corr_id = correlation_id_var.get() return {"report_id": report_id, "data": response.json(), "correlation_id": corr_id} ``` #### Production Standard (Correct) > **Standard**: The blocking operation is offloaded to a non-blocking thread via `anyio.to_thread.run_sync`, isolating concurrent thread consumption using a bounded `CapacityLimiter`. AnyIO automatically handles `ContextVar` propagation under the hood, ensuring trace context persists without starving the main event loop. ```python import contextvars from functools import partial from anyio import to_thread, CapacityLimiter from fastapi import APIRouter, HTTPException, status from app.core.telemetry import correlation_id_var router = APIRouter() # Dedicated capacity limiter isolates legacy billing traffic from default thread pool LEGACY_BILLING_LIMITER = CapacityLimiter(10) def _blocking_fetch_report(report_id: str, corr_id: str) -> dict: import time import requests # Context is explicitly bound; logging and tracing retain active correlation ID time.sleep(2) response = requests.get( f"https://legacy-billing.internal/reports/{report_id}", headers={"X-Correlation-ID": corr_id}, timeout=10.0 ) if response.status_code != 200: raise ValueError(f"External call failed with status {response.status_code}") return response.json() @router.get("/reports/{report_id}") async def get_report(report_id: str): corr_id = correlation_id_var.get("unassigned") # AnyIO preserves active ContextVars and bounds concurrency with dedicated limiter try: report_data = await to_thread.run_sync( partial(_blocking_fetch_report, report_id, corr_id), limiter=LEGACY_BILLING_LIMITER ) except ValueError as exc: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) return {"report_id": report_id, "data": report_data, "correlation_id": corr_id} ``` ### Custom ThreadPoolExecutor with ContextVar Propagation #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: `asyncio.loop.run_in_executor` does not automatically propagate `contextvars.Context` when used with a custom `ThreadPoolExecutor`, resulting in lost trace contexts, missing correlation IDs, and uninitialized telemetry spans in worker threads. ```python import asyncio from concurrent.futures import ThreadPoolExecutor from fastapi import APIRouter from app.core.telemetry import request_context_var router = APIRouter() CUSTOM_EXECUTOR = ThreadPoolExecutor(max_workers=5) def sync_worker(data: str) -> str: # FAILS: request_context_var.get() raises LookupError or returns default value ctx = request_context_var.get() return f"Processed {data} with context {ctx}" @router.post("/process") async def process_data(payload: str): loop = asyncio.get_running_loop() # BUG: run_in_executor does NOT copy or propagate contextvars in standard executors result = await loop.run_in_executor(CUSTOM_EXECUTOR, sync_worker, payload) return {"result": result} ``` #### Production Standard (Correct) > **Standard**: Uses `contextvars.copy_context().run(...)` within the worker thread invocation, guaranteeing that telemetry contexts, span tags, and request-scoped state are cloned into the worker thread without mutation leakage. ```python import asyncio import contextvars from concurrent.futures import ThreadPoolExecutor from typing import Callable, Any, TypeVar from fastapi import APIRouter from app.core.telemetry import request_context_var T = TypeVar("T") router = APIRouter() CUSTOM_EXECUTOR = ThreadPoolExecutor(max_workers=5) def run_with_context(func: Callable[..., T], *args: Any, **kwargs: Any) -> T: """Execute a sync callable within a snapshot of the parent ContextVars.""" ctx = contextvars.copy_context() return ctx.run(func, *args, **kwargs) def sync_worker(data: str) -> str: # ContextVar is safely retrieved inside the propagated thread context ctx_value = request_context_var.get() return f"Processed {data} with context {ctx_value}" @router.post("/process") async def process_data(payload: str): loop = asyncio.get_running_loop() # Invariant: Wrap callable with run_with_context to guarantee propagation result = await loop.run_in_executor( CUSTOM_EXECUTOR, run_with_context, sync_worker, payload ) return {"result": result} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Blocking Worker Threadpool Isolation & ContextVar Telemetry Propagation

FastAPI Blocking Worker Threadpool Isolation & ContextVar Telemetry Propagation defines an industrial-grade behavioral standard for backend engineering. Enforces strict isolation of blocking synchronous dependencies onto dedicated, capacity-bounded worker pools to prevent event loop and default AnyIO threadpool starvation. Guarantees deterministic propagation of contextvars (e.g., OpenTelemetry spans, tenant IDs, correlation IDs) across asyncio-to-thread boundaries without state leaking or dropping. 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: Executing blocking synchronous operations inside async def handlers or offloading to raw executors that drop ContextVar state and exhaust global worker threadpools. 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	Executing blocking synchronous operations inside async def handlers or offloading to raw executors that drop ContextVar state and exhaust global worker threadpools.	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 execute blocking synchronous I/O (e.g., `requests`, `time.sleep`, synchronous DB drivers, synchronous boto3) directly within an `async def` endpoint or dependency.

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.