From 200f8b800f1849f1354e2931bd9cd4cbdb7866cc Mon Sep 17 00:00:00 2001 From: Pyuyi <136783609@qq.com> Date: Tue, 30 Jun 2026 18:09:05 +0800 Subject: [PATCH] fix(api): prevent plugin provider cache stampedes (#37388) Co-authored-by: VeraPyuyi <204892921+VeraPyuyi@users.noreply.github.com> --- api/core/plugin/plugin_service.py | 122 ++++++++++--- .../core/plugin/test_model_runtime_adapter.py | 26 ++- .../services/plugin/test_plugin_service.py | 166 ++++++++++++++++++ 3 files changed, 291 insertions(+), 23 deletions(-) diff --git a/api/core/plugin/plugin_service.py b/api/core/plugin/plugin_service.py index 2ab3f87db72..4b749bb4c90 100644 --- a/api/core/plugin/plugin_service.py +++ b/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]: diff --git a/api/tests/unit_tests/core/plugin/test_model_runtime_adapter.py b/api/tests/unit_tests/core/plugin/test_model_runtime_adapter.py index 17973916779..c3ee4227d25 100644 --- a/api/tests/unit_tests/core/plugin/test_model_runtime_adapter.py +++ b/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() diff --git a/api/tests/unit_tests/services/plugin/test_plugin_service.py b/api/tests/unit_tests/services/plugin/test_plugin_service.py index fca05f94fc7..c7bd4dff08e 100644 --- a/api/tests/unit_tests/services/plugin/test_plugin_service.py +++ b/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 (