From 099ef7d5bf1243a9ae1eaa07ae18d4f427a3dcdd Mon Sep 17 00:00:00 2001 From: qqcomeup Date: Wed, 8 Jul 2026 12:51:24 +0800 Subject: [PATCH] fix: avoid blocking plugin release history refresh (#6084) --- app/api/endpoints/plugin.py | 107 +++++++++++++++++++++------- app/core/cache.py | 22 ++++++ app/helper/plugin.py | 49 ++++++++----- tests/test_plugin_endpoint.py | 129 ++++++++++++++++++++++++++++++++++ tests/test_plugin_helper.py | 103 +++++++++++++++++++++++++++ 5 files changed, 367 insertions(+), 43 deletions(-) diff --git a/app/api/endpoints/plugin.py b/app/api/endpoints/plugin.py index c036da4c..ad2b1cc5 100644 --- a/app/api/endpoints/plugin.py +++ b/app/api/endpoints/plugin.py @@ -1,3 +1,4 @@ +import asyncio import mimetypes import shutil from typing import Annotated, Any, List, Optional @@ -39,6 +40,67 @@ PROTECTED_ROUTES = {"/api/v1/openapi.json", "/docs", "/docs/oauth2-redirect", "/ PLUGIN_PREFIX = f"{settings.API_V1_STR}/plugin" router = APIRouter() +_plugin_release_refresh_tasks: set[asyncio.Task] = set() + + +async def _get_market_plugin_from_repo( + plugin_manager: PluginManager, + plugin_id: str, + repo_url: str, + force: bool, +) -> Optional[schemas.Plugin]: + """ + 只读取指定插件仓库的市场元数据，避免单插件详情触发全部市场刷新。 + """ + market_plugins = await plugin_manager.async_get_plugins_from_market( + repo_url, settings.VERSION_FLAG, force + ) + market_plugin = next( + ( + plugin + for plugin in market_plugins or [] + if plugin.id == plugin_id + ), + None, + ) + if market_plugin or not settings.VERSION_FLAG: + return market_plugin + + compatible_plugins = await plugin_manager.async_get_plugins_from_market( + repo_url, None, force + ) + return next( + ( + plugin + for plugin in compatible_plugins or [] + if plugin.id == plugin_id + ), + None, + ) + + +async def _refresh_plugin_release_versions(plugin_id: str, repo_url: str) -> None: + """ + 后台强制刷新 Release 缓存，接口响应路径优先返回已有缓存。 + """ + try: + async with async_fresh(True): + await PluginHelper().async_get_plugin_release_versions(plugin_id, repo_url) + except Exception as e: + logger.warning(f"后台刷新插件 {plugin_id} Release 列表失败：{e}") + + +def _schedule_plugin_release_refresh(plugin_id: str, repo_url: str) -> None: + """ + 保留后台任务引用，避免任务被回收，同时让 helper 负责同仓库强刷合并。 + """ + task = asyncio.create_task(_refresh_plugin_release_versions(plugin_id, repo_url)) + _plugin_release_refresh_tasks.add(task) + + def _discard_task(completed_task: asyncio.Task) -> None: + _plugin_release_refresh_tasks.discard(completed_task) + + task.add_done_callback(_discard_task) def register_plugin_api(plugin_id: Optional[str] = None): @@ -239,6 +301,15 @@ async def _get_plugin_history_detail( if local_repo_plugin: return _merge_plugin_market_metadata(installed_plugin, local_repo_plugin) + if installed_plugin.repo_url: + market_plugin = await _get_market_plugin_from_repo( + plugin_manager, plugin_id, installed_plugin.repo_url, force + ) + if not market_plugin: + logger.debug(f"插件 {plugin_id} 未从来源仓库获取到更新说明，返回本地插件信息") + return installed_plugin + return _merge_plugin_market_metadata(installed_plugin, market_plugin) + market_plugin = next( ( plugin @@ -359,30 +430,9 @@ async def plugin_releases( } plugin_manager = PluginManager() - market_plugins = await plugin_manager.async_get_plugins_from_market( - repo_url, settings.VERSION_FLAG, force + market_plugin = await _get_market_plugin_from_repo( + plugin_manager, plugin_id, repo_url, force ) - market_plugin = next( - ( - plugin - for plugin in market_plugins or [] - if plugin.id == plugin_id - ), - None, - ) - if not market_plugin and settings.VERSION_FLAG: - compatible_plugins = await plugin_manager.async_get_plugins_from_market( - repo_url, None, force - ) - market_plugin = next( - ( - plugin - for plugin in compatible_plugins or [] - if plugin.id == plugin_id - ), - None, - ) - latest_version = market_plugin.plugin_version if market_plugin else None current_version = plugin_manager.get_local_plugin_version(plugin_id) if not getattr(market_plugin, "release", False): @@ -393,8 +443,15 @@ async def plugin_releases( "items": [], } - async with async_fresh(force): - release_items = await PluginHelper().async_get_plugin_release_versions(plugin_id, repo_url) + plugin_helper = PluginHelper() + has_release_cache = ( + await plugin_helper.async_has_plugin_release_cache(repo_url) + if force + else False + ) + release_items = await plugin_helper.async_get_plugin_release_versions(plugin_id, repo_url) + if force and has_release_cache: + _schedule_plugin_release_refresh(plugin_id, repo_url) items = [] for item in release_items: version = item.get("version") diff --git a/app/core/cache.py b/app/core/cache.py index 470646c8..24076644 100644 --- a/app/core/cache.py +++ b/app/core/cache.py @@ -1214,8 +1214,19 @@ def cached(region: Optional[str] = None, maxsize: Optional[int] = 1024, ttl: Opt """ await cache_backend.clear(region=cache_region) + async def cache_exists(*args, **kwargs) -> bool: + """ + 判断当前参数对应的有效缓存是否存在。 + """ + cache_key = __get_cache_key(args, kwargs) + cached_value = await cache_backend.get(cache_key, region=cache_region) + return should_cache(cached_value) and await async_is_valid_cache_value( + cache_key, cached_value, cache_region + ) + async_wrapper.cache_region = cache_region async_wrapper.cache_clear = cache_clear + async_wrapper.cache_exists = cache_exists return async_wrapper else: # 同步函数使用同步缓存后端 @@ -1246,8 +1257,19 @@ def cached(region: Optional[str] = None, maxsize: Optional[int] = 1024, ttl: Opt """ cache_backend.clear(region=cache_region) + def cache_exists(*args, **kwargs) -> bool: + """ + 判断当前参数对应的有效缓存是否存在。 + """ + cache_key = __get_cache_key(args, kwargs) + cached_value = cache_backend.get(cache_key, region=cache_region) + return should_cache(cached_value) and is_valid_cache_value( + cache_key, cached_value, cache_region + ) + wrapper.cache_region = cache_region wrapper.cache_clear = cache_clear + wrapper.cache_exists = cache_exists return wrapper return decorator diff --git a/app/helper/plugin.py b/app/helper/plugin.py index 6c419e70..f04477d0 100644 --- a/app/helper/plugin.py +++ b/app/helper/plugin.py @@ -2218,35 +2218,48 @@ class PluginHelper(metaclass=WeakSingleton): normal_task_key = (loop, normalized_repo_url, False) force_task_key = (loop, normalized_repo_url, True) with self._release_task_lock: - force_task = self._release_tasks.get(force_task_key) - if force_task and not force_task.done(): - task_key = force_task_key - task = force_task - elif is_fresh(): - pending_normal_task = self._release_tasks.get(normal_task_key) - if pending_normal_task and pending_normal_task.done(): - pending_normal_task = None - task_key = force_task_key - task = loop.create_task( - self._async_refresh_plugin_repo_releases(normalized_repo_url, pending_normal_task) - ) - self._release_tasks[task_key] = task - task.add_done_callback( - lambda completed_task: self._remove_release_task(task_key, completed_task) - ) + if is_fresh(): + force_task = self._release_tasks.get(force_task_key) + if force_task and not force_task.done(): + task_key = force_task_key + task = force_task + else: + pending_normal_task = self._release_tasks.get(normal_task_key) + if pending_normal_task and pending_normal_task.done(): + pending_normal_task = None + task_key = force_task_key + task = loop.create_task( + self._async_refresh_plugin_repo_releases(normalized_repo_url, pending_normal_task) + ) + self._release_tasks[task_key] = task + task.add_done_callback( + lambda completed_task: self._remove_release_task(task_key, completed_task) + ) else: task_key = normal_task_key - task = self._release_tasks.get(task_key) - if task is None or task.done(): + pending_normal_task = self._release_tasks.get(normal_task_key) + if pending_normal_task is None or pending_normal_task.done(): task = loop.create_task(self._async_get_plugin_repo_releases(normalized_repo_url)) self._release_tasks[task_key] = task task.add_done_callback( lambda completed_task: self._remove_release_task(task_key, completed_task) ) + else: + task = pending_normal_task payload = await asyncio.shield(task) return self.__parse_plugin_release_response(pid, payload) + async def async_has_plugin_release_cache(self, repo_url: str) -> bool: + """ + 判断指定仓库的 Release 列表缓存是否已经存在。 + """ + if not repo_url: + return False + return await self._async_get_plugin_repo_releases.cache_exists( + self, repo_url.rstrip("/") + ) + async def _async_refresh_plugin_repo_releases( self, repo_url: str, diff --git a/tests/test_plugin_endpoint.py b/tests/test_plugin_endpoint.py index ad784973..99707a37 100644 --- a/tests/test_plugin_endpoint.py +++ b/tests/test_plugin_endpoint.py @@ -1,6 +1,7 @@ import asyncio from unittest.mock import ANY, AsyncMock, MagicMock, patch +from app.api.endpoints import plugin as plugin_endpoint from app import schemas from app.api.endpoints.plugin import plugin_history from app.api.endpoints.plugin import plugin_releases @@ -68,6 +69,38 @@ def test_plugin_history_returns_installed_plugin_when_remote_missing(): assert result.history == {} +def test_plugin_history_uses_installed_repo_without_refreshing_all_markets(): + """ + 已安装插件记录了来源仓库时，更新说明只刷新该仓库，避免弹窗触发全市场慢刷新。 + """ + installed_plugin = schemas.Plugin( + id="DemoPlugin", + plugin_name="Demo Plugin", + plugin_version="1.0.0", + repo_url="https://github.com/demo/plugins", + installed=True, + ) + market_plugin = schemas.Plugin( + id="DemoPlugin", + repo_url="https://github.com/demo/plugins", + history={"v1.1.0": "- 新增更新说明"}, + ) + plugin_manager = MagicMock() + plugin_manager.get_local_plugins.return_value = [installed_plugin] + plugin_manager.get_local_repo_plugins.return_value = [] + plugin_manager.async_get_plugins_from_market = AsyncMock(return_value=[market_plugin]) + plugin_manager.async_get_online_plugins = AsyncMock(return_value=[]) + + with patch("app.api.endpoints.plugin.PluginManager", return_value=plugin_manager): + result = asyncio.run(plugin_history("DemoPlugin", None, True)) + + assert result.history == {"v1.1.0": "- 新增更新说明"} + plugin_manager.async_get_plugins_from_market.assert_awaited_once_with( + "https://github.com/demo/plugins", settings.VERSION_FLAG, True + ) + plugin_manager.async_get_online_plugins.assert_not_awaited() + + def test_plugin_releases_returns_supported_versions_with_latest_and_current(monkeypatch): """ release 列表接口返回可安装版本，并标记当前 package 最新版本与本地已安装版本。 @@ -154,6 +187,7 @@ def test_plugin_releases_falls_back_to_compatible_base_package(monkeypatch): ) plugin_manager.get_local_plugin_version.return_value = None plugin_helper = MagicMock() + plugin_helper.async_has_plugin_release_cache = AsyncMock(return_value=False) plugin_helper.async_get_plugin_release_versions = AsyncMock(return_value=[]) with ( @@ -185,6 +219,7 @@ def test_plugin_releases_uses_force_refresh_for_market_metadata(monkeypatch): plugin_manager.async_get_plugins_from_market = AsyncMock(return_value=[market_plugin]) plugin_manager.get_local_plugin_version.return_value = None plugin_helper = MagicMock() + plugin_helper.async_has_plugin_release_cache = AsyncMock(return_value=False) plugin_helper.async_get_plugin_release_versions = AsyncMock(return_value=[]) with ( @@ -203,6 +238,100 @@ def test_plugin_releases_uses_force_refresh_for_market_metadata(monkeypatch): ) +def test_plugin_releases_force_uses_cached_release_response_and_schedules_refresh(monkeypatch): + """ + 手动刷新时 package 元数据仍强刷，但 Release 明细先读缓存并后台刷新，避免弹窗阻塞。 + """ + from app.core.cache import is_fresh + + market_plugin = schemas.Plugin( + id="DemoPlugin", + plugin_version="1.2.3", + repo_url="https://github.com/demo/plugins", + release=True, + ) + plugin_manager = MagicMock() + plugin_manager.async_get_plugins_from_market = AsyncMock(return_value=[market_plugin]) + plugin_manager.get_local_plugin_version.return_value = None + fresh_states = [] + plugin_helper = MagicMock() + plugin_helper.async_has_plugin_release_cache = AsyncMock(return_value=True) + + async def fake_releases(*_args): + fresh_states.append(is_fresh()) + return [ + { + "version": "1.2.3", + "tag_name": "DemoPlugin_v1.2.3", + "asset_name": "demoplugin_v1.2.3.zip", + } + ] + + plugin_helper.async_get_plugin_release_versions = fake_releases + scheduled = [] + + def fake_schedule(plugin_id, repo_url): + scheduled.append((plugin_id, repo_url)) + + with ( + patch("app.api.endpoints.plugin.PluginManager", return_value=plugin_manager), + patch("app.api.endpoints.plugin.PluginHelper", return_value=plugin_helper), + patch.object(plugin_endpoint, "_schedule_plugin_release_refresh", fake_schedule), + ): + result = asyncio.run(plugin_releases("DemoPlugin", None, "https://github.com/demo/plugins", True)) + + assert result["release_supported"] is True + assert fresh_states == [False] + assert scheduled == [("DemoPlugin", "https://github.com/demo/plugins")] + plugin_helper.async_has_plugin_release_cache.assert_awaited_once_with( + "https://github.com/demo/plugins" + ) + plugin_manager.async_get_plugins_from_market.assert_awaited_once_with( + "https://github.com/demo/plugins", settings.VERSION_FLAG, True + ) + + +def test_plugin_releases_force_skips_background_refresh_without_release_cache(monkeypatch): + """ + 冷缓存 force 请求已在响应路径读取 Release，不能马上再启动一次重复强刷。 + """ + market_plugin = schemas.Plugin( + id="DemoPlugin", + plugin_version="1.2.3", + repo_url="https://github.com/demo/plugins", + release=True, + ) + plugin_manager = MagicMock() + plugin_manager.async_get_plugins_from_market = AsyncMock(return_value=[market_plugin]) + plugin_manager.get_local_plugin_version.return_value = None + plugin_helper = MagicMock() + plugin_helper.async_has_plugin_release_cache = AsyncMock(return_value=False) + plugin_helper.async_get_plugin_release_versions = AsyncMock(return_value=[ + { + "version": "1.2.3", + "tag_name": "DemoPlugin_v1.2.3", + "asset_name": "demoplugin_v1.2.3.zip", + } + ]) + scheduled = [] + + def fake_schedule(plugin_id, repo_url): + scheduled.append((plugin_id, repo_url)) + + with ( + patch("app.api.endpoints.plugin.PluginManager", return_value=plugin_manager), + patch("app.api.endpoints.plugin.PluginHelper", return_value=plugin_helper), + patch.object(plugin_endpoint, "_schedule_plugin_release_refresh", fake_schedule), + ): + result = asyncio.run(plugin_releases("DemoPlugin", None, "https://github.com/demo/plugins", True)) + + assert result["release_supported"] is True + assert scheduled == [] + plugin_helper.async_has_plugin_release_cache.assert_awaited_once_with( + "https://github.com/demo/plugins" + ) + + def test_plugin_releases_hides_items_when_market_plugin_does_not_enable_release(monkeypatch): """ 接口是否支持 Release 安装要与当前 package 的 release 声明保持一致。 diff --git a/tests/test_plugin_helper.py b/tests/test_plugin_helper.py index 09acb137..eeb6eed0 100644 --- a/tests/test_plugin_helper.py +++ b/tests/test_plugin_helper.py @@ -499,6 +499,109 @@ class TestPluginHelper: assert [item["version"] for item in cached_result] == ["1.2.3"] assert request_count == 2 + def test_async_normal_release_read_does_not_wait_for_pending_force_refresh(self, monkeypatch): + """普通读取遇到后台强刷时仍优先返回已有缓存，避免页面响应被强刷阻塞。""" + try: + from app.core.cache import async_fresh + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + old_payload = [{ + "tag_name": "DemoPlugin_v1.2.2", + "assets": [{"name": "demoplugin_v1.2.2.zip", "id": 1}], + }] + fresh_payload = [{ + "tag_name": "DemoPlugin_v1.2.3", + "assets": [{"name": "demoplugin_v1.2.3.zip", "id": 2}], + }] + force_request_started = asyncio.Event() + release_force_request = asyncio.Event() + request_count = 0 + + async def fake_request(*_args, **_kwargs): + nonlocal request_count + request_count += 1 + if request_count == 1: + return _FakeTextResponse(200, old_payload) + force_request_started.set() + await release_force_request.wait() + return _FakeTextResponse(200, fresh_payload) + + async def run_test(): + helper = PluginHelper() + await helper.async_get_plugin_release_versions.cache_clear() + monkeypatch.setattr(helper, "_PluginHelper__async_request_with_fallback", fake_request) + initial = await helper.async_get_plugin_release_versions("DemoPlugin", REPO_URL) + async with async_fresh(True): + force_task = asyncio.create_task( + helper.async_get_plugin_release_versions("DemoPlugin", REPO_URL) + ) + await force_request_started.wait() + normal_task = asyncio.create_task( + helper.async_get_plugin_release_versions("DemoPlugin", REPO_URL) + ) + normal_before_force_finished = await asyncio.wait_for(normal_task, timeout=1) + force_done_before_normal_finished = force_task.done() + release_force_request.set() + force_result = await force_task + cached_result = await helper.async_get_plugin_release_versions("DemoPlugin", REPO_URL) + return ( + initial, + force_done_before_normal_finished, + normal_before_force_finished, + force_result, + cached_result, + ) + + ( + initial, + force_done_before_normal_finished, + normal_before_force_finished, + force_result, + cached_result, + ) = asyncio.run(run_test()) + + assert [item["version"] for item in initial] == ["1.2.2"] + assert force_done_before_normal_finished is False + assert [item["version"] for item in normal_before_force_finished] == ["1.2.2"] + assert [item["version"] for item in force_result] == ["1.2.3"] + assert [item["version"] for item in cached_result] == ["1.2.3"] + assert request_count == 2 + + def test_async_has_plugin_release_cache_reflects_repository_cache(self, monkeypatch): + """Release 缓存探针只判断仓库级缓存是否已经存在，不触发网络请求。""" + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + payload = [{ + "tag_name": "DemoPlugin_v1.2.3", + "assets": [{"name": "demoplugin_v1.2.3.zip", "id": 1}], + }] + request_count = 0 + + async def fake_request(*_args, **_kwargs): + nonlocal request_count + request_count += 1 + return _FakeTextResponse(200, payload) + + async def run_test(): + helper = PluginHelper() + await helper.async_get_plugin_release_versions.cache_clear() + monkeypatch.setattr(helper, "_PluginHelper__async_request_with_fallback", fake_request) + before = await helper.async_has_plugin_release_cache(REPO_URL) + await helper.async_get_plugin_release_versions("DemoPlugin", REPO_URL) + after = await helper.async_has_plugin_release_cache(REPO_URL) + return before, after + + before, after = asyncio.run(run_test()) + + assert before is False + assert after is True + assert request_count == 1 + def test_failed_forced_release_refresh_preserves_cached_repository_payload(self, monkeypatch): """GitHub 强刷失败时不以空值覆盖该仓库已有 Release 缓存。""" try: