fix(api): prevent plugin provider cache stampedes (#37388)

Co-authored-by: VeraPyuyi <204892921+VeraPyuyi@users.noreply.github.com>

api/core/plugin/plugin_service.py

@@ -16,10 +16,11 @@ import logging
 import time
 from collections.abc import Mapping, Sequence
 from mimetypes import guess_type
-from typing import ClassVar
+from typing import Any, ClassVar
 
 from pydantic import BaseModel, TypeAdapter, ValidationError
 from redis import RedisError
+from redis.exceptions import LockError
 from sqlalchemy import delete, select, update
 from sqlalchemy.orm import Session
 from yarl import URL
@@ -82,6 +83,10 @@ class PluginService:
     REDIS_TTL = 60 * 5  # 5 minutes
     PLUGIN_MODEL_PROVIDERS_REDIS_KEY_PREFIX = "plugin_model_providers:tenant_id:"
     PLUGIN_MODEL_PROVIDERS_GENERATION_REDIS_KEY_PREFIX = "plugin_model_providers_generation:tenant_id:"
+    PLUGIN_MODEL_PROVIDERS_LOCK_REDIS_KEY_PREFIX = "plugin_model_providers_refresh_lock:tenant_id:"
+    PLUGIN_MODEL_PROVIDERS_LOCK_TTL = 30
+    PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT = 2.0
+    PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_INTERVAL = 0.05
     PLUGIN_INSTALL_TASK_TERMINAL_STATUSES = (PluginInstallTaskStatus.Success, PluginInstallTaskStatus.Failed)
     # Mirror the detail-panel endpoint query size so list reconciliation and
     # the visible endpoint drawer exercise the same daemon pagination path.
@@ -98,6 +103,10 @@ class PluginService:
     def _get_plugin_model_providers_generation_cache_key(cls, tenant_id: str) -> str:
         return f"{cls.PLUGIN_MODEL_PROVIDERS_GENERATION_REDIS_KEY_PREFIX}{tenant_id}"
 
+    @classmethod
+    def _get_plugin_model_providers_lock_key(cls, tenant_id: str, generation: int) -> str:
+        return f"{cls.PLUGIN_MODEL_PROVIDERS_LOCK_REDIS_KEY_PREFIX}{tenant_id}:generation:{generation}"
+
     @staticmethod
     def _get_provider_short_name_alias(provider: PluginModelProviderEntity) -> str:
         """
@@ -197,32 +206,41 @@ class PluginService:
         cls, tenant_id: str, *, client: PluginModelClient | None = None
     ) -> tuple[ProviderEntity, ...] | None:
         generation = cls._load_plugin_model_providers_generation(tenant_id)
+        cached_providers, _ = cls._load_cached_plugin_model_providers_for_generation(tenant_id, generation)
+        return cached_providers
+
+    @classmethod
+    def _load_cached_plugin_model_providers_for_generation(
+        cls, tenant_id: str, generation: int | None
+    ) -> tuple[tuple[ProviderEntity, ...] | None, bool]:
         if generation is not None:
             in_memory_cached_providers = cls._load_in_memory_plugin_model_providers(tenant_id, generation)
             if in_memory_cached_providers is not None:
-                return in_memory_cached_providers
+                return in_memory_cached_providers, True
+
+        if generation is None:
+            return None, False
 
         cache_keys = []
-        if generation is not None:
-            cache_keys.append(cls._get_plugin_model_providers_cache_key(tenant_id, generation))
-            if generation == 0:
-                cache_keys.append(cls._get_plugin_model_providers_cache_key(tenant_id))
+        cache_keys.append(cls._get_plugin_model_providers_cache_key(tenant_id, generation))
+        if generation == 0:
+            cache_keys.append(cls._get_plugin_model_providers_cache_key(tenant_id))
 
         if not cache_keys:
-            return None
+            return None, True
 
         try:
             cached_provider_entries = redis_client.mget(cache_keys)
-        except (RedisError, RuntimeError):
+        except (LockError, RedisError, RuntimeError):
             logger.warning("Failed to read cached plugin model providers for tenant %s.", tenant_id, exc_info=True)
-            return None
+            return None, False
 
         if len(cached_provider_entries) != len(cache_keys):
             logger.warning(
                 "Unexpected cached plugin model providers response size for tenant %s.",
                 tenant_id,
             )
-            return None
+            return None, False
 
         for cache_key, cached_providers in zip(cache_keys, cached_provider_entries):
             if not cached_providers:
@@ -232,7 +250,7 @@ class PluginService:
                 providers = tuple(_provider_entities_adapter.validate_json(cached_providers))
                 if generation is not None:
                     cls._store_in_memory_plugin_model_providers(tenant_id, generation, providers)
-                return providers
+                return providers, True
             except (TypeError, ValueError, ValidationError):
                 logger.warning(
                     "Invalid cached plugin model providers for tenant %s; deleting cache key %s.",
@@ -249,7 +267,7 @@ class PluginService:
                         exc_info=True,
                     )
 
-        return None
+        return None, True
 
     @classmethod
     def _store_cached_plugin_model_providers(
@@ -262,6 +280,49 @@ class PluginService:
         except (RedisError, RuntimeError):
             logger.warning("Failed to cache plugin model providers for tenant %s.", tenant_id, exc_info=True)
 
+    @classmethod
+    def _try_acquire_plugin_model_providers_lock(cls, tenant_id: str, generation: int) -> tuple[Any | None, bool]:
+        lock_key = cls._get_plugin_model_providers_lock_key(tenant_id, generation)
+        try:
+            lock = redis_client.lock(lock_key, timeout=cls.PLUGIN_MODEL_PROVIDERS_LOCK_TTL, blocking=False)
+            acquired = lock.acquire(blocking=False)
+        except (RedisError, RuntimeError):
+            logger.warning(
+                "Failed to acquire plugin model providers refresh lock for tenant %s.",
+                tenant_id,
+                exc_info=True,
+            )
+            return None, False
+
+        if not acquired:
+            return None, True
+
+        return lock, True
+
+    @classmethod
+    def _release_plugin_model_providers_lock(cls, tenant_id: str, lock: Any) -> None:
+        try:
+            lock.release()
+        except (LockError, RedisError, RuntimeError):
+            logger.warning(
+                "Failed to release plugin model providers refresh lock for tenant %s.",
+                tenant_id,
+                exc_info=True,
+            )
+
+    @classmethod
+    def _wait_for_plugin_model_providers_refresh(
+        cls, tenant_id: str, *, client: PluginModelClient | None = None
+    ) -> tuple[ProviderEntity, ...] | None:
+        deadline = time.monotonic() + cls.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT
+        while time.monotonic() < deadline:
+            time.sleep(cls.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_INTERVAL)
+            cached_providers = cls._load_cached_plugin_model_providers(tenant_id, client=client)
+            if cached_providers is not None:
+                return cached_providers
+
+        return None
+
     @classmethod
     def invalidate_plugin_model_providers_cache(cls, tenant_id: str) -> None:
         """Invalidate tenant-scoped provider metadata across Redis and worker-local mirrors."""
@@ -287,21 +348,38 @@ class PluginService:
         are intentionally owned by this service so tenant isolation and cache
         expiry are handled in one place.
         """
-        cached_providers = cls._load_cached_plugin_model_providers(tenant_id, client=client)
+        generation = cls._load_plugin_model_providers_generation(tenant_id)
+        cached_providers, cache_available = cls._load_cached_plugin_model_providers_for_generation(
+            tenant_id, generation
+        )
         if cached_providers is not None:
             return cached_providers
 
+        refresh_lock: Any | None = None
+        refresh_generation = generation
+        if generation is not None and cache_available:
+            lock_wait_deadline = time.monotonic() + cls.PLUGIN_MODEL_PROVIDERS_LOCK_TTL
+            while time.monotonic() < lock_wait_deadline:
+                refresh_lock, lock_available = cls._try_acquire_plugin_model_providers_lock(tenant_id, generation)
+                if refresh_lock is not None or not lock_available:
+                    break
+                refreshed_providers = cls._wait_for_plugin_model_providers_refresh(tenant_id, client=client)
+                if refreshed_providers is not None:
+                    return refreshed_providers
+
         model_client = client or PluginModelClient()
-        providers = tuple(
-            cls._to_provider_entity(provider) for provider in model_client.fetch_model_providers(tenant_id)
-        )
-        if not providers:
+        try:
+            providers = tuple(
+                cls._to_provider_entity(provider) for provider in model_client.fetch_model_providers(tenant_id)
+            )
+            generation = cls._load_plugin_model_providers_generation(tenant_id)
+            if generation is not None and generation == refresh_generation:
+                cls._store_in_memory_plugin_model_providers(tenant_id, generation, providers)
+                cls._store_cached_plugin_model_providers(tenant_id, generation, providers)
             return providers
-        generation = cls._load_plugin_model_providers_generation(tenant_id)
-        if generation is not None:
-            cls._store_in_memory_plugin_model_providers(tenant_id, generation, providers)
-            cls._store_cached_plugin_model_providers(tenant_id, generation, providers)
-        return providers
+        finally:
+            if refresh_lock is not None:
+                cls._release_plugin_model_providers_lock(tenant_id, refresh_lock)
 
     @staticmethod
     def fetch_latest_plugin_version(plugin_ids: Sequence[str]) -> Mapping[str, LatestPluginCache | None]:

api/tests/unit_tests/core/plugin/test_model_runtime_adapter.py

@@ -44,6 +44,29 @@ class _FakeRedis:
     def delete(self, key: str) -> None:
         self._values.pop(key, None)
 
+    def lock(self, key: str, *, timeout: int, blocking: bool) -> "_FakeRedisLock":
+        return _FakeRedisLock(self, key)
+
+
+class _FakeRedisLock:
+    def __init__(self, redis: _FakeRedis, key: str) -> None:
+        self._redis = redis
+        self._key = key
+        self._acquired = False
+
+    def acquire(self, *, blocking: bool) -> bool:
+        if self._key in self._redis._values:
+            return False
+
+        self._redis._values[self._key] = "locked"
+        self._acquired = True
+        return True
+
+    def release(self) -> None:
+        if self._acquired:
+            self._redis.delete(self._key)
+            self._acquired = False
+
 
 @pytest.fixture(autouse=True)
 def clear_plugin_model_provider_memory_cache() -> None:
@@ -416,9 +439,10 @@ class TestPluginModelRuntime:
                 mget=Mock(return_value=[None, None]),
                 delete=Mock(),
                 setex=Mock(),
+                lock=Mock(return_value=SimpleNamespace(acquire=Mock(return_value=True), release=Mock())),
             ),
         )
