diff --git a/app/adapters/network/http.py b/app/adapters/network/http.py index 7f578fbc6..d032a3b34 100644 --- a/app/adapters/network/http.py +++ b/app/adapters/network/http.py @@ -25,6 +25,8 @@ from requests.adapters import HTTPAdapter from urllib3.exceptions import InsecureRequestWarning from app.runtime.correlation import with_correlation_header +from app.runtime.log import logger +from app.runtime.tasks import get_task_registry urllib3.disable_warnings(InsecureRequestWarning) @@ -143,21 +145,25 @@ class _NonClosingTransportProxy: 包装共享传输对象，转发请求但不转发局部关闭调用。 每次请求创建的 AsyncClient 退出时保留底层连接池；连接池最终由 - aclose_shared_async_transports 统一关闭。 + aclose_shared_async_transports 统一关闭；连续取池超时的旧池在借用结束后释放。 """ - __slots__ = ("_wrapped",) + __slots__ = ("_wrapped", "_recovery") - def __init__(self, wrapped: httpx2.AsyncBaseTransport): + def __init__(self, wrapped: httpx2.AsyncBaseTransport) -> None: """ 保存由进程统一管理生命周期的底层传输对象。 """ self._wrapped = wrapped + with _shared_async_transports_lock: + self._recovery = _shared_transport_recovery.get(wrapped) - async def __aenter__(self) -> "_NonClosingTransportProxy": # pragma: no cover - 简单转发 + async def __aenter__(self) -> "_NonClosingTransportProxy": """ 进入局部上下文时继续复用共享传输对象。 """ + if self._recovery is not None: + self._recovery.users += 1 return self async def __aexit__( @@ -169,8 +175,10 @@ class _NonClosingTransportProxy: """ 退出局部上下文时保留由进程统一管理的连接池。 """ - # 故意 no-op：不向底层 transport 传播 __aexit__，避免连接池被清空 - return None + if self._recovery is not None: + self._recovery.users -= 1 + if self._recovery.retired and self._recovery.users == 0: + _schedule_retired_transport_close(self._wrapped) async def aclose(self) -> None: """ @@ -183,7 +191,19 @@ class _NonClosingTransportProxy: """ 将异步请求转发给共享底层传输。 """ - return await self._wrapped.handle_async_request(request) + try: + response = await self._wrapped.handle_async_request(request) + except httpx2.PoolTimeout: + if self._recovery is not None: + _record_shared_pool_timeout(self._wrapped, self._recovery) + raise + except httpx2.RequestError: + if self._recovery is not None: + self._recovery.failures = 0 + raise + if self._recovery is not None: + self._recovery.failures = 0 + return response _SharedTransportKey = Tuple[ @@ -207,6 +227,31 @@ _SharedTransportKey = Tuple[ _shared_async_transports: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, collections.OrderedDict[_SharedTransportKey, httpx2.AsyncHTTPTransport]] = weakref.WeakKeyDictionary() # 不同线程各自驱动的事件循环并发首次写入外层弱字典时，需要互斥保护 _shared_async_transports_lock = threading.Lock() +# 连续三次取池超时才更换池，普通连接/读取超时不触发恢复。 +_SHARED_POOL_TIMEOUT_THRESHOLD = 3 + + +class _SharedTransportRecovery: + """ + 记录单个共享池的失败与借用状态，退役后等局部客户端退出再关闭。 + """ + + def __init__(self, key: _SharedTransportKey) -> None: + """ + 初始化池状态，以配置键约束最多两代在途连接池。 + + :param key: 当前池配置，用于识别同配置下仍未关闭的退役池 + """ + self.key = key + self.users = 0 + self.failures = 0 + self.retired = False + self.closing = False + self.closed = False + + +_shared_transport_recovery: weakref.WeakKeyDictionary[httpx2.AsyncBaseTransport, _SharedTransportRecovery] = weakref.WeakKeyDictionary() +_retired_async_transports: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, set[httpx2.AsyncBaseTransport]] = weakref.WeakKeyDictionary() # 每个事件循环允许的最大共享 transport 桶数；超出后按 LRU 淘汰最久未用桶。 _MAX_SHARED_TRANSPORTS_PER_LOOP = 32 # 默认的最大 keep-alive 连接数 @@ -283,7 +328,107 @@ def _discard_pending_eviction_task(task: asyncio.Task[Any]) -> None: with _shared_async_transports_lock: _pending_eviction_tasks.discard(task) if not task.cancelled(): - task.exception() + error = task.exception() + if error is not None: + logger.warning(f"共享 HTTP 连接池关闭失败：{type(error).__name__}") + + +def _create_shared_async_transport( + key: _SharedTransportKey, +) -> httpx2.AsyncHTTPTransport: + """ + 按已有配置创建新池，不复用故障池的内部连接或请求队列。 + """ + proxy, verify, http2, trust_env, max_keepalive, max_connections, expiry = key + transport = httpx2.AsyncHTTPTransport( + http2=http2, proxy=proxy, verify=verify, trust_env=trust_env, + limits=httpx2.Limits( + max_keepalive_connections=max_keepalive, + max_connections=max_connections, + keepalive_expiry=expiry, + ), + ) + with _shared_async_transports_lock: + _shared_transport_recovery[transport] = _SharedTransportRecovery(key) + return transport + + +async def _close_retired_async_transport(transport: httpx2.AsyncBaseTransport) -> None: + """ + 关闭退役池并释放恢复状态；关闭异常仍由托管任务收集。 + """ + try: + await transport.aclose() + except BaseException: + with _shared_async_transports_lock: + _shared_transport_recovery[transport].closing = False + raise + else: + with _shared_async_transports_lock: + recovery = _shared_transport_recovery.get(transport) + if recovery is not None: + recovery.closed = True + retired = _retired_async_transports.get(asyncio.get_running_loop()) + if retired is not None: + retired.discard(transport) + + +def _schedule_retired_transport_close(transport: httpx2.AsyncBaseTransport) -> None: + """ + 将退役池关闭任务交给已有淘汰任务集合持有，避免取消调用方中断释放。 + """ + with _shared_async_transports_lock: + recovery = _shared_transport_recovery[transport] + if recovery.closing or recovery.closed: + return + recovery.closing = True + try: + task = get_task_registry().create( + _close_retired_async_transport(transport), + owner="http.transport.retire", + cancel_on_shutdown=False, + ) + except RuntimeError: + recovery.closing = False + logger.warning("后台任务登记器已停止，退役 HTTP 连接池保留至宿主关闭时回收") + return + _pending_eviction_tasks.add(task) + task.add_done_callback(_discard_pending_eviction_task) + + +def _record_shared_pool_timeout( + transport: httpx2.AsyncBaseTransport, recovery: _SharedTransportRecovery, +) -> None: + """ + 连续取池失败时只替换仍为当前代的池；旧代请求退出前不重复轮换。 + + 不重放当前请求，因此非幂等请求仍只发送一次。 + """ + recovery.failures += 1 + if recovery.retired or recovery.failures < _SHARED_POOL_TIMEOUT_THRESHOLD: + return + loop = asyncio.get_running_loop() + with _shared_async_transports_lock: + per_loop = _shared_async_transports.get(loop) + if any( + _shared_transport_recovery[old].key == recovery.key + for old in _retired_async_transports.get(loop, ()) + ): + return + if per_loop is None: + return + for key, current in per_loop.items(): + if current is transport: + replacement = _create_shared_async_transport(key) + per_loop[key] = replacement + recovery.retired = True + with _shared_async_transports_lock: + _retired_async_transports.setdefault(loop, set()).add(transport) + logger.warning( + f"共享 HTTP 连接池连续 {recovery.failures} 次 PoolTimeout，" + f"后续请求切换新池，旧池等待 {recovery.users} 个在途客户端退出后关闭" + ) + return def _get_shared_async_transport( @@ -335,17 +480,7 @@ def _get_shared_async_transport( return transport # 首次见到这个配置，创建新的共享 transport 桶 - transport = httpx2.AsyncHTTPTransport( - http2=http2, - proxy=proxy, - verify=verify, - trust_env=trust_env, - limits=httpx2.Limits( - max_keepalive_connections=max_keepalive_connections, - max_connections=max_connections, - keepalive_expiry=keepalive_expiry, - ), - ) + transport = _create_shared_async_transport(key) per_loop[key] = transport # LRU 淘汰：超出上限时关闭并移除最久未用桶 @@ -376,15 +511,22 @@ async def aclose_shared_async_transports() -> None: # 弹出而非 get+clear，避免外层 dict 残留空 OrderedDict 占位 with _shared_async_transports_lock: per_loop = _shared_async_transports.pop(loop, None) + retired = _retired_async_transports.get(loop, set()) pending_evictions = [ task for task in _pending_eviction_tasks if task.get_loop() is loop ] + retiring = [ + transport for transport in retired + if not _shared_transport_recovery[transport].closing + ] + for transport in retiring: + _shared_transport_recovery[transport].closing = True transports = list(per_loop.values()) if per_loop else [] if per_loop: per_loop.clear() - if not transports and not pending_evictions: + if not transports and not pending_evictions and not retiring: return # 并行关闭：每个 transport 的 TLS close_notify 各占一个 RTT， # 顺序等待会线性放大 shutdown 耗时；return_exceptions 让单点失败 @@ -392,6 +534,7 @@ async def aclose_shared_async_transports() -> None: await asyncio.gather( *pending_evictions, *(t.aclose() for t in transports), + *(_close_retired_async_transport(t) for t in retiring), return_exceptions=True, ) with _shared_async_transports_lock: diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index dcd9bc181..392b19202 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -763,7 +763,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 1123 | -| 内部导入边 | 9,584 | +| 内部导入边 | 9,586 | | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 49（债务已清零，49 条精确 containment） | @@ -778,7 +778,7 @@ Outbox 按 topic 有界并行恢复复用 `ThreadHelper`，生命周期关闭通 外部 MCP schema 紧凑投影由纯函数模块 `app.agent.tools.projection` 拥有，仅在请求头显式 opt-in 后应用；新增 1 个模块和 1 条合法端点依赖边，未新增 SCC。`list_exposed_tools` 返回类型注解使端点类型债务减少 5 项。 -目录监控成功通知在批次完成后复用主事件循环开启静默窗口，新增 2 条合法 Chain 到 Runtime 依赖边（含父包展开），未新增模块或 SCC。成功通知聚合复用既有 owner 和领域模型，统一交由通知模板渲染；移除汇总发送端的模板上下文直连后，原聚合实现的 14 条新增依赖边减为 12 条，未新增模块或 SCC。社区勋章名单读取复用 `app.foundation.url` 标准化 GitHub 镜像地址，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责，预览证据由 `app.application.transfer.projection` 只读投影；宿主依赖边现为 9,584；站点 API 依赖显式导入 SiteQueryPort 与 SiteStagingPort，新增 1 条合法类型边；订阅独立日志复用运行时上下文、系统日志端口与资源鉴权，新增 19 条合法依赖边（含父包展开），未新增模块或 SCC；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条 Schema 到 Domain 的合法依赖边（含父包展开）；种子下载限流窗口由 `app.domain.torrent` 定义，失败冷却 owner 复用同一语义，新增 1 条合法 Chain 到 Domain 的导入边；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条模块导入边；资源过滤动作 `app.workflow.actions.filter_torrents` 与动作包共用的领域还原函数同样把 Schema 资源还原为领域对象后再交给过滤链，新增 6 条模块导入边。音乐来源链与识别缓存复用领域身份核验，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端，经 `RequestUtils` 出站，宿主不再导入 pyTelegramBotAPI，启用 Telegram 的实例也不会因该 SDK 加载 redis；该依赖暂时保留给插件使用。Web Push 协议（aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递）由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现，消息端点与 `WebPushModule` 共用，宿主不再导入 pywebpush 及其连带的 aiohttp；该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端（经 `RequestUtils` 出站）与网络适配器 `app.adapters.network.feishu` 事件长连接（websocket-client 同步收发 pbbp2 帧），宿主不再导入 lark-oapi 及其上万个生成模块，长连接也不再依赖 SDK 的模块级事件循环；插件经 `app.sdk.feishu` 复用同一长连接，lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放，默认关闭，未开放时不生成文档。当前 `app.startup.lifecycle` 为 587、`app.factory` 为 600、`app.main` 为 602。专辑订阅入库对账新增 `app.chain.subscribe.music` 一个职责 owner；按注入的下载/整理历史 Port 读取事实，复用音频标签/CUE 读取和领域覆盖规则，依赖图增加 18 条合法边，不引入跨层数据库或 Adapter 直连。订阅业务活动新增 `app.application.subscription.activity` 及对应 DB Model、Oper、Adapter 四个模块，三入口与功能前提交对照均只增加这四个模块；性能基线仅同步结构计数，保留原有历史耗时采样和预算。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。 +目录监控成功通知在批次完成后复用主事件循环开启静默窗口，新增 2 条合法 Chain 到 Runtime 依赖边（含父包展开），未新增模块或 SCC。成功通知聚合复用既有 owner 和领域模型，统一交由通知模板渲染；移除汇总发送端的模板上下文直连后，原聚合实现的 14 条新增依赖边减为 12 条，未新增模块或 SCC。社区勋章名单读取复用 `app.foundation.url` 标准化 GitHub 镜像地址，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责，预览证据由 `app.application.transfer.projection` 只读投影；宿主依赖边现为 9,586；站点 API 依赖显式导入 SiteQueryPort 与 SiteStagingPort，新增 1 条合法类型边；订阅独立日志复用运行时上下文、系统日志端口与资源鉴权，新增 19 条合法依赖边（含父包展开），未新增模块或 SCC；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条 Schema 到 Domain 的合法依赖边（含父包展开）；种子下载限流窗口由 `app.domain.torrent` 定义，失败冷却 owner 复用同一语义，新增 1 条合法 Chain 到 Domain 的导入边；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条模块导入边；资源过滤动作 `app.workflow.actions.filter_torrents` 与动作包共用的领域还原函数同样把 Schema 资源还原为领域对象后再交给过滤链，新增 6 条模块导入边。音乐来源链与识别缓存复用领域身份核验，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端，经 `RequestUtils` 出站，宿主不再导入 pyTelegramBotAPI，启用 Telegram 的实例也不会因该 SDK 加载 redis；该依赖暂时保留给插件使用。Web Push 协议（aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递）由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现，消息端点与 `WebPushModule` 共用，宿主不再导入 pywebpush 及其连带的 aiohttp；该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端（经 `RequestUtils` 出站）与网络适配器 `app.adapters.network.feishu` 事件长连接（websocket-client 同步收发 pbbp2 帧），宿主不再导入 lark-oapi 及其上万个生成模块，长连接也不再依赖 SDK 的模块级事件循环；插件经 `app.sdk.feishu` 复用同一长连接，lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放，默认关闭，未开放时不生成文档。当前 `app.startup.lifecycle` 为 587、`app.factory` 为 600、`app.main` 为 602。专辑订阅入库对账新增 `app.chain.subscribe.music` 一个职责 owner；按注入的下载/整理历史 Port 读取事实，复用音频标签/CUE 读取和领域覆盖规则，依赖图增加 18 条合法边，不引入跨层数据库或 Adapter 直连。订阅业务活动新增 `app.application.subscription.activity` 及对应 DB Model、Oper、Adapter 四个模块，三入口与功能前提交对照均只增加这四个模块；性能基线仅同步结构计数，保留原有历史耗时采样和预算。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。 架构专项验证分为两个 CI 投影：`Check event semantic policy` 先运行依赖、Adapter、出口和 Event 语义门禁，`Check host architecture snapshot` 再执行快照测试及一次 diff --git a/docs/refactor/optimization-checklist.md b/docs/refactor/optimization-checklist.md index beab5aa67..c52a2d332 100644 --- a/docs/refactor/optimization-checklist.md +++ b/docs/refactor/optimization-checklist.md @@ -100,7 +100,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 1123 / 9,584 | Outbox 恢复的日志与关闭屏障新增 5 条合法依赖边（含父包展开），未新增模块或 SCC。 `dependency-baseline.json` 当前快照；豆瓣短评与影评新增 `app.schemas.comment` 一个模块和 3 条合法依赖边（含父包展开），模块方法契约同步登记同步与异步入口，未新增 SCC 或跨层直连。 `dependency-baseline.json` 当前快照；站点 API 依赖显式导入 `SiteQueryPort` 与 `SiteStagingPort`，新增 1 条合法类型边；集数定位直接导入 FileItem owner，替换 1 条间接导出依赖边，模块数、边数、SCC 与边界不变；监控分发器直接导入 FileItem owner，以 1 条 Schema 依赖边替换间接导出依赖，模块数、边数、SCC 与边界不变；工作流扫描、整理与下载动作直接复用已有 Schema owner，新增 3 条合法依赖边，模块数、SCC 与边界不变；工作流媒体过滤与下载查询复用已有 Schema 元素类型，新增 2 条合法依赖边，模块数、SCC 与边界不变；TheAudioDB 相关艺术家结果复用既有领域类型，新增 2 条合法 Chain 到 Domain 类型依赖边（含父包展开），模块数、SCC 与边界不变；群晖 Chat 渠道类型改为直接导入已有 Schema owner，移除 1 条间接导出依赖边，模块数、SCC 与边界不变；系统链、转移规划、订阅查询和规则服务改为直接导入已有 Schema owner，新增 2 条 owner 依赖边并移除 4 条间接导出依赖边，模块数、SCC 与层间边界不变；Telegram Rich Message 失败降级复用 `app.schemas.exception` 的 `ImmediateException`，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。目录监控成功通知在批次完成后复用主事件循环开启静默窗口，新增 2 条合法 Chain 到 Runtime 依赖边（含父包展开），未新增模块或 SCC。成功通知聚合复用既有 owner 和领域模型，统一交由通知模板渲染；移除汇总发送端的模板上下文直连后，原聚合实现的 14 条新增依赖边减为 12 条，未新增模块或 SCC。社区勋章名单读取复用 `app.foundation.url` 标准化 GitHub 镜像地址，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。专辑订阅入库对账新增 1 个 Chain owner 和 18 条合法依赖边（含父包展开），仍经注入 Port 读取历史并通过既有 mutation 写入；订阅业务活动新增 4 个模块和 50 条合法内部依赖边（含父包展开），Application 仅定义事实与持久化 Port，DB Adapter/Oper 实现存取，组合根注入，未增加 Application/Chain 到具体 Adapter 或数据库的直连；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条合法依赖边（含父包展开）；种子失败冷却复用 `app.domain.torrent` 的限流语义，新增 1 条合法导入边；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条合法模块导入边，资源过滤动作复用同一还原再新增 6 条；LLM 独占连接关闭复用宿主线程池；歌词字形转换复用 `app.foundation.text`；订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口，将非 TMDB 影视身份投影为 TMDB 身份；本次安全与可靠性修复新增 4 条合法依赖边：WebAgent 复用渠道类型、宿主历史清理复用文件系统代理、两个下载器复用纯领域种子哈希规则，未新增模块或 SCC；IP 归属地适配器复用 `app.schemas.common.JsonData` 声明原始 JSON 返回，新增 2 条合法类型依赖边（含父包展开），未新增模块、循环或越层依赖；TMDb 认证与片单读取复用模块既有异常，新增 2 条同模块依赖边；Session 类型使扫描器识别原有两处 close，发送与会话关闭路径未变；Audiences 构造参数复用既有 RequestUtils 类型，新增 3 条合法依赖边（含父包展开）；插件实例历史脏行清理复用既有 `app.schemas.plugin.validate_plugin_id` 进行 fail-closed 身份校验，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC | +| 宿主 Python 模块 / 内部依赖边 | 1123 / 9,586 | HTTP 共享池自恢复复用运行时日志与 TaskRegistry，新增 2 条合法 Adapter 到 Runtime 依赖边，未新增模块、SCC 或外连入口。 Outbox 恢复的日志与关闭屏障新增 5 条合法依赖边（含父包展开），未新增模块或 SCC。 `dependency-baseline.json` 当前快照；豆瓣短评与影评新增 `app.schemas.comment` 一个模块和 3 条合法依赖边（含父包展开），模块方法契约同步登记同步与异步入口，未新增 SCC 或跨层直连。 `dependency-baseline.json` 当前快照；站点 API 依赖显式导入 `SiteQueryPort` 与 `SiteStagingPort`，新增 1 条合法类型边；集数定位直接导入 FileItem owner，替换 1 条间接导出依赖边，模块数、边数、SCC 与边界不变；监控分发器直接导入 FileItem owner，以 1 条 Schema 依赖边替换间接导出依赖，模块数、边数、SCC 与边界不变；工作流扫描、整理与下载动作直接复用已有 Schema owner，新增 3 条合法依赖边，模块数、SCC 与边界不变；工作流媒体过滤与下载查询复用已有 Schema 元素类型，新增 2 条合法依赖边，模块数、SCC 与边界不变；TheAudioDB 相关艺术家结果复用既有领域类型，新增 2 条合法 Chain 到 Domain 类型依赖边（含父包展开），模块数、SCC 与边界不变；群晖 Chat 渠道类型改为直接导入已有 Schema owner，移除 1 条间接导出依赖边，模块数、SCC 与边界不变；系统链、转移规划、订阅查询和规则服务改为直接导入已有 Schema owner，新增 2 条 owner 依赖边并移除 4 条间接导出依赖边，模块数、SCC 与层间边界不变；Telegram Rich Message 失败降级复用 `app.schemas.exception` 的 `ImmediateException`，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。目录监控成功通知在批次完成后复用主事件循环开启静默窗口，新增 2 条合法 Chain 到 Runtime 依赖边（含父包展开），未新增模块或 SCC。成功通知聚合复用既有 owner 和领域模型，统一交由通知模板渲染；移除汇总发送端的模板上下文直连后，原聚合实现的 14 条新增依赖边减为 12 条，未新增模块或 SCC。社区勋章名单读取复用 `app.foundation.url` 标准化 GitHub 镜像地址，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。专辑订阅入库对账新增 1 个 Chain owner 和 18 条合法依赖边（含父包展开），仍经注入 Port 读取历史并通过既有 mutation 写入；订阅业务活动新增 4 个模块和 50 条合法内部依赖边（含父包展开），Application 仅定义事实与持久化 Port，DB Adapter/Oper 实现存取，组合根注入，未增加 Application/Chain 到具体 Adapter 或数据库的直连；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条合法依赖边（含父包展开）；种子失败冷却复用 `app.domain.torrent` 的限流语义，新增 1 条合法导入边；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条合法模块导入边，资源过滤动作复用同一还原再新增 6 条；LLM 独占连接关闭复用宿主线程池；歌词字形转换复用 `app.foundation.text`；订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口，将非 TMDB 影视身份投影为 TMDB 身份；本次安全与可靠性修复新增 4 条合法依赖边：WebAgent 复用渠道类型、宿主历史清理复用文件系统代理、两个下载器复用纯领域种子哈希规则，未新增模块或 SCC；IP 归属地适配器复用 `app.schemas.common.JsonData` 声明原始 JSON 返回，新增 2 条合法类型依赖边（含父包展开），未新增模块、循环或越层依赖；TMDb 认证与片单读取复用模块既有异常，新增 2 条同模块依赖边；Session 类型使扫描器识别原有两处 close，发送与会话关闭路径未变；Audiences 构造参数复用既有 RequestUtils 类型，新增 3 条合法依赖边（含父包展开）；插件实例历史脏行清理复用既有 `app.schemas.plugin.validate_plugin_id` 进行 fail-closed 身份校验，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | diff --git a/docs/rules/09-external-response.md b/docs/rules/09-external-response.md index f70748086..321df9ec1 100644 --- a/docs/rules/09-external-response.md +++ b/docs/rules/09-external-response.md @@ -23,6 +23,15 @@ are forbidden. - User-Agent headers - Retry logic +The shared asynchronous transport tracks consecutive `PoolTimeout` failures per +event loop and configuration. After three failures it replaces that pool for +subsequent requests and logs the recovery. Responses and other request errors +reset the count. It does not replay the failed request. The retired pool closes +after its existing client contexts exit, including streaming reads and +cancellation. While a retired pool remains unclosed, the same configuration +cannot rotate again, bounding the active generations to two. Caller-owned +clients retain their own transport lifecycle. + ```python # 此处展示适配器实现；应用服务和业务链通过注入的接口访问网络能力 from app.adapters.network.http import RequestUtils diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index cd0c0f4ed..a344b5d28 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -211,7 +211,7 @@ "bindings": [ "import:httpx2 as httpx2" ], - "fingerprint": "73879afc31531cc27241f21e0edf3b8e493a7d9aef973d0c2e28165d006e631d", + "fingerprint": "cf347be42a531743cdae292eecd31f714be16ad13584a88f49a9102688262682", "kind": "raw_transport", "source": "app.adapters.network.http", "target": "httpx2", @@ -225,8 +225,8 @@ "AsyncRequestUtils.get_stream._open_stream|call:AsyncClient", "AsyncRequestUtils.get_stream._open_stream|call:Timeout", "AsyncRequestUtils.get_stream._open_stream|call:stream", - "_get_shared_async_transport|call:AsyncHTTPTransport", - "_get_shared_async_transport|call:Limits", + "_create_shared_async_transport|call:AsyncHTTPTransport", + "_create_shared_async_transport|call:Limits", "_h2_proxy_breaker_key|call:URL" ] }, @@ -974,8 +974,8 @@ "runtime_only": true } }, - "edge_count": 9584, - "edge_sha256": "fc5c745bb4d0a28cd0c7e5536d74e92cd122acee1dda876ce8cc1a46771a2f23", + "edge_count": 9586, + "edge_sha256": "da7db077e5f3e0a34e90b0a5d695e9ee2221bdf23417473a6384fcddb0da5c67", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -1096,6 +1096,8 @@ "app.adapters.network.feishu -> app.runtime.log", "app.adapters.network.http -> app.runtime", "app.adapters.network.http -> app.runtime.correlation", + "app.adapters.network.http -> app.runtime.log", + "app.adapters.network.http -> app.runtime.tasks", "app.adapters.network.webpush -> app.adapters", "app.adapters.network.webpush -> app.adapters.network", "app.adapters.network.webpush -> app.adapters.network.http", diff --git a/tests/fixtures/architecture/dependency-policy.json b/tests/fixtures/architecture/dependency-policy.json index e146062c3..5d5d2e344 100644 --- a/tests/fixtures/architecture/dependency-policy.json +++ b/tests/fixtures/architecture/dependency-policy.json @@ -105,7 +105,7 @@ {"source": "app.adapters.network.doh", "target": "socket.getaddrinfo", "kind": "protocol_operation", "fingerprint": "4ff03419dfacc6bf582b7d4421dd5a0666a63f8ca79be2b1e625f4c8f4c96b71"}, {"source": "app.adapters.network.doh", "target": "urllib.request", "kind": "raw_transport", "fingerprint": "6f5f5fd3da02a9e780ea5e7cc1e47bd962314a1a358f14b4ee698485f96ab52b"}, {"source": "app.adapters.network.http", "target": "httpx", "kind": "raw_transport", "fingerprint": "c00458c868240f1adfe6278ba22bd5c4782e9bd2fcf896114c2288bf82981ee3"}, - {"source": "app.adapters.network.http", "target": "httpx2", "kind": "raw_transport", "fingerprint": "73879afc31531cc27241f21e0edf3b8e493a7d9aef973d0c2e28165d006e631d"}, + {"source": "app.adapters.network.http", "target": "httpx2", "kind": "raw_transport", "fingerprint": "cf347be42a531743cdae292eecd31f714be16ad13584a88f49a9102688262682"}, {"source": "app.adapters.network.http", "target": "requests", "kind": "raw_transport", "fingerprint": "c95df710a8536328f5ee005c0ce065b9528f2f4ebae1f077315102d8c0f4c01f"}, {"source": "app.adapters.network.ip", "target": "socket.gethostbyname", "kind": "protocol_operation", "fingerprint": "a43e8969e2f26546fcf925b258728b309f1b9b966ff3c7b48a64273b8c82d048"}, {"source": "app.adapters.network.resolver", "target": "socket.getaddrinfo", "kind": "protocol_operation", "fingerprint": "1b1ce6763d730d7aa9c35d32a668560013c939467dcb3087357ffa53653ff306"} diff --git a/tests/test_architecture_egress.py b/tests/test_architecture_egress.py index cd4789412..8be006f84 100644 --- a/tests/test_architecture_egress.py +++ b/tests/test_architecture_egress.py @@ -98,7 +98,8 @@ FROZEN_EGRESS_FINGERPRINT_BY_EDGE = { ("app.adapters.network.doh", "socket.getaddrinfo"): "4ff03419dfacc6bf582b7d4421dd5a0666a63f8ca79be2b1e625f4c8f4c96b71", ("app.adapters.network.doh", "urllib.request"): "6f5f5fd3da02a9e780ea5e7cc1e47bd962314a1a358f14b4ee698485f96ab52b", ("app.adapters.network.http", "httpx"): "c00458c868240f1adfe6278ba22bd5c4782e9bd2fcf896114c2288bf82981ee3", - ("app.adapters.network.http", "httpx2"): "73879afc31531cc27241f21e0edf3b8e493a7d9aef973d0c2e28165d006e631d", + # 池构造移入同一 owner 的专用函数以支持恢复；调用种类、数量与外连边均未增加。 + ("app.adapters.network.http", "httpx2"): "cf347be42a531743cdae292eecd31f714be16ad13584a88f49a9102688262682", ("app.adapters.network.http", "requests"): "c95df710a8536328f5ee005c0ce065b9528f2f4ebae1f077315102d8c0f4c01f", ("app.adapters.network.http", "urllib3"): "66cbd8ec4e7552bd458db0baada30f1953e6d0493793822d23a2199567bca98a", ("app.adapters.network.ip", "socket.gethostbyname"): "a43e8969e2f26546fcf925b258728b309f1b9b966ff3c7b48a64273b8c82d048", diff --git a/tests/test_http_pool_recovery.py b/tests/test_http_pool_recovery.py new file mode 100644 index 000000000..43e1c587e --- /dev/null +++ b/tests/test_http_pool_recovery.py @@ -0,0 +1,416 @@ +""" +共享异步 HTTP 池的连续超时恢复、在途响应和关闭边界回归。 +""" + +import asyncio +from collections.abc import AsyncIterator +from typing import Any, Optional + +import httpx2 +import pytest +import pytest_asyncio +from httpcore2 import AsyncMockBackend + +from app.adapters.network import http as http_module +from app.adapters.network.http import AsyncRequestUtils + +URL = "https://raw.githubusercontent.com/demo/repo/main/package.v3.json" + + +class _ResponseStream(httpx2.AsyncByteStream): + """ + 记录响应体的生命周期，以便发现旧池关闭早于响应退出。 + """ + + def __init__(self, transport: "_ControlledTransport") -> None: + """ + 把当前响应登记为仍在读取的流。 + """ + self.transport = transport + self.closed = False + transport.streams += 1 + + async def __aiter__(self) -> AsyncIterator[bytes]: + """ + 返回离线索引载荷，保持真实 HTTPX 流式读取路径。 + """ + assert not self.transport.closed + yield b'{"Demo":{"version":"1.0"}}' + + async def aclose(self) -> None: + """ + 记录响应流关闭，允许断言它先于退役池关闭。 + """ + if not self.closed: + self.closed = True + self.transport.streams -= 1 + + +class _ControlledTransport(httpx2.AsyncBaseTransport): + """ + 在传输边界注入取池超时，不访问真实代理或网络。 + """ + + def __init__(self, **kwargs: Any) -> None: + """ + 保存新池的完整配置和请求记录，默认模拟无法分配连接。 + """ + self.kwargs = kwargs + self.error: Optional[type[httpx2.RequestError]] = httpx2.PoolTimeout + self.requests: list[httpx2.Request] = [] + self.streams = 0 + self.closed = False + self.close_calls = 0 + self.close_error = False + self.release_requests: Optional[asyncio.Event] = None + + async def handle_async_request(self, request: httpx2.Request) -> httpx2.Response: + """ + 失败时保留真实异常类别，成功时交给真实客户端消费响应体。 + """ + assert not self.closed + self.requests.append(request) + if self.release_requests is not None: + await self.release_requests.wait() + if self.error is not None: + raise self.error("injected pool state", request=request) + return httpx2.Response(200, stream=_ResponseStream(self), request=request) + + async def aclose(self) -> None: + """ + 验证活跃响应未被中途关闭，并允许注入一次关闭失败。 + """ + self.close_calls += 1 + if self.close_error: + self.close_error = False + raise RuntimeError("injected close failure") + assert self.streams == 0 + self.closed = True + + +@pytest_asyncio.fixture +async def pools(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[list[_ControlledTransport]]: + """ + 每个事件循环创建独立受控池，并在用例结束后回收全部托管任务。 + """ + created: list[_ControlledTransport] = [] + + def create(**kwargs: Any) -> _ControlledTransport: + """ + 首个池模拟故障，替换池默认可正常返回响应。 + """ + transport = _ControlledTransport(**kwargs) + if created: + transport.error = None + created.append(transport) + return transport + + monkeypatch.setattr(http_module.httpx2, "AsyncHTTPTransport", create) + try: + yield created + finally: + await http_module.aclose_shared_async_transports() + await asyncio.sleep(0) + loop = asyncio.get_running_loop() + assert not http_module._retired_async_transports.get(loop) + assert not any(task.get_loop() is loop for task in http_module._pending_eviction_tasks) + + +async def _timeout(request: AsyncRequestUtils, *, stream: bool = False, method: str = "GET") -> None: + """ + 通过公开请求入口触发一次取池超时，并核对异常传播。 + """ + with pytest.raises(httpx2.PoolTimeout): + if stream: + async with request.get_stream(URL, raise_exception=True): + pytest.fail("故障池不应返回响应") + else: + await request.request(method, URL, raise_exception=True) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("stream", [False, True]) +async def test_repeated_pool_timeouts_replace_only_failed_pool(pools, stream: bool) -> None: + """ + 普通和流式请求均在连续三次取池超时后恢复，保留代理及池限制配置。 + """ + request = AsyncRequestUtils(proxies="http://proxy.example:7890", timeout=15) + for _ in range(2): + await _timeout(request, stream=stream) + assert len(pools) == 1 + await _timeout(request, stream=stream) + await asyncio.sleep(0) + assert len(pools) == 2 + assert pools[0].closed + assert len(pools[0].requests) == 3 + assert pools[0].kwargs.keys() == pools[1].kwargs.keys() + for key in ("proxy", "verify", "http2", "trust_env"): + assert pools[0].kwargs[key] == pools[1].kwargs[key] + assert vars(pools[0].kwargs["limits"]) == vars(pools[1].kwargs["limits"]) + async with request.get_stream(URL, raise_exception=True) as response: + assert await response.aread() == b'{"Demo":{"version":"1.0"}}' + assert len(pools) == 2 + + +@pytest.mark.asyncio +async def test_retired_pool_drains_stream_and_limits_generations(pools) -> None: + """ + 退役池保留在途流；旧代未退出时，新代再超时也不会增加第三代。 + """ + request = AsyncRequestUtils() + await _timeout(request) + pools[0].error = None + async with request.get_stream(URL, raise_exception=True) as original: + pools[0].error = httpx2.PoolTimeout + for _ in range(3): + await _timeout(request) + await asyncio.sleep(0) + assert len(pools) == 2 + assert not pools[0].closed + pools[1].error = httpx2.PoolTimeout + for _ in range(4): + await _timeout(request) + assert len(pools) == 2 + assert await original.aread() == b'{"Demo":{"version":"1.0"}}' + await asyncio.sleep(0) + assert pools[0].closed + await _timeout(request) + assert len(pools) == 3 + assert await request.get(URL) is not None + + +@pytest.mark.asyncio +async def test_success_resets_consecutive_pool_failures(pools) -> None: + """ + 有请求成功获得连接就清零失败计数，避免零散拥塞触发重建。 + """ + request = AsyncRequestUtils() + await _timeout(request) + await _timeout(request) + pools[0].error = None + assert await request.get(URL) is not None + pools[0].error = httpx2.PoolTimeout + await _timeout(request) + await _timeout(request) + assert len(pools) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("error", [httpx2.ConnectTimeout, httpx2.ReadTimeout]) +async def test_other_timeouts_do_not_replace_pool(pools, error) -> None: + """ + 网络连接和读取超时维持既有语义，不作为池状态故障触发重建。 + """ + request = AsyncRequestUtils() + await _timeout(request) + pools[0].error = error + for _ in range(4): + assert await request.get(URL) is None + pools[0].error = httpx2.PoolTimeout + await _timeout(request) + await _timeout(request) + assert len(pools) == 1 + + +@pytest.mark.asyncio +async def test_post_is_not_replayed_and_false_exception_contract_is_preserved(pools) -> None: + """ + 恢复只影响后续请求，POST 不重放，默认失败仍返回 None。 + """ + request = AsyncRequestUtils() + for _ in range(3): + assert await request.request("POST", URL) is None + assert len(pools[0].requests) == 3 + assert all(request.method == "POST" for request in pools[0].requests) + assert await request.request("POST", URL) is not None + assert len(pools[1].requests) == 1 + + +@pytest.mark.asyncio +async def test_caller_owned_client_is_not_replaced(pools) -> None: + """ + 调用方自管客户端的池和关闭权保持不变。 + """ + transport = _ControlledTransport() + async with httpx2.AsyncClient(transport=transport) as client: + request = AsyncRequestUtils(client=client) + for _ in range(4): + await _timeout(request) + assert not transport.closed + assert not pools + assert transport.closed + + +@pytest.mark.asyncio +async def test_close_failure_retains_old_generation_until_shutdown_retry(pools) -> None: + """ + 退役关闭失败时保留 owner 阻止继续轮换，并由关闭入口重试回收。 + """ + request = AsyncRequestUtils() + await _timeout(request) + pools[0].close_error = True + await _timeout(request) + await _timeout(request) + await asyncio.sleep(0) + assert not pools[0].closed + pools[1].error = httpx2.PoolTimeout + for _ in range(3): + await _timeout(request) + assert len(pools) == 2 + await http_module.aclose_shared_async_transports() + assert pools[0].closed + assert pools[0].close_calls == 2 + + +@pytest.mark.asyncio +async def test_registry_shutdown_keeps_retired_pool_for_explicit_close(pools, monkeypatch) -> None: + """ + 任务登记器停止接收后，归还借用仍保留原异常，并把旧池交给显式关闭入口。 + """ + from app.runtime.tasks import TaskRegistry + + registry = TaskRegistry() + assert await registry.shutdown() + monkeypatch.setattr(http_module, "get_task_registry", lambda: registry) + request = AsyncRequestUtils() + for _ in range(3): + await _timeout(request) + assert not pools[0].closed + assert await request.get(URL) is not None + await http_module.aclose_shared_async_transports() + assert pools[0].closed + + +@pytest.mark.asyncio +async def test_plugin_index_recovers_through_real_streaming_entry(pools, monkeypatch) -> None: + """ + 插件索引调用真实异步读取路径，在共享池更换后下一次查询成功。 + """ + from app.adapters.external.plugin.client import PluginMarketTransport + + monkeypatch.setattr( + PluginMarketTransport, + "_build_github_request_strategies", + lambda **_kwargs: [("代理", URL, {"proxies": "http://proxy.example:7890"})], + ) + transport = PluginMarketTransport() + for _ in range(3): + with pytest.raises(RuntimeError, match="连接失败"): + await transport._fetch_plugin_index_async(URL, None) + assert await transport._fetch_plugin_index_async(URL, None) == {"Demo": {"version": "1.0"}} + assert len(pools) == 2 + + +@pytest.mark.asyncio +async def test_concurrent_old_generation_failures_do_not_rotate_replacement(pools) -> None: + """ + 同一波旧代请求延迟返回超时时只轮换一次，不误退役后继池。 + """ + request = AsyncRequestUtils() + await _timeout(request) + pools[0].release_requests = asyncio.Event() + tasks = [asyncio.create_task(request.request("POST", URL)) for _ in range(8)] + await asyncio.sleep(0) + assert len(pools[0].requests) == 9 + pools[0].release_requests.set() + assert await asyncio.gather(*tasks) == [None] * 8 + await asyncio.sleep(0) + assert len(pools) == 2 + assert pools[0].closed + assert len(pools[1].requests) == 0 + assert await request.get(URL) is not None + + +@pytest.mark.asyncio +async def test_cancelled_stream_releases_retired_pool(pools) -> None: + """ + 持有旧流的任务被取消后也会归还借用，退役池最终关闭。 + """ + request = AsyncRequestUtils() + await _timeout(request) + pools[0].error = None + entered = asyncio.Event() + + async def hold_stream() -> None: + """ + 持有真实客户端流上下文直到测试取消任务。 + """ + async with request.get_stream(URL, raise_exception=True): + entered.set() + await asyncio.Event().wait() + + task = asyncio.create_task(hold_stream()) + try: + await entered.wait() + pools[0].error = httpx2.PoolTimeout + for _ in range(3): + await _timeout(request) + assert not pools[0].closed + finally: + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + await asyncio.sleep(0) + assert pools[0].closed + + +@pytest.mark.asyncio +async def test_recovery_isolated_by_transport_configuration(pools) -> None: + """ + 一个配置桶退役时，其他配置仍复用原池。 + """ + failing = AsyncRequestUtils(http2=True) + healthy = AsyncRequestUtils(http2=False) + await _timeout(failing) + assert await healthy.get(URL) is not None + await _timeout(failing) + await _timeout(failing) + assert len(pools) == 3 + assert await healthy.get(URL) is not None + assert len(pools[1].requests) == 2 + assert not pools[1].closed + + +@pytest.mark.asyncio +async def test_real_connect_pool_recovers_without_interrupting_active_response(monkeypatch) -> None: + """ + 用真实 HTTPX/httpcore CONNECT 单连接池制造排队超时，验证新池恢复及旧流排空。 + + 只在网络后端回放字节，不替换池分配、超时异常或客户端关闭逻辑。 + """ + original_transport = httpx2.AsyncHTTPTransport + created: list[httpx2.AsyncHTTPTransport] = [] + + def create(**kwargs: Any) -> httpx2.AsyncHTTPTransport: + """ + 保留真实代理池，仅将 TCP/TLS I/O 替换为离线 CONNECT 响应。 + """ + transport = original_transport(**kwargs) + transport._pool._network_backend = AsyncMockBackend([ + b"HTTP/1.1 200 Connection established\r\n\r\n", + b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello", + ]) + created.append(transport) + return transport + + monkeypatch.setattr(http_module.httpx2, "AsyncHTTPTransport", create) + request = AsyncRequestUtils( + proxies="http://proxy.example:7890", http2=False, max_connections=1, + ) + try: + async with request.get_stream(URL, raise_exception=True) as original: + for _ in range(3): + with pytest.raises(httpx2.PoolTimeout): + await request.request( + "GET", URL, raise_exception=True, + timeout=httpx2.Timeout(1, pool=0.01), + ) + assert len(created) == 2 + assert len(created[0]._pool.connections) == 1 + assert await request.get(URL) == "hello" + assert await original.aread() == b"hello" + await asyncio.sleep(0) + assert created[0]._pool.connections == [] + finally: + await http_module.aclose_shared_async_transports() + assert created[1]._pool.connections == []