diff --git a/app/adapters/external/market.py b/app/adapters/external/market.py index 9f8cf6ddf..f8dfb9c3b 100644 --- a/app/adapters/external/market.py +++ b/app/adapters/external/market.py @@ -15,7 +15,7 @@ import traceback import uuid import zipfile from pathlib import Path, PurePosixPath, PureWindowsPath -from typing import Dict, List, Optional, Tuple, Set, Callable, Awaitable, Sequence +from typing import Dict, List, Optional, Tuple, Set, Callable, Awaitable, Iterator, Sequence from urllib.parse import parse_qs, quote, unquote, urlparse, urlsplit import aiofiles @@ -739,6 +739,53 @@ class PluginHelper(metaclass=WeakSingleton): if isinstance(release_info, dict) ] + @classmethod + def _iter_plugin_release_page_requests( + cls, + repo_url: str, + ) -> Iterator[Tuple[str, dict]]: + """按需生成仓库 Release 分页请求，统一仓库解析、请求头和页数上限。""" + if not repo_url: + return + + user, repo = cls.get_repo_info(repo_url) + if not user or not repo: + return + + user_repo = f"{user}/{repo}" + headers = settings.REPO_GITHUB_HEADERS(repo=user_repo) + for page in range(1, 11): + release_api = ( + f"https://api.github.com/repos/{user_repo}/releases" + f"?per_page=100&page={page}" + ) + yield cls.__append_cache_buster(release_api), headers + + @classmethod + def _merge_plugin_release_page( + cls, + repo_url: str, + response, + releases: List[dict], + ) -> Optional[bool]: + """合并一页 Release 响应；返回真继续、假结束，None 表示整次读取失败。""" + if response is None or response.status_code != 200: + return None + + try: + payload = response.json() + except Exception as error: + logger.error(f"解析插件仓库 {repo_url} Release 列表失败：{error}") + return None + + if not payload: + return False + if not isinstance(payload, list): + return None + + releases.extend(cls.__normalize_plugin_release_response(payload)) + return len(payload) >= 100 + @cached(maxsize=128, ttl=1800) def get_plugins(self, repo_url: str, package_version: Optional[str] = None) -> Optional[Dict[str, dict]]: @@ -761,39 +808,25 @@ class PluginHelper(metaclass=WeakSingleton): """ 按仓库获取 GitHub Release 原始分页数据，供仓库内所有插件共享。 """ - if not repo_url: - return [] - - user, repo = self.get_repo_info(repo_url) - if not user or not repo: - return [] - - user_repo = f"{user}/{repo}" releases = [] - for page in range(1, 11): - release_api = f"https://api.github.com/repos/{user_repo}/releases?per_page=100&page={page}" - release_api = self.__append_cache_buster(release_api) + for release_api, headers in self._iter_plugin_release_page_requests( + repo_url + ): res = self.__request_with_fallback( release_api, - headers=settings.REPO_GITHUB_HEADERS(repo=user_repo), + headers=headers, timeout=30, is_api=True, ) - if res is None or res.status_code != 200: - return None - - try: - payload = res.json() - if not payload: - break - if not isinstance(payload, list): - return None - releases.extend(self.__normalize_plugin_release_response(payload)) - if len(payload) < 100: - break - except Exception as e: - logger.error(f"解析插件仓库 {repo_url} Release 列表失败：{e}") + should_continue = self._merge_plugin_release_page( + repo_url, + res, + releases, + ) + if should_continue is None: return None + if not should_continue: + break return releases def get_plugin_release_versions(self, pid: str, repo_url: str) -> List[dict]: @@ -2295,39 +2328,25 @@ class PluginHelper(metaclass=WeakSingleton): """ 异步按仓库获取 GitHub Release 原始分页数据。 """ - if not repo_url: - return [] - - user, repo = self.get_repo_info(repo_url) - if not user or not repo: - return [] - - user_repo = f"{user}/{repo}" releases = [] - for page in range(1, 11): - release_api = f"https://api.github.com/repos/{user_repo}/releases?per_page=100&page={page}" - release_api = self.__append_cache_buster(release_api) + for release_api, headers in self._iter_plugin_release_page_requests( + repo_url + ): res = await self.__async_request_with_fallback( release_api, - headers=settings.REPO_GITHUB_HEADERS(repo=user_repo), + headers=headers, timeout=30, is_api=True, ) - if res is None or res.status_code != 200: - return None - - try: - payload = res.json() - if not payload: - break - if not isinstance(payload, list): - return None - releases.extend(self.__normalize_plugin_release_response(payload)) - if len(payload) < 100: - break - except Exception as e: - logger.error(f"解析插件仓库 {repo_url} Release 列表失败：{e}") + should_continue = self._merge_plugin_release_page( + repo_url, + res, + releases, + ) + if should_continue is None: return None + if not should_continue: + break return releases async def async_get_plugin_release_versions(self, pid: str, repo_url: str) -> List[dict]: diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index c2551a9cf..f2b4ceb4d 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -6,7 +6,7 @@ > 审计范围：宿主后端；排除 `app/plugins/**` 运行时插件副本 > 规范优先级：`AGENTS.md` 与 `docs/rules/` 高于本文 > 相关文档：`docs/architecture-overview.md`、`docs/refactor/backend-architecture-governance.md`、`docs/refactor/backend-module-refactor-compatibility.md` -> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口；阶段 4 已统一七种消息渠道的宿主回环与后台执行边界；阶段 5 已补齐事件窗口聚合任务的生命周期所有权；阶段 6 已统一插件文件操作的取消完成语义；阶段 7 已统一插件协程补偿的终态等待；阶段 8 已统一宿主同步函数的异步线程池入口；阶段 9 已统一工作流运行时的宿主获取路径；阶段 10 已统一模块、插件与调度运行时的显式 getter 调用；阶段 11 已清除系统配置 getter 的 Oper 形别名；阶段 12 已完成工作流域的显式 Chain 数据端口迁移；阶段 13 已收口用户、交互与消息链的数据端口；阶段 14 已收口音乐订阅数据端口；阶段 15 已收口站点数据端口；阶段 16 已收口媒体服务器数据端口；阶段 17 已收口下载数据端口；阶段 18 已收口主订阅数据端口；阶段 19 已收口整理数据端口；阶段 20 已收口 Agent 数据端口；阶段 21 已收口监控历史端口；阶段 22 已统一服务配置应用边界；阶段 23 已补齐媒体服务器 API 遗留的类形配置读取路径；阶段 24 已清除 Scheduler 内部无 owner 的协程提交双轨；阶段 25 已补齐 TaskRegistry 跨线程 owner 并迁移整理 AI 接管；阶段 26 已统一 Agent 会话清理提交；阶段 27 已统一历史 AI 进度 owner；阶段 28 已托管旧插件订阅统计线程；阶段 29 已统一 Emby 系条目转换并清零重复代码白名单；阶段 30 已收口插件市场请求级子任务；阶段 31 已托管搜索 AI 推荐任务；阶段 32 已清除事件调度器绕过生命周期 owner 的投递回退；阶段 33 已统一宿主 Agent 运行时的获取路径；阶段 34 已统一 durable-required 事件与 Outbox topic 事实源；阶段 35 已统一 LLM provider 管理 API 的运行时解析路径；阶段 36 已统一 WebAgent 音频能力访问边界；阶段 37 已统一插件输入事件发布路径；阶段 38 已统一 WebAgent 通知事件监听与队列边界；阶段 39 已补齐搜索 SSE 断线时的上游任务清理；阶段 40 已补齐异步防抖取消的终态所有权；阶段 41 已统一优雅重启兜底线程的唯一所有权；阶段 42 已补齐 Telegram typing 的多实例隔离和终态 owner；阶段 43 已统一 Discord typing 的异步 owner 和 shutdown 收尾；阶段 44 已清除 WebAgent 测试临时事件循环提前关闭产生的 CI 红注解；阶段 45 已统一影视与字幕搜索的请求级逐页任务编排；阶段 46 已收口启动性能门禁的托管 runner 假失败与诊断输出；阶段 47 已补齐 Agent 渠道流式刷新任务的重入 owner；阶段 48 已统一工件上传 action 的 Node 24 主版本；阶段 49 已统一插件安装的同步/异步代际解析事实源；阶段 50 已统一插件市场 GitHub 请求降级策略；阶段 51 已统一插件索引请求与响应三态策略。 +> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口；阶段 4 已统一七种消息渠道的宿主回环与后台执行边界；阶段 5 已补齐事件窗口聚合任务的生命周期所有权；阶段 6 已统一插件文件操作的取消完成语义；阶段 7 已统一插件协程补偿的终态等待；阶段 8 已统一宿主同步函数的异步线程池入口；阶段 9 已统一工作流运行时的宿主获取路径；阶段 10 已统一模块、插件与调度运行时的显式 getter 调用；阶段 11 已清除系统配置 getter 的 Oper 形别名；阶段 12 已完成工作流域的显式 Chain 数据端口迁移；阶段 13 已收口用户、交互与消息链的数据端口；阶段 14 已收口音乐订阅数据端口；阶段 15 已收口站点数据端口；阶段 16 已收口媒体服务器数据端口；阶段 17 已收口下载数据端口；阶段 18 已收口主订阅数据端口；阶段 19 已收口整理数据端口；阶段 20 已收口 Agent 数据端口；阶段 21 已收口监控历史端口；阶段 22 已统一服务配置应用边界；阶段 23 已补齐媒体服务器 API 遗留的类形配置读取路径；阶段 24 已清除 Scheduler 内部无 owner 的协程提交双轨；阶段 25 已补齐 TaskRegistry 跨线程 owner 并迁移整理 AI 接管；阶段 26 已统一 Agent 会话清理提交；阶段 27 已统一历史 AI 进度 owner；阶段 28 已托管旧插件订阅统计线程；阶段 29 已统一 Emby 系条目转换并清零重复代码白名单；阶段 30 已收口插件市场请求级子任务；阶段 31 已托管搜索 AI 推荐任务；阶段 32 已清除事件调度器绕过生命周期 owner 的投递回退；阶段 33 已统一宿主 Agent 运行时的获取路径；阶段 34 已统一 durable-required 事件与 Outbox topic 事实源；阶段 35 已统一 LLM provider 管理 API 的运行时解析路径；阶段 36 已统一 WebAgent 音频能力访问边界；阶段 37 已统一插件输入事件发布路径；阶段 38 已统一 WebAgent 通知事件监听与队列边界；阶段 39 已补齐搜索 SSE 断线时的上游任务清理；阶段 40 已补齐异步防抖取消的终态所有权；阶段 41 已统一优雅重启兜底线程的唯一所有权；阶段 42 已补齐 Telegram typing 的多实例隔离和终态 owner；阶段 43 已统一 Discord typing 的异步 owner 和 shutdown 收尾；阶段 44 已清除 WebAgent 测试临时事件循环提前关闭产生的 CI 红注解；阶段 45 已统一影视与字幕搜索的请求级逐页任务编排；阶段 46 已收口启动性能门禁的托管 runner 假失败与诊断输出；阶段 47 已补齐 Agent 渠道流式刷新任务的重入 owner；阶段 48 已统一工件上传 action 的 Node 24 主版本；阶段 49 已统一插件安装的同步/异步代际解析事实源；阶段 50 已统一插件市场 GitHub 请求降级策略；阶段 51 已统一插件索引请求与响应三态策略；阶段 52 已统一插件 Release 分页策略。 ## 当前复核结论（2026-08-24） @@ -541,6 +541,16 @@ - 回归测试覆盖两条入口的 URL、请求头和结果一致，并锁定 404、非 200、有效字典、非法 JSON 和非字典 JSON 的既有区别。缓存装饰器、公开签名、安装返回语义、SDK/Compat、V1/V2/V3 插件 ABI 均未改变；未修改插件仓。 +### 长期整改阶段 52：插件 Release 分页策略统一（2026-08-24） + +- 同步 `_get_plugin_repo_releases()` 与异步版本原先各自维护仓库解析、十页上限、请求头、HTTP 状态、JSON + 类型校验、缓存快照规范化以及满 100 条才续页的规则；Release 安装能力因此有两份易漂移的事实源。 +- `_iter_plugin_release_page_requests()` 现在按需生成唯一分页计划，`_merge_plugin_release_page()` 唯一解释页面 + 响应并返回继续、自然结束或整次失败三态；同步与异步方法只保留各自 HTTP await 差异。 +- 回归测试覆盖两条入口的两页 URL、参数和 101 条规范化结果一致，并锁定空页、短页、满页、坏状态、坏 + payload 与 JSON 异常。仓库级共享缓存、强刷合并、TaskRegistry owner、取消传播、公开 Release 列表、 + SDK/Compat 和 V1/V2/V3 插件 ABI 均未改变；未修改插件仓。 + ### 总体判断 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**： diff --git a/tests/test_plugin_release_page_policy.py b/tests/test_plugin_release_page_policy.py new file mode 100644 index 000000000..c59c725ad --- /dev/null +++ b/tests/test_plugin_release_page_policy.py @@ -0,0 +1,112 @@ +"""插件 Release 同步与异步分页策略一致性测试。""" + +import pytest + +from app.adapters.external.market import PluginHelper + + +class _PageResponse: + """提供同步与异步分页入口共同消费的最小响应合同。""" + + def __init__(self, payload, status_code: int = 200) -> None: + """保存待返回的 JSON 数据和 HTTP 状态。""" + self._payload = payload + self.status_code = status_code + + def json(self): + """返回当前页预置的 JSON 数据。""" + if isinstance(self._payload, Exception): + raise self._payload + return self._payload + + +def _release_payload(size: int, *, offset: int = 0) -> list[dict]: + """生成可区分页次的 GitHub Release 最小响应。""" + return [ + { + "tag_name": f"DemoPlugin_v1.0.{offset + index}", + "assets": [ + {"name": f"demoplugin_v1.0.{offset + index}.zip"}, + ], + } + for index in range(size) + ] + + +@pytest.mark.asyncio +async def test_sync_and_async_release_pages_share_policy(monkeypatch) -> None: + """同步与异步分页必须请求相同页并产生相同规范化仓库快照。""" + helper = PluginHelper() + repo_url = "https://github.com/policy-owner/release-repository" + responses = [ + _PageResponse(_release_payload(100)), + _PageResponse(_release_payload(1, offset=100)), + ] + sync_requests: list[tuple[str, dict]] = [] + async_requests: list[tuple[str, dict]] = [] + + def sync_request(url: str, **kwargs): + """记录同步分页请求并按页返回响应。""" + sync_requests.append((url, kwargs)) + return responses[len(sync_requests) - 1] + + async def async_request(url: str, **kwargs): + """记录异步分页请求并采用相同响应序列。""" + async_requests.append((url, kwargs)) + return responses[len(async_requests) - 1] + + monkeypatch.setattr( + helper, + "_PluginHelper__request_with_fallback", + sync_request, + ) + monkeypatch.setattr( + helper, + "_PluginHelper__async_request_with_fallback", + async_request, + ) + + helper._get_plugin_repo_releases.cache_clear() + await helper._async_get_plugin_repo_releases.cache_clear() + sync_result = helper._get_plugin_repo_releases(repo_url) + # 两个装饰器共享仓库级缓存；清除后再覆盖异步网络入口。 + await helper._async_get_plugin_repo_releases.cache_clear() + async_result = await helper._async_get_plugin_repo_releases(repo_url) + + assert sync_result == async_result + assert len(sync_result or []) == 101 + assert sync_requests == async_requests + assert [request[0].rsplit("page=", 1)[1] for request in sync_requests] == [ + "1", + "2", + ] + + +@pytest.mark.parametrize( + ("response", "expected", "release_count"), + [ + (None, None, 0), + (_PageResponse([], 200), False, 0), + (_PageResponse([{"tag_name": "DemoPlugin_v1.0.0"}], 200), False, 1), + (_PageResponse(_release_payload(100), 200), True, 100), + (_PageResponse({"message": "bad payload"}, 200), None, 0), + (_PageResponse([], 503), None, 0), + (_PageResponse(ValueError("invalid json"), 200), None, 0), + ], +) +def test_release_page_merge_preserves_stop_and_failure_contract( + response, + expected: bool | None, + release_count: int, +) -> None: + """统一分页解析必须区分继续、自然结束和整次仓库读取失败。""" + releases: list[dict] = [] + + result = PluginHelper._merge_plugin_release_page( + "https://github.com/policy-owner/release-repository", + response, + releases, + ) + + assert result is expected + assert len(releases) == release_count