fix: fall back to the batch API when the global plugin manifest is empty (#41107)

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

api/core/helper/marketplace.py

@@ -74,7 +74,7 @@ def record_install_plugin_event(plugin_unique_identifier: str) -> None:
     response.raise_for_status()
 
 
-def fetch_global_plugin_manifest(cache_key_prefix: str, cache_ttl: int) -> None:
+def fetch_global_plugin_manifest(cache_key_prefix: str, cache_ttl: int) -> int:
     """
     Fetch all plugin manifests from marketplace and cache them in Redis.
     This should be called once per check cycle to populate the instance-level cache.
@@ -83,6 +83,11 @@ def fetch_global_plugin_manifest(cache_key_prefix: str, cache_ttl: int) -> None:
         cache_key_prefix: Redis key prefix for caching plugin manifests
         cache_ttl: Cache TTL in seconds
 
+    Returns:
+        The number of plugin snapshots that were cached. Callers should treat ``0`` as an
+        unusable snapshot rather than an empty marketplace, and fall back to per-plugin
+        lookups instead of silently skipping upgrades.
+
     Raises:
         httpx.HTTPError: If the HTTP request fails
         Exception: If any other error occurs during fetching or caching
@@ -95,6 +100,7 @@ def fetch_global_plugin_manifest(cache_key_prefix: str, cache_ttl: int) -> None:
     plugins_data = raw_json.get("plugins", [])
 
     # Parse and cache all plugin snapshots
+    cached_count = 0
     for plugin_data in plugins_data:
         plugin_snapshot = MarketplacePluginSnapshot.model_validate(plugin_data)
         redis_client.setex(
@@ -102,3 +108,6 @@ def fetch_global_plugin_manifest(cache_key_prefix: str, cache_ttl: int) -> None:
             time=cache_ttl,
             value=plugin_snapshot.model_dump_json(),
         )
+        cached_count += 1
+
+    return cached_count

api/schedule/check_upgradable_plugin_task.py

@@ -54,14 +54,23 @@ def check_upgradable_plugin_task():
     # This reduces load on marketplace from 300k requests to 1 request per check cycle
     logger.info("fetching global plugin manifest from marketplace")
     try:
-        fetch_global_plugin_manifest(CACHE_REDIS_KEY_PREFIX, CACHE_REDIS_TTL)
-        logger.info("successfully fetched and cached global plugin manifest")
+        cached_count = fetch_global_plugin_manifest(CACHE_REDIS_KEY_PREFIX, CACHE_REDIS_TTL)
     except Exception as e:
+        # The marketplace is unreachable, so the per-plugin fallback would fail as well.
+        # Skip this cycle instead of issuing one doomed request per tenant.
         logger.exception("failed to fetch global plugin manifest")
         click.echo(click.style(f"failed to fetch global plugin manifest: {e}", fg="red"))
         click.echo(click.style("skipping plugin upgrade check for this cycle", fg="yellow"))
         return
 
+    if cached_count == 0:
+        # The snapshot was served but carries no plugins. The marketplace itself is reachable,
+        # so let the per-tenant tasks fall back to the batch API rather than upgrading nothing.
+        logger.warning("global plugin manifest contained no plugins; falling back to per-plugin marketplace lookups")
+        click.echo(click.style("global plugin manifest is empty, falling back to batch lookups", fg="yellow"))
+    else:
+        logger.info("successfully fetched and cached %d plugin manifests", cached_count)
+
     for i in range(0, total_strategies, MAX_CONCURRENT_CHECK_TASKS):
         batch_strategies = strategies[i : i + MAX_CONCURRENT_CHECK_TASKS]
         for strategy in batch_strategies:

api/tasks/process_tenant_plugin_autoupgrade_check_task.py

@@ -6,7 +6,8 @@ import typing
 import click
 from celery import shared_task
 
-from core.plugin.entities.marketplace import MarketplacePluginSnapshot
+from core.helper.marketplace import batch_fetch_plugin_manifests, get_plugin_pkg_url
+from core.plugin.entities.marketplace import MarketplacePluginDeclaration, MarketplacePluginSnapshot
 from core.plugin.entities.plugin import PluginInstallation, PluginInstallationSource
 from core.plugin.impl.plugin import PluginInstaller
 from core.plugin.plugin_service import PluginService
@@ -54,25 +55,86 @@ def _get_cached_manifest(plugin_id: str) -> typing.Union[MarketplacePluginSnapsh
         return False
 
 
+def _set_cached_manifest(plugin_id: str, snapshot: MarketplacePluginSnapshot | None) -> None:
+    """
+    Cache a plugin snapshot in Redis, or record that the marketplace has no such plugin.
+
+    Caching misses as well as hits keeps the fallback cheap: a plugin that the marketplace
+    does not know about is looked up once per TTL instead of once per check cycle.
+    """
+    try:
+        key = _get_redis_cache_key(plugin_id)
+        if snapshot is None:
+            redis_client.setex(key, CACHE_REDIS_TTL, json.dumps(None))
+        else:
+            redis_client.setex(key, CACHE_REDIS_TTL, snapshot.model_dump_json())
+    except Exception:
+        # If Redis fails, continue without caching
+        logger.exception("Failed to set cached manifest for plugin %s", plugin_id)
+
+
+def _snapshot_from_declaration(declaration: MarketplacePluginDeclaration) -> MarketplacePluginSnapshot:
+    """Adapt a batch-API declaration to the snapshot shape used by the upgrade check."""
+    return MarketplacePluginSnapshot(
+        org=declaration.org,
+        name=declaration.name,
+        latest_version=declaration.latest_version,
+        latest_package_identifier=declaration.latest_package_identifier,
+        latest_package_url=get_plugin_pkg_url(declaration.latest_package_identifier),
+    )
+
+
 def marketplace_batch_fetch_plugin_manifests(
     plugin_ids_plain_list: list[str],
 ) -> list[MarketplacePluginSnapshot]:
     """
-    Fetch plugin manifests from Redis cache only.
-    This function assumes fetch_global_plugin_manifest() has been called
-    to pre-populate the cache with all marketplace plugins.
+    Fetch plugin manifests, preferring the pre-populated Redis cache.
+
+    fetch_global_plugin_manifest() normally warms the cache with the whole marketplace.
+    When that snapshot is unavailable or incomplete, fall back to the per-plugin batch API
+    so upgrades still happen instead of being silently skipped.
     """
     result: list[MarketplacePluginSnapshot] = []
+    uncached_plugin_ids: list[str] = []
 
     # Check Redis cache for each plugin
     for plugin_id in plugin_ids_plain_list:
         cached_result = _get_cached_manifest(plugin_id)
-        if not isinstance(cached_result, MarketplacePluginSnapshot):
-            # cached_result is False (not in cache) or None (cached as not found)
-            logger.warning("plugin %s not found in cache, skipping", plugin_id)
+        if isinstance(cached_result, MarketplacePluginSnapshot):
+            result.append(cached_result)
+        elif cached_result is None:
+            # Cached as not found in the marketplace; nothing to upgrade to.
             continue
+        else:
+            uncached_plugin_ids.append(plugin_id)
 
-        result.append(cached_result)
+    if not uncached_plugin_ids:
+        return result
+
+    logger.info(
+        "%d plugin manifests missing from the global snapshot, fetching them from the marketplace",
+        len(uncached_plugin_ids),
+    )
+
+    try:
+        declarations = batch_fetch_plugin_manifests(uncached_plugin_ids)
+    except Exception:
+        logger.exception(
+            "failed to fetch plugin manifests from marketplace, skipping %d plugins", len(uncached_plugin_ids)
+        )
+        return result
+
+    fetched_plugin_ids = set()
+    for declaration in declarations:
+        snapshot = _snapshot_from_declaration(declaration)
+        fetched_plugin_ids.add(declaration.plugin_id)
+        _set_cached_manifest(declaration.plugin_id, snapshot)
+        result.append(snapshot)
+
+    # Remember which plugins the marketplace does not serve, so we stop asking every cycle.
+    for plugin_id in uncached_plugin_ids:
+        if plugin_id not in fetched_plugin_ids:
+            _set_cached_manifest(plugin_id, None)
 
     return result
 

api/tests/unit_tests/core/helper/test_marketplace.py

@@ -114,8 +114,35 @@ def test_fetch_global_plugin_manifest_caches_each_plugin(mocker: MockerFixture)
     )
     setex_mock = mocker.patch("core.helper.marketplace.redis_client.setex")
 
-    fetch_global_plugin_manifest("prefix:", 60)
+    cached_count = fetch_global_plugin_manifest("prefix:", 60)
 
+    assert cached_count == 2
     assert validate_mock.call_count == 2
     setex_mock.assert_any_call(name="prefix:plugin-a", time=60, value='{"id":"a"}')
     setex_mock.assert_any_call(name="prefix:plugin-b", time=60, value='{"id":"b"}')
+
+
+def test_fetch_global_plugin_manifest_reports_zero_for_empty_snapshot(mocker: MockerFixture) -> None:
+    """An empty snapshot must be reported so callers can fall back instead of upgrading nothing."""
+    empty_plugins: list[dict[str, object]] = []
+    response = MagicMock()
+    response.raise_for_status.return_value = None
+    response.json.return_value = {"metadata": {"plugin_count": 0}, "plugins": empty_plugins}
+    mocker.patch("core.helper.marketplace.httpx.get", return_value=response)
+    setex_mock = mocker.patch("core.helper.marketplace.redis_client.setex")
+
+    cached_count = fetch_global_plugin_manifest("prefix:", 60)
+
+    assert cached_count == 0
+    setex_mock.assert_not_called()
+
+
+def test_fetch_global_plugin_manifest_reports_zero_when_plugins_key_missing(mocker: MockerFixture) -> None:
+    response = MagicMock()
+    response.raise_for_status.return_value = None
+    response.json.return_value = {"metadata": {"plugin_count": 0}}
+    mocker.patch("core.helper.marketplace.httpx.get", return_value=response)
+    setex_mock = mocker.patch("core.helper.marketplace.redis_client.setex")
+
+    assert fetch_global_plugin_manifest("prefix:", 60) == 0
+    setex_mock.assert_not_called()

api/tests/unit_tests/tasks/test_process_tenant_plugin_autoupgrade_check_task.py

@@ -319,3 +319,106 @@ class TestErrorIsolation:
 
         assert any("foo" in s for s in seen)
         assert any("bar" in s for s in seen)
+
+
+def _make_declaration(plugin_id: str, latest_version: str):
+    """Minimal stand-in for a MarketplacePluginDeclaration returned by the batch API."""
+    org, name = plugin_id.split("/", 1)
+    return SimpleNamespace(
+        plugin_id=plugin_id,
+        org=org,
+        name=name,
+        latest_version=latest_version,
+        latest_package_identifier=f"{plugin_id}:{latest_version}@cafe1234",
+    )
+
+
+class TestManifestCacheFallback:
+    """
+    The global snapshot is a best-effort optimisation. When it does not cover a plugin,
+    the check must fall back to the batch API instead of silently skipping the upgrade.
+    """
+
+    def _fetch(self, plugin_ids, *, cached, declarations=None, batch_error=None):
+        from tasks.process_tenant_plugin_autoupgrade_check_task import (
+            marketplace_batch_fetch_plugin_manifests,
+        )
+
+        redis_mock = MagicMock()
+        redis_mock.get.side_effect = lambda key: cached.get(key)
+
+        batch_mock = MagicMock()
+        if batch_error is not None:
+            batch_mock.side_effect = batch_error
+        else:
+            batch_mock.return_value = declarations or []
+
+        with (
+            patch(f"{MODULE}.redis_client", redis_mock),
+            patch(f"{MODULE}.batch_fetch_plugin_manifests", batch_mock),
+        ):
+            result = marketplace_batch_fetch_plugin_manifests(plugin_ids)
+
+        return result, batch_mock, redis_mock
+
+    def test_cache_hits_do_not_touch_the_marketplace(self):
+        """The whole point of the snapshot: a warm cache must not generate any request."""
+        key = "plugin_autoupgrade_check_task:cached_plugin_snapshot:langgenius/foo"
+        cached = {key: _make_manifest("langgenius/foo", "1.1.0").model_dump_json()}
+
+        result, batch_mock, _ = self._fetch(["langgenius/foo"], cached=cached)
+
+        assert [r.latest_version for r in result] == ["1.1.0"]
+        batch_mock.assert_not_called()
+
+    def test_empty_snapshot_falls_back_to_batch_api(self):
+        """Regression: an empty global snapshot used to skip every plugin silently."""
+        result, batch_mock, _ = self._fetch(
+            ["langgenius/foo"],
+            cached={},
+            declarations=[_make_declaration("langgenius/foo", "1.2.0")],
+        )
+
+        batch_mock.assert_called_once_with(["langgenius/foo"])
+        assert [r.plugin_id for r in result] == ["langgenius/foo"]
+        assert [r.latest_version for r in result] == ["1.2.0"]
+
+    def test_fallback_results_are_cached(self):
+        _, _, redis_mock = self._fetch(
+            ["langgenius/foo"],
+            cached={},
+            declarations=[_make_declaration("langgenius/foo", "1.2.0")],
+        )
+
+        cached_keys = [call.args[0] for call in redis_mock.setex.call_args_list]
+        assert "plugin_autoupgrade_check_task:cached_plugin_snapshot:langgenius/foo" in cached_keys
+
+    def test_plugins_unknown_to_the_marketplace_are_negatively_cached(self):
+        """Avoid re-asking about a plugin the marketplace does not serve on every cycle."""
+        _, _, redis_mock = self._fetch(["langgenius/ghost"], cached={}, declarations=[])
+
+        redis_mock.setex.assert_called_once()
+        key, _ttl, value = redis_mock.setex.call_args.args
+        assert key == "plugin_autoupgrade_check_task:cached_plugin_snapshot:langgenius/ghost"
+        assert value == "null"
+
+    def test_negatively_cached_plugins_are_not_refetched(self):
+        key = "plugin_autoupgrade_check_task:cached_plugin_snapshot:langgenius/ghost"
+
+        result, batch_mock, _ = self._fetch(["langgenius/ghost"], cached={key: "null"})
+
+        assert result == []
+        batch_mock.assert_not_called()
+
+    def test_marketplace_failure_keeps_cached_results(self):
+        """A failing batch call must not lose the plugins the snapshot did cover."""
+        key = "plugin_autoupgrade_check_task:cached_plugin_snapshot:langgenius/foo"
+        cached = {key: _make_manifest("langgenius/foo", "1.1.0").model_dump_json()}
+
+        result, _, _ = self._fetch(
+            ["langgenius/foo", "langgenius/bar"],
+            cached=cached,
+            batch_error=RuntimeError("marketplace down"),
+        )
+
+        assert [r.plugin_id for r in result] == ["langgenius/foo"]