mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-31 20:34:48 +08:00
fix(network): recover shared pools after repeated pool timeouts
This commit is contained in:
jxxghp
@@ -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:
|
||||
|
||||
@@ -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` 再执行快照测试及一次
|
||||
|
||||
@@ -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 等基线均为零 |
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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"}
|
||||
|
||||
@@ -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",
|
||||
|
||||
416
tests/test_http_pool_recovery.py
Normal file
416
tests/test_http_pool_recovery.py
Normal file
@@ -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 == []
|
||||
Reference in New Issue
Block a user