🐛 Refactor router route building to make it thread-safe, mainly relevant for tests running in parallel threads (uncommon)

fastapi/routing.py

@@ -7,6 +7,7 @@ import inspect
 import json
 import os
 import stat
+import threading
 import types
 from collections.abc import (
     AsyncIterator,
@@ -1571,6 +1572,9 @@ class RouteContext:
 class _IncludedRouter(BaseRoute):
     original_router: "APIRouter"
     include_context: _RouterIncludeContext
+    _effective_routes_lock: Any = field(
+        default_factory=threading.Lock, repr=False, compare=False
+    )
     _effective_candidates: list["_EffectiveRouteContext | _IncludedRouter"] = field(
         default_factory=list
     )
@@ -1584,44 +1588,53 @@ class _IncludedRouter(BaseRoute):
         routes_version = self.original_router._get_routes_version()
         if routes_version == self._effective_candidates_version:
             return self._effective_candidates
-        self._effective_candidates = []
-        candidates = self.original_router.routes
-        for route in candidates:
-            if isinstance(route, _IncludedRouter):
-                child_context = self.include_context.combine(route.include_context)
-                child_branch = _IncludedRouter(
-                    original_router=route.original_router,
-                    include_context=child_context,
-                )
-                self._effective_candidates.append(child_branch)
-                continue
-            route_context = self._build_effective_context(route)
-            if route_context is not None:
-                self._effective_candidates.append(route_context)
-        self._effective_candidates_version = routes_version
-        return self._effective_candidates
+        with self._effective_routes_lock:
+            routes_version = self.original_router._get_routes_version()
+            if routes_version == self._effective_candidates_version:
+                return self._effective_candidates
+            effective_candidates: list[_EffectiveRouteContext | _IncludedRouter] = []
+            for route in self.original_router.routes:
+                if isinstance(route, _IncludedRouter):
+                    child_context = self.include_context.combine(route.include_context)
+                    child_branch = _IncludedRouter(
+                        original_router=route.original_router,
+                        include_context=child_context,
+                    )
+                    effective_candidates.append(child_branch)
+                    continue
+                route_context = self._build_effective_context(route)
+                if route_context is not None:
+                    effective_candidates.append(route_context)
+            self._effective_candidates = effective_candidates
+            self._effective_candidates_version = routes_version
+            return effective_candidates
 
     def effective_low_priority_routes(self) -> list["_EffectiveRouteContext"]:
         routes_version = self.original_router._get_routes_version()
         if routes_version == self._effective_low_priority_routes_version:
             return self._effective_low_priority_routes
-        self._effective_low_priority_routes = []
-        for route in self.original_router._low_priority_routes:
-            route_context = self._build_effective_context(route)
-            if route_context is not None:
-                self._effective_low_priority_routes.append(route_context)
-        for route in self.original_router.routes:
-            if isinstance(route, _IncludedRouter):
-                child_context = self.include_context.combine(route.include_context)
-                child_branch = _IncludedRouter(
-                    original_router=route.original_router,
-                    include_context=child_context,
-                )
-                self._effective_low_priority_routes.extend(
-                    child_branch.effective_low_priority_routes()
-                )
-        self._effective_low_priority_routes_version = routes_version
-        return self._effective_low_priority_routes
+        with self._effective_routes_lock:
+            routes_version = self.original_router._get_routes_version()
+            if routes_version == self._effective_low_priority_routes_version:
+                return self._effective_low_priority_routes
+            effective_low_priority_routes: list[_EffectiveRouteContext] = []
+            for route in self.original_router._low_priority_routes:
+                route_context = self._build_effective_context(route)
+                if route_context is not None:
+                    effective_low_priority_routes.append(route_context)
+            for route in self.original_router.routes:
+                if isinstance(route, _IncludedRouter):
+                    child_context = self.include_context.combine(route.include_context)
+                    child_branch = _IncludedRouter(
+                        original_router=route.original_router,
+                        include_context=child_context,
+                    )
+                    effective_low_priority_routes.extend(
+                        child_branch.effective_low_priority_routes()
+                    )
+            self._effective_low_priority_routes = effective_low_priority_routes
+            self._effective_low_priority_routes_version = routes_version
+            return effective_low_priority_routes
 
     def _build_effective_context(
         self, route: BaseRoute

tests/test_router_include_context.py

@@ -1,3 +1,5 @@
+import threading
+from collections.abc import Sequence
 from typing import Annotated, cast
 
 import pytest
@@ -898,6 +900,66 @@ async def test_included_unknown_route_is_ignored_and_can_return_default_404():
     assert messages[0]["status"] == 404
 
 
+def test_included_router_candidate_cache_is_thread_safe():
+    router = APIRouter()
+    route_count = 120
+    thread_count = 6
+    for index in range(route_count):
+
+        @router.get(f"/items/{index}")
+        def read_item(index: int = index):
+            return {"index": index}
+
+    app = FastAPI()
+    app.include_router(router)
+    included_router = cast(_IncludedRouter, app.router.routes[-1])
+    barrier = threading.Barrier(thread_count)
+    results: list[Sequence[object]] = []
+
+    def build_candidates() -> None:
+        barrier.wait()
+        results.append(included_router.effective_candidates())
+
+    threads = [threading.Thread(target=build_candidates) for _ in range(thread_count)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert len(results) == thread_count
+    assert all(result is results[0] for result in results)
+    assert len(results[0]) == route_count
+    assert TestClient(app).get("/items/0").json() == {"index": 0}
+
+
+def test_included_router_low_priority_cache_rechecks_version_after_lock(monkeypatch):
+    router = APIRouter()
+    app = FastAPI()
+    app.include_router(router)
+    included_router = cast(_IncludedRouter, app.router.routes[-1])
+    routes_version = router._get_routes_version()
+    version_checked = threading.Event()
+
+    def get_routes_version() -> int:
+        version_checked.set()
+        return routes_version
+
+    monkeypatch.setattr(router, "_get_routes_version", get_routes_version)
+    result: list[Sequence[object]] = []
+
+    def build_low_priority_routes() -> None:
+        result.append(included_router.effective_low_priority_routes())
+
+    with included_router._effective_routes_lock:
+        thread = threading.Thread(target=build_low_priority_routes)
+        thread.start()
+        assert version_checked.wait(timeout=1)
+        included_router._effective_low_priority_routes_version = routes_version
+    thread.join()
+
+    assert result == [[]]
+
+
 def test_no_prefix_include_validation_sees_effective_starlette_route_candidates():
     def endpoint(request):  # pragma: no cover
         return PlainTextResponse("ok")