fix: FileSystemProvider reload race condition (#3938)

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

src/fastmcp/server/providers/filesystem.py

@@ -28,8 +28,9 @@ Example:
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 from pathlib import Path
+from typing import Any
 
 from fastmcp.prompts.base import Prompt
 from fastmcp.resources.base import Resource
@@ -96,16 +97,20 @@ class FileSystemProvider(LocalProvider):
         self._warned_files: dict[Path, float] = {}
         # Lock for serializing reload operations (created lazily)
         self._reload_lock: asyncio.Lock | None = None
+        # Generation counter to deduplicate concurrent reloads
+        self._reload_generation: int = 0
 
         # Always load once at init to catch errors early
         self._load_components()
 
     def _load_components(self) -> None:
         """Discover and register all components from the filesystem."""
-        # Clear existing components if reloading
         if self._loaded:
             self._components.clear()
 
+        if not self._root.exists():
+            logger.warning("FileSystemProvider root does not exist: %s", self._root)
+
         result = discover_and_import(self._root)
 
         # Log warnings for failed files (only once per file version)
@@ -154,73 +159,67 @@ class FileSystemProvider(LocalProvider):
         else:
             logger.debug("Ignoring unknown component type: %r", type(component))
 
-    async def _ensure_loaded(self) -> None:
-        """Ensure components are loaded, reloading if in reload mode.
+    async def _with_reload(self, coro_fn: Callable[..., Any], *args: Any) -> Any:
+        """Acquire the reload lock, reload if needed, then run *coro_fn*.
 
-        Uses a lock to serialize concurrent reload operations and runs
-        filesystem I/O off the event loop using asyncio.to_thread.
+        Holding the lock across both the reload and the read prevents
+        concurrent readers from seeing a partially-rebuilt ``_components``
+        dict (the ``clear()`` + re-register window).
+
+        A generation counter deduplicates concurrent reload requests:
+        if another caller already reloaded while we waited for the lock,
+        we skip the redundant reload.
         """
         if not self._reload and self._loaded:
-            return
+            return await coro_fn(*args)
 
         # Create lock lazily (can't create in __init__ without event loop)
         if self._reload_lock is None:
             self._reload_lock = asyncio.Lock()
 
+        generation_before = self._reload_generation
+
         async with self._reload_lock:
-            # Double-check after acquiring lock
-            if self._reload or not self._loaded:
+            if not self._loaded or (
+                self._reload and self._reload_generation == generation_before
+            ):
                 await asyncio.to_thread(self._load_components)
+                self._reload_generation += 1
+            return await coro_fn(*args)
 
     # Override provider methods to support reload mode
 
     async def _list_tools(self) -> Sequence[Tool]:
-        """Return all tools, reloading if in reload mode."""
-        await self._ensure_loaded()
-        return await super()._list_tools()
+        return await self._with_reload(super()._list_tools)
 
     async def _get_tool(
         self, name: str, version: VersionSpec | None = None
     ) -> Tool | None:
-        """Get a tool by name, reloading if in reload mode."""
-        await self._ensure_loaded()
-        return await super()._get_tool(name, version)
+        return await self._with_reload(super()._get_tool, name, version)
 
     async def _list_resources(self) -> Sequence[Resource]:
-        """Return all resources, reloading if in reload mode."""
-        await self._ensure_loaded()
-        return await super()._list_resources()
+        return await self._with_reload(super()._list_resources)
 
     async def _get_resource(
         self, uri: str, version: VersionSpec | None = None
     ) -> Resource | None:
-        """Get a resource by URI, reloading if in reload mode."""
-        await self._ensure_loaded()
-        return await super()._get_resource(uri, version)
+        return await self._with_reload(super()._get_resource, uri, version)
 
     async def _list_resource_templates(self) -> Sequence[ResourceTemplate]:
-        """Return all resource templates, reloading if in reload mode."""
-        await self._ensure_loaded()
-        return await super()._list_resource_templates()
+        return await self._with_reload(super()._list_resource_templates)
 
     async def _get_resource_template(
         self, uri: str, version: VersionSpec | None = None
     ) -> ResourceTemplate | None:
-        """Get a resource template, reloading if in reload mode."""
-        await self._ensure_loaded()
-        return await super()._get_resource_template(uri, version)
+        return await self._with_reload(super()._get_resource_template, uri, version)
 
     async def _list_prompts(self) -> Sequence[Prompt]:
-        """Return all prompts, reloading if in reload mode."""
-        await self._ensure_loaded()
-        return await super()._list_prompts()
+        return await self._with_reload(super()._list_prompts)
 
     async def _get_prompt(
         self, name: str, version: VersionSpec | None = None
     ) -> Prompt | None:
-        """Get a prompt by name, reloading if in reload mode."""
-        await self._ensure_loaded()
-        return await super()._get_prompt(name, version)
+        return await self._with_reload(super()._get_prompt, name, version)
 
     def __repr__(self) -> str:
         return f"FileSystemProvider(root={self._root!r}, reload={self._reload})"

tests/fs/test_provider.py

@@ -1,5 +1,6 @@
 """Tests for FileSystemProvider."""
 
+import asyncio
 import time
 from pathlib import Path
 
@@ -183,7 +184,7 @@ def original() -> str:
         assert provider._loaded
         assert len(provider._components) == 1
 
-        # Add another file - should be picked up on next _ensure_loaded
+        # Add another file - should be picked up on next read
         (tmp_path / "tool2.py").write_text(
             """\
 from fastmcp.tools import tool
@@ -194,9 +195,9 @@ def added() -> str:
 """
         )
 
-        # With reload=True, _ensure_loaded re-scans
-        await provider._ensure_loaded()
-        assert len(provider._components) == 2
+        # With reload=True, reading triggers a re-scan
+        tools = await provider._list_tools()
+        assert len(tools) == 2
 
     async def test_warning_deduplication_same_file(self, tmp_path: Path, capsys):
         """Warnings for the same broken file should not repeat."""
@@ -211,7 +212,7 @@ def added() -> str:
         assert "WARNING" in captured.err and "Failed to import" in captured.err
 
         # Second load (same file, unchanged) - should NOT warn again
-        await provider._ensure_loaded()
+        await provider._list_tools()
         captured = capsys.readouterr()
         assert "Failed to import" not in captured.err
 
@@ -232,7 +233,7 @@ def added() -> str:
         bad_file.write_text("syntax error here !!!")
 
         # Next load - should warn again (file changed)
-        await provider._ensure_loaded()
+        await provider._list_tools()
         captured = capsys.readouterr()
         # Check for warning indicator (rich may truncate long paths)
         assert "WARNING" in captured.err and "Failed to import" in captured.err
@@ -262,7 +263,7 @@ def my_tool() -> str:
         )
 
         # Load again - should NOT warn, file is fixed
-        await provider._ensure_loaded()
+        await provider._list_tools()
         captured = capsys.readouterr()
         assert "Failed to import" not in captured.err
         assert len(provider._components) == 1
@@ -272,12 +273,50 @@ def my_tool() -> str:
         bad_file.write_text("1/0  # broken again")
 
         # Should warn again
-        await provider._ensure_loaded()
+        await provider._list_tools()
         captured = capsys.readouterr()
         # Check for warning indicator (rich may truncate long paths)
         assert "WARNING" in captured.err and "Failed to import" in captured.err
 
 
+class TestFileSystemProviderReloadRace:
+    """Test that concurrent readers don't see empty components during reload."""
+
+    async def test_concurrent_reader_never_sees_empty(self, tmp_path: Path):
+        """A reader during reload should see either old or new components, never empty."""
+        (tmp_path / "tool.py").write_text(
+            """\
+from fastmcp.tools import tool
+
+@tool
+def my_tool() -> str:
+    return "hello"
+"""
+        )
+
+        provider = FileSystemProvider(tmp_path, reload=True)
+        assert len(await provider._list_tools()) == 1
+
+        observed_empty = False
+
+        async def reader():
+            nonlocal observed_empty
+            for _ in range(20):
+                tools = await provider._list_tools()
+                if len(tools) == 0:
+                    observed_empty = True
+                await asyncio.sleep(0)
+
+        async def reloader():
+            for _ in range(5):
+                provider._loaded = False
+                await provider._list_tools()  # triggers reload
+                await asyncio.sleep(0)
+
+        await asyncio.gather(reader(), reloader())
+        assert not observed_empty, "Reader saw empty components during reload"
+
+
 class TestFileSystemProviderIntegration:
     """Integration tests with FastMCP server."""