-        monkeypatch.setattr(plugin_service_module.dify_config, "PLUGIN_MODEL_PROVIDERS_CACHE_TTL", 300)
+        monkeypatch.setattr(plugin_service_module.dify_config, "PLUGIN_MODEL_PROVIDERS_CACHE_TTL", 0)
         runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService)
 
         runtime.fetch_model_providers()

api/tests/unit_tests/services/plugin/test_plugin_service.py

@@ -235,6 +235,172 @@ class TestPluginModelProviderCache:
         client.fetch_model_providers.assert_called_once_with("tenant-1")
         assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
 
+    def test_fetch_plugin_model_providers_waits_for_concurrent_refresh_cache_fill(self) -> None:
+        """A cache miss waits for the active tenant refresh instead of stampeding the daemon."""
+        cached_provider = _build_provider_entity()
+        cached_payload = TypeAdapter(list[ProviderEntity]).dump_json([cached_provider]).decode("utf-8")
+        cache_key = _provider_cache_key("tenant-1", 0)
+        legacy_cache_key = _provider_cache_key("tenant-1")
+
+        with (
+            patch(f"{MODULE}.redis_client") as redis_client,
+            patch(f"{MODULE}.time.sleep") as sleep,
+        ):
+            redis_client.get.return_value = None
+            redis_client.mget.side_effect = [[None, None], [cached_payload, None]]
+            redis_client.lock.return_value.acquire.return_value = False
+            client = Mock()
+            client.fetch_model_providers.return_value = [_build_plugin_model_provider(provider="anthropic")]
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        redis_client.lock.assert_called_once_with(
+            PluginService._get_plugin_model_providers_lock_key("tenant-1", 0),
+            timeout=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_TTL,
+            blocking=False,
+        )
+        redis_client.lock.return_value.acquire.assert_called_once_with(blocking=False)
+        assert redis_client.mget.call_args_list == [
+            call([cache_key, legacy_cache_key]),
+            call([cache_key, legacy_cache_key]),
+        ]
+        sleep.assert_called()
+        client.fetch_model_providers.assert_not_called()
+        assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
+
+    def test_fetch_plugin_model_providers_retries_lock_after_wait_timeout(self) -> None:
+        """Only a lock owner should refresh the daemon when the first refresh takes too long."""
+        with (
+            patch(f"{MODULE}.redis_client") as redis_client,
+            patch(f"{MODULE}.time.sleep"),
+            patch(f"{MODULE}.PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT", 0),
+        ):
+            redis_client.get.return_value = None
+            redis_client.mget.return_value = [None, None]
+            redis_client.lock.return_value.acquire.side_effect = [False, True]
+            client = Mock()
+            client.fetch_model_providers.return_value = [_build_plugin_model_provider()]
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        assert redis_client.lock.return_value.acquire.call_args_list == [
+            call(blocking=False),
+            call(blocking=False),
+        ]
+        client.fetch_model_providers.assert_called_once_with("tenant-1")
+        redis_client.lock.return_value.release.assert_called_once_with()
+        assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
+
+    def test_fetch_plugin_model_providers_releases_owned_refresh_lock_after_store(self) -> None:
+        """The refresh owner releases only its token after storing provider metadata."""
+        cache_key = _provider_cache_key("tenant-1", 0)
+        legacy_cache_key = _provider_cache_key("tenant-1")
+
+        with patch(f"{MODULE}.redis_client") as redis_client:
+            redis_client.get.return_value = None
+            redis_client.mget.return_value = [None, None]
+            redis_client.lock.return_value.acquire.return_value = True
+            client = Mock()
+            client.fetch_model_providers.return_value = [_build_plugin_model_provider()]
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        redis_client.lock.assert_called_once_with(
+            PluginService._get_plugin_model_providers_lock_key("tenant-1", 0),
+            timeout=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_TTL,
+            blocking=False,
+        )
+        redis_client.lock.return_value.acquire.assert_called_once_with(blocking=False)
+        redis_client.mget.assert_called_once_with([cache_key, legacy_cache_key])
+        redis_client.lock.return_value.release.assert_called_once_with()
+        redis_client.eval.assert_not_called()
+        client.fetch_model_providers.assert_called_once_with("tenant-1")
+        assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
+
+    def test_fetch_plugin_model_providers_skips_wait_when_refresh_lock_fails(self) -> None:
+        """Lock API failures should fall back directly instead of adding timeout latency."""
+        with (
+            patch(f"{MODULE}.redis_client") as redis_client,
+            patch(f"{MODULE}.time.sleep") as sleep,
+        ):
+            redis_client.get.return_value = None
+            redis_client.mget.return_value = [None, None]
+            redis_client.lock.side_effect = RedisError("redis unavailable")
+            redis_client.set.side_effect = AssertionError("raw redis set must not be used for refresh locks")
+            client = Mock()
+            client.fetch_model_providers.return_value = [_build_plugin_model_provider()]
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        sleep.assert_not_called()
+        redis_client.lock.assert_called_once()
+        client.fetch_model_providers.assert_called_once_with("tenant-1")
+        assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
+
+    def test_fetch_plugin_model_providers_caches_empty_provider_list(self) -> None:
+        """An empty provider list is still a valid refresh result for single-flight waiters."""
+        cache_key = _provider_cache_key("tenant-1", 0)
+        with patch(f"{MODULE}.redis_client") as redis_client:
+            redis_client.get.return_value = None
+            redis_client.mget.return_value = [None, None]
+            redis_client.lock.return_value.acquire.return_value = True
+            client = Mock()
+            client.fetch_model_providers.return_value = []
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        assert result == ()
+        redis_client.setex.assert_called_once()
+        assert redis_client.setex.call_args.args[0] == cache_key
+        redis_client.lock.return_value.release.assert_called_once_with()
+
+    def test_fetch_plugin_model_providers_skips_cache_write_when_generation_changes_during_refresh(self) -> None:
+        """A refresh that started before invalidation must not populate the newer generation cache."""
+        with patch(f"{MODULE}.redis_client") as redis_client:
+            redis_client.get.side_effect = [None, "1"]
+            redis_client.mget.return_value = [None, None]
+            redis_client.lock.return_value.acquire.return_value = True
+            client = Mock()
+            client.fetch_model_providers.return_value = []
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        assert result == ()
+        client.fetch_model_providers.assert_called_once_with("tenant-1")
+        redis_client.setex.assert_not_called()
+        redis_client.lock.return_value.release.assert_called_once_with()
+
+    def test_fetch_plugin_model_providers_reuses_cached_empty_provider_list(self) -> None:
+        """A cached empty list should prevent repeated daemon fetches for tenants without plugin models."""
+        empty_payload = TypeAdapter(list[ProviderEntity]).dump_json([]).decode("utf-8")
+        cache_key = _provider_cache_key("tenant-1", 0)
+        legacy_cache_key = _provider_cache_key("tenant-1")
+
+        with patch(f"{MODULE}.redis_client") as redis_client:
+            redis_client.get.return_value = None
+            redis_client.mget.return_value = [empty_payload, None]
+            client = Mock()
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        assert result == ()
+        redis_client.mget.assert_called_once_with([cache_key, legacy_cache_key])
+        client.fetch_model_providers.assert_not_called()
+
     def test_fetch_plugin_model_providers_creates_default_client_on_cache_miss(self) -> None:
         """The service owns plugin daemon access when no runtime-provided client is injected."""
         with (