diff --git a/api/core/helper/marketplace.py b/api/core/helper/marketplace.py index e14c7185858..b45f14e7215 100644 --- a/api/core/helper/marketplace.py +++ b/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 diff --git a/api/schedule/check_upgradable_plugin_task.py b/api/schedule/check_upgradable_plugin_task.py index 8d913f1eb13..6ccc5361e32 100644 --- a/api/schedule/check_upgradable_plugin_task.py +++ b/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: diff --git a/api/tasks/process_tenant_plugin_autoupgrade_check_task.py b/api/tasks/process_tenant_plugin_autoupgrade_check_task.py index a796af5f3d3..8457117b757 100644 --- a/api/tasks/process_tenant_plugin_autoupgrade_check_task.py +++ b/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 diff --git a/api/tests/unit_tests/core/helper/test_marketplace.py b/api/tests/unit_tests/core/helper/test_marketplace.py index a14a9fe582d..6d9d37f4c93 100644 --- a/api/tests/unit_tests/core/helper/test_marketplace.py +++ b/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() diff --git a/api/tests/unit_tests/tasks/test_process_tenant_plugin_autoupgrade_check_task.py b/api/tests/unit_tests/tasks/test_process_tenant_plugin_autoupgrade_check_task.py index d26ae0b5f13..97e8db5fcc8 100644 --- a/api/tests/unit_tests/tasks/test_process_tenant_plugin_autoupgrade_check_task.py +++ b/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"]