diff --git a/fastapi/routing.py b/fastapi/routing.py index c442b122b..a42508e44 100644 --- a/fastapi/routing.py +++ b/fastapi/routing.py @@ -1571,20 +1571,20 @@ class RouteContext: class _IncludedRouter(BaseRoute): original_router: "APIRouter" include_context: _RouterIncludeContext - _effective_candidates: list["_EffectiveRouteContext | _IncludedRouter"] = field( - default_factory=list - ) - _effective_candidates_version: int | None = None - _effective_low_priority_routes: list["_EffectiveRouteContext"] = field( - default_factory=list - ) - _effective_low_priority_routes_version: int | None = None + _effective_candidates_cache: "tuple[int, list[_EffectiveRouteContext | _IncludedRouter]] | None" = None + _effective_low_priority_routes_cache: "tuple[int, list[_EffectiveRouteContext]] | None" = None def effective_candidates(self) -> list["_EffectiveRouteContext | _IncludedRouter"]: routes_version = self.original_router._get_routes_version() - if routes_version == self._effective_candidates_version: - return self._effective_candidates - self._effective_candidates = [] + cached = self._effective_candidates_cache + if cached is not None and cached[0] == routes_version: + return cached[1] + # Build into a local list and only publish it once fully built, + # together with its routes version, as a single atomic attribute + # assignment, so that concurrent callers never observe (or append + # into) a partially built shared cache and a published version always + # matches the list that was built for it. + effective_candidates: list[_EffectiveRouteContext | _IncludedRouter] = [] candidates = self.original_router.routes for route in candidates: if isinstance(route, _IncludedRouter): @@ -1593,23 +1593,29 @@ class _IncludedRouter(BaseRoute): original_router=route.original_router, include_context=child_context, ) - self._effective_candidates.append(child_branch) + 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 + effective_candidates.append(route_context) + self._effective_candidates_cache = (routes_version, effective_candidates) + 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 = [] + cached = self._effective_low_priority_routes_cache + if cached is not None and cached[0] == routes_version: + return cached[1] + # Build into a local list and only publish it once fully built, + # together with its routes version, as a single atomic attribute + # assignment, so that concurrent callers never observe (or append + # into) a partially built shared cache and a published version always + # matches the list that was built for it. + 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: - self._effective_low_priority_routes.append(route_context) + 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) @@ -1617,11 +1623,14 @@ class _IncludedRouter(BaseRoute): original_router=route.original_router, include_context=child_context, ) - self._effective_low_priority_routes.extend( + 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 + self._effective_low_priority_routes_cache = ( + routes_version, + effective_low_priority_routes, + ) + return effective_low_priority_routes def _build_effective_context( self, route: BaseRoute diff --git a/tests/test_router_include_context.py b/tests/test_router_include_context.py index cb8dc81fa..25f0c7e1d 100644 --- a/tests/test_router_include_context.py +++ b/tests/test_router_include_context.py @@ -1,3 +1,6 @@ +import inspect +import sys +import threading from typing import Annotated, cast import pytest @@ -1011,3 +1014,137 @@ async def test_apirouter_handle_fallback_without_include_context(): assert messages[0]["type"] == "http.response.start" assert messages[0]["status"] == 200 assert messages[1]["body"] == b"items" + + +def _hammer_rebuild_from_threads(rebuild, *, n_threads: int = 8) -> list[int]: + barrier = threading.Barrier(n_threads) + sizes: list[int] = [] + lock = threading.Lock() + + def worker() -> None: + barrier.wait() + size = len(rebuild()) + with lock: + sizes.append(size) + + threads = [threading.Thread(target=worker) for _ in range(n_threads)] + old_interval = sys.getswitchinterval() + # Shrink the GIL switch interval so the threads interleave inside the + # sub-millisecond cache rebuild window. + sys.setswitchinterval(1e-6) + try: + for thread in threads: + thread.start() + for thread in threads: + thread.join() + finally: + sys.setswitchinterval(old_interval) + return sizes + + +def test_effective_candidates_concurrent_rebuild_is_thread_safe(): + n_routes = 200 + + def make_endpoint(i: int): + def endpoint(): # pragma: no cover + return {"i": i} + + return endpoint + + app = FastAPI() + router = APIRouter() + for i in range(n_routes): + router.add_api_route(f"/route-{i}", make_endpoint(i), methods=["GET"]) + app.include_router(router) + included = next(r for r in app.router.routes if isinstance(r, _IncludedRouter)) + + for _ in range(5): + included._effective_candidates_cache = None # force a rebuild + sizes = _hammer_rebuild_from_threads(included.effective_candidates) + assert sizes == [n_routes] * len(sizes) + assert len(included.effective_candidates()) == n_routes + + +def test_effective_low_priority_routes_concurrent_rebuild_is_thread_safe(): + n_frontends = 50 + + app = FastAPI() + router = APIRouter() + for i in range(n_frontends): + child = APIRouter() + child.frontend(f"/front-{i}", directory=f"front-{i}", check_dir=False) + router.include_router(child) + app.include_router(router) + included = next(r for r in app.router.routes if isinstance(r, _IncludedRouter)) + + for _ in range(5): + included._effective_low_priority_routes_cache = None # force a rebuild + sizes = _hammer_rebuild_from_threads(included.effective_low_priority_routes) + assert sizes == [n_frontends] * len(sizes) + assert len(included.effective_low_priority_routes()) == n_frontends + + +def test_route_added_during_in_flight_rebuild_is_not_lost(): + # Cross-version race: a rebuild that read the old routes version and then + # publishes its list *after* a rebuild for the new version has already + # published must not hide the new route until the next route mutation. + # Because the cache is published atomically as a (version, list) tuple, the + # stale publication at worst costs one extra rebuild. + def make_endpoint(i: int): + def endpoint(): # pragma: no cover + return {"i": i} + + return endpoint + + app = FastAPI() + router = APIRouter() + for i in range(3): + router.add_api_route(f"/old-{i}", make_endpoint(i), methods=["GET"]) + app.include_router(router) + included = next(r for r in app.router.routes if isinstance(r, _IncludedRouter)) + + rebuild_code = _IncludedRouter.effective_candidates.__code__ + source_lines, first_lineno = inspect.getsourcelines( + _IncludedRouter.effective_candidates + ) + publish_lineno = first_lineno + next( + index + for index, line in enumerate(source_lines) + if "self._effective_candidates_cache = (" in line + ) + + ready = threading.Event() + resume = threading.Event() + + def pause_before_publish(frame, event, arg): # pragma: no cover + if ( + event == "line" + and frame.f_code is rebuild_code + and frame.f_lineno == publish_lineno + ): + sys.settrace(None) # pause only once + ready.set() + resume.wait(timeout=5) + return pause_before_publish + + def stale_rebuild() -> None: # pragma: no cover + sys.settrace(pause_before_publish) + try: + included.effective_candidates() + finally: + sys.settrace(None) + + thread = threading.Thread(target=stale_rebuild) + thread.start() + assert ready.wait(timeout=5), "the rebuild never reached the publish point" + # The in-flight rebuild built its list for the old routes version and is + # paused just before publishing it. Add a route, bumping the version. + router.add_api_route("/new-route", make_endpoint(3), methods=["GET"]) + # A rebuild for the new version completes and publishes the new route. + assert len(included.effective_candidates()) == 4 + # The paused rebuild resumes and publishes its outdated list last. + resume.set() + thread.join() + # The new route must still be served: the stale cache entry carries the old + # version, so the next call rebuilds instead of serving it as current. + assert len(included.effective_candidates()) == 4