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

@@ -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)

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()