From 059a1fe090540302d920f4d3c84958b6dc4d9ef4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9D=9E=E6=B3=95=E6=93=8D=E4=BD=9C?= Date: Fri, 3 Jul 2026 15:32:07 +0800 Subject: [PATCH] chore: compress large plugin_model obj in redis (#38374) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> --- api/core/plugin/plugin_service.py | 33 ++++++++++- .../services/plugin/test_plugin_service.py | 56 +++++++++++++++++++ 2 files changed, 87 insertions(+), 2 deletions(-) diff --git a/api/core/plugin/plugin_service.py b/api/core/plugin/plugin_service.py index 694599a5c2d..a6d83e85dbc 100644 --- a/api/core/plugin/plugin_service.py +++ b/api/core/plugin/plugin_service.py @@ -4,6 +4,8 @@ This module owns plugin daemon management calls that are shared by API services and core runtimes. Plugin model provider discovery is cached here, alongside plugin install, uninstall, and upgrade invalidation, so all cache mutations for plugin-owned provider metadata stay tenant-scoped and in one place. +Provider cache payloads may be stored as prefixed zlib bytes; readers also +accept legacy plain JSON payloads for rolling upgrades and existing Redis keys. The console plugin list also normalizes endpoint setup counters against live endpoint records. Some plugin daemon builds return stale ``endpoints_*`` @@ -14,6 +16,7 @@ metadata. import logging import time +import zlib from collections.abc import Iterator, Mapping, Sequence from contextlib import contextmanager from mimetypes import guess_type @@ -92,6 +95,8 @@ class PluginService: PLUGIN_MODEL_PROVIDERS_LOCK_TTL = 30 PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT = 2.0 PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_INTERVAL = 0.05 + PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX = b"\x00dify-plugin-model-providers-zlib-v1:" + PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_MIN_BYTES = 64 * 1024 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. @@ -142,6 +147,27 @@ class PluginService: declaration.provider_name = cls._get_provider_short_name_alias(provider) return declaration + @classmethod + def _encode_plugin_model_providers_cache_payload(cls, payload: bytes) -> bytes: + if len(payload) < cls.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_MIN_BYTES: + return payload + + return cls.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX + zlib.compress(payload, level=1) + + @classmethod + def _decode_plugin_model_providers_cache_payload(cls, payload: bytes | bytearray | str) -> bytes | bytearray | str: + if isinstance(payload, str): + return payload + + prefix = cls.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX + if not payload.startswith(prefix): + return payload + + try: + return zlib.decompress(payload[len(prefix) :]) + except zlib.error as exc: + raise ValueError("Invalid compressed plugin model providers cache payload.") from exc + @classmethod def _load_plugin_model_providers_generation(cls, tenant_id: str) -> int | None: cache_key = cls._get_plugin_model_providers_generation_cache_key(tenant_id) @@ -199,7 +225,8 @@ class PluginService: continue try: - providers = tuple(_provider_entities_adapter.validate_json(cached_providers)) + payload = cls._decode_plugin_model_providers_cache_payload(cached_providers) + providers = tuple(_provider_entities_adapter.validate_json(payload)) return providers, True except (TypeError, ValueError, ValidationError): logger.warning( @@ -225,7 +252,9 @@ class PluginService: ) -> None: cache_key = cls._get_plugin_model_providers_cache_key(tenant_id, generation) try: - payload = _provider_entities_adapter.dump_json(list(providers)) + payload = cls._encode_plugin_model_providers_cache_payload( + _provider_entities_adapter.dump_json(list(providers)) + ) redis_client.setex(cache_key, dify_config.PLUGIN_MODEL_PROVIDERS_CACHE_TTL, payload) except (RedisError, RuntimeError): logger.warning("Failed to cache plugin model providers for tenant %s.", tenant_id, exc_info=True) 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 f847b868e89..11230b25b40 100644 --- a/api/tests/unit_tests/services/plugin/test_plugin_service.py +++ b/api/tests/unit_tests/services/plugin/test_plugin_service.py @@ -1,5 +1,6 @@ import datetime import uuid +import zlib from types import SimpleNamespace from unittest.mock import MagicMock, Mock, call, patch @@ -128,6 +129,61 @@ class TestFetchLatestPluginVersion: class TestPluginModelProviderCache: + def test_store_cached_plugin_model_providers_compresses_large_payload(self) -> None: + """Large provider metadata payloads are compressed before being stored in Redis.""" + large_provider = _build_provider_entity() + large_provider.label = I18nObject(en_US="OpenAI " * 10000) + raw_payload = TypeAdapter(list[ProviderEntity]).dump_json([large_provider]) + cache_key = _provider_cache_key("tenant-1", 0) + + with ( + patch(f"{MODULE}.redis_client") as redis_client, + patch(f"{MODULE}.dify_config") as mock_config, + ): + mock_config.PLUGIN_MODEL_PROVIDERS_CACHE_TTL = 86400 + + from core.plugin.plugin_service import PluginService + + PluginService._store_cached_plugin_model_providers("tenant-1", 0, [large_provider]) + + redis_client.setex.assert_called_once() + stored_key, ttl, stored_payload = redis_client.setex.call_args.args + assert stored_key == cache_key + assert ttl == 86400 + assert isinstance(stored_payload, bytes) + prefix = PluginService.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX + assert stored_payload.startswith(prefix) + assert len(stored_payload) < len(raw_payload) + assert zlib.decompress(stored_payload[len(prefix) :]) == raw_payload + + def test_fetch_plugin_model_providers_reads_compressed_cached_provider_without_calling_daemon(self) -> None: + """Compressed tenant cache entries are decoded before provider schema validation.""" + cached_provider = _build_provider_entity() + cached_provider.label = I18nObject(en_US="OpenAI " * 10000) + cached_payload = TypeAdapter(list[ProviderEntity]).dump_json([cached_provider]) + generation_key = _provider_generation_key("tenant-1") + cache_key = _provider_cache_key("tenant-1", 0) + + from core.plugin.plugin_service import PluginService + + compressed_payload = PluginService.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX + zlib.compress( + cached_payload, level=1 + ) + + with patch(f"{MODULE}.redis_client") as redis_client: + redis_client.get.return_value = None + redis_client.mget.return_value = [compressed_payload] + client = Mock() + + result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client) + + assert [provider.provider for provider in result] == ["langgenius/openai/openai"] + assert result[0].label.en_us == "OpenAI " * 10000 + client.fetch_model_providers.assert_not_called() + redis_client.setex.assert_not_called() + redis_client.get.assert_called_once_with(generation_key) + redis_client.mget.assert_called_once_with([cache_key]) + def test_fetch_plugin_model_providers_returns_cached_provider_without_calling_daemon(self) -> None: """A valid tenant cache entry is reused across runtime calls without plugin daemon access.""" cached_provider = _build_provider_entity()