mirror of
https://github.com/langgenius/dify.git
synced 2026-09-24 23:22:26 +08:00
fix: can't debug model plugins (#38500)
This commit is contained in:
@@ -92,6 +92,7 @@ class PluginService:
|
||||
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_REMOTE_DEBUG_REDIS_KEY_PREFIX = "plugin_model_providers_remote_debug:tenant_id:"
|
||||
PLUGIN_MODEL_PROVIDERS_LOCK_TTL = 30
|
||||
PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT = 2.0
|
||||
PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_INTERVAL = 0.05
|
||||
@@ -117,6 +118,10 @@ class PluginService:
|
||||
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}"
|
||||
|
||||
@classmethod
|
||||
def _get_plugin_model_providers_remote_debug_cache_key(cls, tenant_id: str) -> str:
|
||||
return f"{cls.PLUGIN_MODEL_PROVIDERS_REMOTE_DEBUG_REDIS_KEY_PREFIX}{tenant_id}"
|
||||
|
||||
@staticmethod
|
||||
def _get_provider_short_name_alias(provider: PluginModelProviderEntity) -> str:
|
||||
"""
|
||||
@@ -259,6 +264,111 @@ class PluginService:
|
||||
except (RedisError, RuntimeError):
|
||||
logger.warning("Failed to cache plugin model providers for tenant %s.", tenant_id, exc_info=True)
|
||||
|
||||
@classmethod
|
||||
def _get_remote_model_plugin_cache_marker(cls, plugins: Sequence[PluginEntity]) -> str | None:
|
||||
remote_model_plugins = sorted(
|
||||
f"{plugin.plugin_id}:{plugin.plugin_unique_identifier}"
|
||||
for plugin in plugins
|
||||
if plugin.source == PluginInstallationSource.Remote
|
||||
)
|
||||
if not remote_model_plugins:
|
||||
return None
|
||||
|
||||
return "\n".join(remote_model_plugins)
|
||||
|
||||
@classmethod
|
||||
def _load_cached_remote_model_plugin_marker(cls, tenant_id: str) -> str | None:
|
||||
cache_key = cls._get_plugin_model_providers_remote_debug_cache_key(tenant_id)
|
||||
try:
|
||||
cached_marker = redis_client.get(cache_key)
|
||||
except (RedisError, RuntimeError):
|
||||
logger.warning("Failed to read remote debug model plugin marker for tenant %s.", tenant_id, exc_info=True)
|
||||
return None
|
||||
|
||||
if cached_marker is None:
|
||||
return None
|
||||
if isinstance(cached_marker, bytes):
|
||||
try:
|
||||
return cached_marker.decode()
|
||||
except UnicodeDecodeError:
|
||||
logger.warning(
|
||||
"Invalid remote debug model plugin marker for tenant %s; deleting cache marker.",
|
||||
tenant_id,
|
||||
exc_info=True,
|
||||
)
|
||||
try:
|
||||
redis_client.delete(cache_key)
|
||||
except (RedisError, RuntimeError):
|
||||
logger.warning(
|
||||
"Failed to delete invalid remote debug model plugin marker for tenant %s.",
|
||||
tenant_id,
|
||||
exc_info=True,
|
||||
)
|
||||
return None
|
||||
if isinstance(cached_marker, str):
|
||||
return cached_marker
|
||||
|
||||
logger.warning("Invalid remote debug model plugin marker for tenant %s; deleting cache marker.", tenant_id)
|
||||
try:
|
||||
redis_client.delete(cache_key)
|
||||
except (RedisError, RuntimeError):
|
||||
logger.warning(
|
||||
"Failed to delete invalid remote debug model plugin marker for tenant %s.",
|
||||
tenant_id,
|
||||
exc_info=True,
|
||||
)
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def _store_cached_remote_model_plugin_marker(cls, tenant_id: str, marker: str | None) -> None:
|
||||
cache_key = cls._get_plugin_model_providers_remote_debug_cache_key(tenant_id)
|
||||
try:
|
||||
if marker is None:
|
||||
redis_client.delete(cache_key)
|
||||
else:
|
||||
redis_client.setex(cache_key, dify_config.PLUGIN_MODEL_PROVIDERS_CACHE_TTL, marker)
|
||||
except (RedisError, RuntimeError):
|
||||
logger.warning("Failed to cache remote debug model plugin marker for tenant %s.", tenant_id, exc_info=True)
|
||||
|
||||
@classmethod
|
||||
def _load_cached_plugin_model_provider_plugin_ids(cls, tenant_id: str) -> set[str] | None:
|
||||
"""Return plugin ids represented by the current provider cache, or None when no usable cache exists."""
|
||||
generation = cls._load_plugin_model_providers_generation(tenant_id)
|
||||
cached_providers, _ = cls._load_cached_plugin_model_providers_for_generation(tenant_id, generation)
|
||||
if cached_providers is None:
|
||||
return None
|
||||
|
||||
plugin_ids: set[str] = set()
|
||||
for provider in cached_providers:
|
||||
last_slash = provider.provider.rfind("/")
|
||||
if last_slash > 0:
|
||||
plugin_ids.add(provider.provider[:last_slash])
|
||||
|
||||
return plugin_ids
|
||||
|
||||
@classmethod
|
||||
def _should_invalidate_model_provider_cache_for_remote_model_plugins(
|
||||
cls,
|
||||
tenant_id: str,
|
||||
plugins: Sequence[PluginEntity],
|
||||
) -> bool:
|
||||
remote_model_plugin_marker = cls._get_remote_model_plugin_cache_marker(plugins)
|
||||
cached_remote_model_plugin_marker = cls._load_cached_remote_model_plugin_marker(tenant_id)
|
||||
if remote_model_plugin_marker is None:
|
||||
return cached_remote_model_plugin_marker is not None
|
||||
|
||||
if remote_model_plugin_marker != cached_remote_model_plugin_marker:
|
||||
return True
|
||||
|
||||
remote_model_plugin_ids = {
|
||||
plugin.plugin_id for plugin in plugins if plugin.source == PluginInstallationSource.Remote
|
||||
}
|
||||
cached_plugin_ids = cls._load_cached_plugin_model_provider_plugin_ids(tenant_id)
|
||||
if cached_plugin_ids is None:
|
||||
return False
|
||||
|
||||
return not remote_model_plugin_ids.issubset(cached_plugin_ids)
|
||||
|
||||
@classmethod
|
||||
@contextmanager
|
||||
def _plugin_model_providers_refresh_lock(
|
||||
@@ -571,7 +681,21 @@ class PluginService:
|
||||
This keeps pagination usable before category is persisted on installation rows.
|
||||
"""
|
||||
manager = PluginInstaller()
|
||||
return manager.list_plugins_by_category(tenant_id, category, page, page_size)
|
||||
plugins = manager.list_plugins_by_category(tenant_id, category, page, page_size)
|
||||
if category == PluginCategory.Model:
|
||||
should_invalidate_model_provider_cache = (
|
||||
PluginService._should_invalidate_model_provider_cache_for_remote_model_plugins(
|
||||
tenant_id,
|
||||
plugins.list,
|
||||
)
|
||||
)
|
||||
if should_invalidate_model_provider_cache:
|
||||
PluginService.invalidate_plugin_model_providers_cache(tenant_id)
|
||||
|
||||
remote_model_plugin_marker = PluginService._get_remote_model_plugin_cache_marker(plugins.list)
|
||||
PluginService._store_cached_remote_model_plugin_marker(tenant_id, remote_model_plugin_marker)
|
||||
|
||||
return plugins
|
||||
|
||||
@staticmethod
|
||||
def _normalize_endpoint_count(value: object) -> int:
|
||||
|
||||
@@ -8,7 +8,7 @@ import zstandard
|
||||
from pydantic import TypeAdapter
|
||||
from redis import RedisError
|
||||
|
||||
from core.plugin.entities.plugin import PluginInstallationSource
|
||||
from core.plugin.entities.plugin import PluginCategory, PluginInstallationSource
|
||||
from core.plugin.entities.plugin_daemon import PluginInstallTask, PluginInstallTaskStatus, PluginModelProviderEntity
|
||||
from graphon.model_runtime.entities.common_entities import I18nObject
|
||||
from graphon.model_runtime.entities.provider_entities import ConfigurateMethod, ProviderEntity
|
||||
@@ -71,6 +71,16 @@ def _build_install_task(*, task_id: str = "task-1", status: PluginInstallTaskSta
|
||||
)
|
||||
|
||||
|
||||
def _build_remote_model_plugin(
|
||||
*, plugin_id: str = "langgenius/debug-model", plugin_unique_identifier: str = "langgenius/debug-model:1.0.0"
|
||||
) -> SimpleNamespace:
|
||||
return SimpleNamespace(
|
||||
plugin_id=plugin_id,
|
||||
plugin_unique_identifier=plugin_unique_identifier,
|
||||
source=PluginInstallationSource.Remote,
|
||||
)
|
||||
|
||||
|
||||
def _provider_cache_key(tenant_id: str, generation: int | None = None) -> str:
|
||||
if generation is None:
|
||||
return f"plugin_model_providers:tenant_id:{tenant_id}"
|
||||
@@ -797,6 +807,144 @@ class TestPluginListEndpointCounts:
|
||||
|
||||
|
||||
class TestPluginModelProviderCacheInvalidation:
|
||||
def test_get_debugging_key_does_not_invalidate_model_provider_cache(self) -> None:
|
||||
"""Reading a debug key does not mean a debug runtime has registered a model provider."""
|
||||
with (
|
||||
patch(f"{MODULE}.PluginDebuggingClient") as debugging_client_cls,
|
||||
patch(f"{MODULE}.PluginService.invalidate_plugin_model_providers_cache") as invalidate_cache,
|
||||
):
|
||||
debugging_client_cls.return_value.get_debugging_key.return_value = "debug-key"
|
||||
|
||||
from core.plugin.plugin_service import PluginService
|
||||
|
||||
result = PluginService.get_debugging_key("tenant-1")
|
||||
|
||||
assert result == "debug-key"
|
||||
debugging_client_cls.return_value.get_debugging_key.assert_called_once_with("tenant-1")
|
||||
invalidate_cache.assert_not_called()
|
||||
|
||||
def test_list_model_category_invalidates_when_remote_model_plugin_is_missing_from_provider_cache(self) -> None:
|
||||
"""Remote model plugins are daemon-registered, so category reads repair a stale provider cache."""
|
||||
remote_plugin = _build_remote_model_plugin()
|
||||
remote_plugin_marker = "langgenius/debug-model:langgenius/debug-model:1.0.0"
|
||||
plugins = SimpleNamespace(list=[remote_plugin], has_more=False)
|
||||
|
||||
with (
|
||||
patch(f"{MODULE}.PluginInstaller") as installer_cls,
|
||||
patch(
|
||||
f"{MODULE}.PluginService._load_cached_remote_model_plugin_marker",
|
||||
return_value=remote_plugin_marker,
|
||||
),
|
||||
patch(
|
||||
f"{MODULE}.PluginService._load_cached_plugin_model_provider_plugin_ids",
|
||||
return_value={"langgenius/openai"},
|
||||
),
|
||||
patch(f"{MODULE}.PluginService.invalidate_plugin_model_providers_cache") as invalidate_cache,
|
||||
patch(f"{MODULE}.PluginService._store_cached_remote_model_plugin_marker") as store_marker,
|
||||
):
|
||||
installer_cls.return_value.list_plugins_by_category.return_value = plugins
|
||||
|
||||
from core.plugin.plugin_service import PluginService
|
||||
|
||||
result = PluginService.list_by_category("tenant-1", PluginCategory.Model, 1, 100)
|
||||
|
||||
assert result is plugins
|
||||
installer_cls.return_value.list_plugins_by_category.assert_called_once_with(
|
||||
"tenant-1", PluginCategory.Model, 1, 100
|
||||
)
|
||||
invalidate_cache.assert_called_once_with("tenant-1")
|
||||
store_marker.assert_called_once_with("tenant-1", remote_plugin_marker)
|
||||
|
||||
def test_list_model_category_invalidates_when_remote_model_plugin_identity_changes(self) -> None:
|
||||
"""A debug model plugin can share plugin_id with an installed plugin, so identity changes bust cache too."""
|
||||
remote_plugin = _build_remote_model_plugin(
|
||||
plugin_id="langgenius/openai",
|
||||
plugin_unique_identifier="langgenius/openai:debug",
|
||||
)
|
||||
remote_plugin_marker = "langgenius/openai:langgenius/openai:debug"
|
||||
plugins = SimpleNamespace(list=[remote_plugin], has_more=False)
|
||||
|
||||
with (
|
||||
patch(f"{MODULE}.PluginInstaller") as installer_cls,
|
||||
patch(
|
||||
f"{MODULE}.PluginService._load_cached_remote_model_plugin_marker",
|
||||
return_value="langgenius/openai:langgenius/openai:1.0.0",
|
||||
),
|
||||
patch(
|
||||
f"{MODULE}.PluginService._load_cached_plugin_model_provider_plugin_ids",
|
||||
return_value={"langgenius/openai"},
|
||||
) as load_cached_provider_plugin_ids,
|
||||
patch(f"{MODULE}.PluginService.invalidate_plugin_model_providers_cache") as invalidate_cache,
|
||||
patch(f"{MODULE}.PluginService._store_cached_remote_model_plugin_marker") as store_marker,
|
||||
):
|
||||
installer_cls.return_value.list_plugins_by_category.return_value = plugins
|
||||
|
||||
from core.plugin.plugin_service import PluginService
|
||||
|
||||
result = PluginService.list_by_category("tenant-1", PluginCategory.Model, 1, 100)
|
||||
|
||||
assert result is plugins
|
||||
invalidate_cache.assert_called_once_with("tenant-1")
|
||||
load_cached_provider_plugin_ids.assert_not_called()
|
||||
store_marker.assert_called_once_with("tenant-1", remote_plugin_marker)
|
||||
|
||||
def test_list_model_category_keeps_provider_cache_when_remote_model_plugin_is_already_cached(self) -> None:
|
||||
"""A connected remote model plugin should not force provider cache churn once represented."""
|
||||
remote_plugin = _build_remote_model_plugin()
|
||||
remote_plugin_marker = "langgenius/debug-model:langgenius/debug-model:1.0.0"
|
||||
plugins = SimpleNamespace(list=[remote_plugin], has_more=False)
|
||||
|
||||
with (
|
||||
patch(f"{MODULE}.PluginInstaller") as installer_cls,
|
||||
patch(
|
||||
f"{MODULE}.PluginService._load_cached_remote_model_plugin_marker",
|
||||
return_value=remote_plugin_marker,
|
||||
),
|
||||
patch(
|
||||
f"{MODULE}.PluginService._load_cached_plugin_model_provider_plugin_ids",
|
||||
return_value={"langgenius/debug-model"},
|
||||
),
|
||||
patch(f"{MODULE}.PluginService.invalidate_plugin_model_providers_cache") as invalidate_cache,
|
||||
patch(f"{MODULE}.PluginService._store_cached_remote_model_plugin_marker") as store_marker,
|
||||
):
|
||||
installer_cls.return_value.list_plugins_by_category.return_value = plugins
|
||||
|
||||
from core.plugin.plugin_service import PluginService
|
||||
|
||||
result = PluginService.list_by_category("tenant-1", PluginCategory.Model, 1, 100)
|
||||
|
||||
assert result is plugins
|
||||
invalidate_cache.assert_not_called()
|
||||
store_marker.assert_called_once_with("tenant-1", remote_plugin_marker)
|
||||
|
||||
def test_list_model_category_invalidates_when_remote_model_plugin_disconnects(self) -> None:
|
||||
"""The current model category result clears provider cache when the previous debug model disappears."""
|
||||
installed_plugin = SimpleNamespace(
|
||||
plugin_id="langgenius/openai",
|
||||
plugin_unique_identifier="langgenius/openai:1.0.0",
|
||||
source=PluginInstallationSource.Marketplace,
|
||||
)
|
||||
plugins = SimpleNamespace(list=[installed_plugin], has_more=True)
|
||||
|
||||
with (
|
||||
patch(f"{MODULE}.PluginInstaller") as installer_cls,
|
||||
patch(
|
||||
f"{MODULE}.PluginService._load_cached_remote_model_plugin_marker",
|
||||
return_value="langgenius/debug-model:langgenius/debug-model:1.0.0",
|
||||
),
|
||||
patch(f"{MODULE}.PluginService.invalidate_plugin_model_providers_cache") as invalidate_cache,
|
||||
patch(f"{MODULE}.PluginService._store_cached_remote_model_plugin_marker") as store_marker,
|
||||
):
|
||||
installer_cls.return_value.list_plugins_by_category.return_value = plugins
|
||||
|
||||
from core.plugin.plugin_service import PluginService
|
||||
|
||||
result = PluginService.list_by_category("tenant-1", PluginCategory.Model, 1, 100)
|
||||
|
||||
assert result is plugins
|
||||
invalidate_cache.assert_called_once_with("tenant-1")
|
||||
store_marker.assert_called_once_with("tenant-1", None)
|
||||
|
||||
def test_fetch_install_task_invalidates_model_provider_cache_when_finished(self) -> None:
|
||||
"""Finished plugin install tasks invalidate tenant provider cache."""
|
||||
task = _build_install_task(status=PluginInstallTaskStatus.Success)
|
||||
|
||||
Reference in New Issue
Block a user