diff --git a/app/adapters/network/doh.py b/app/adapters/network/doh.py index a18eb2291..3ad3ebada 100644 --- a/app/adapters/network/doh.py +++ b/app/adapters/network/doh.py @@ -3,25 +3,26 @@ doh函数的实现。 author: https://github.com/C5H12O5/syno-videoinfo-plugin """ import base64 -import concurrent -import concurrent.futures import json import socket import struct import urllib import urllib.request +from concurrent.futures import as_completed from threading import Lock from typing import Dict, Optional +from app.foundation.singleton import Singleton +from app.runtime.execution import OwnedThreadPoolExecutor from app.runtime.log import logger from app.runtime.reload import ConfigReloadMixin from app.runtime.settings import get_runtime_setting -from app.foundation.singleton import Singleton # DoH 关闭时需要释放线程池；保持惰性创建可避免未启用 DoH 时占用进程级资源 -_executor: Optional[concurrent.futures.ThreadPoolExecutor] = None +_executor: Optional[OwnedThreadPoolExecutor] = None _executor_lock = Lock() _doh_enabled = False +_DOH_EXECUTOR_STOP_TIMEOUT_SECONDS = 10.0 # 定义默认的DoH配置 _doh_timeout = 5 @@ -36,20 +37,23 @@ def _doh_setting(key: str): return get_runtime_setting(key) -def _get_executor_locked() -> concurrent.futures.ThreadPoolExecutor: - """在持有执行器锁时按需获取 DoH 查询线程池""" +def _get_executor_locked() -> OwnedThreadPoolExecutor: + """在持有执行器锁时按需获取 DoH 查询线程池。""" global _executor if _executor is None: - _executor = concurrent.futures.ThreadPoolExecutor() + _executor = OwnedThreadPoolExecutor() return _executor -def enable_doh(enable: bool) -> None: +def enable_doh(enable: bool) -> bool: """ - 对 socket.getaddrinfo 进行补丁 + 对 socket.getaddrinfo 进行补丁。 + + :param enable: 是否启用 DoH 解析 + :return: 状态切换成功时返回 True；旧 executor 未收敛时返回 False """ - global _doh_enabled + global _doh_enabled, _executor def _patched_getaddrinfo(host: str, *args, **kwargs): """ @@ -73,19 +77,31 @@ def enable_doh(enable: bool) -> None: executor.submit(_doh_query, resolver, host) for resolver in _doh_setting("DOH_RESOLVERS").split(",") ] - for future in concurrent.futures.as_completed(futures): + for future in as_completed(futures): ip = future.result() if ip is not None: logger.info(f"已解析 [{host}] 为 [{ip}]") - with _doh_lock: - _doh_cache[host] = ip + # 关闭可能在查询等待期间恢复系统 DNS；关闭后的结果不得回填到下一轮配置。 + with _executor_lock: + cache_allowed = _doh_enabled + if cache_allowed: + with _doh_lock: + _doh_cache[host] = ip host = ip break return _orig_getaddrinfo(host, *args, **kwargs) with _executor_lock: + if enable and _executor is not None and not _executor.accepting: + # 上一轮 shutdown 超时后必须继续持有原 owner；只有真实收敛才能替换执行器。 + if not _executor.shutdown_bounded(timeout=0.0): + _doh_enabled = False + socket.getaddrinfo = _orig_getaddrinfo + return False + _executor = None _doh_enabled = enable socket.getaddrinfo = _patched_getaddrinfo if enable else _orig_getaddrinfo + return True class DohHelper(ConfigReloadMixin, metaclass=Singleton): @@ -101,29 +117,40 @@ class DohHelper(ConfigReloadMixin, metaclass=Singleton): def on_config_changed(self) -> None: """配置变化时清理缓存并重新应用 DoH 状态。""" if not _doh_setting("DOH_ENABLE"): - self.shutdown() + if not self.shutdown(): + logger.error("DoH配置关闭后查询线程池未在预算内收敛") return with _doh_lock: # DOH配置有变动的情况下，清空缓存 _doh_cache.clear() - enable_doh(True) + if not enable_doh(True): + logger.error("DoH查询线程池尚未收敛，暂不重新启用") def get_reload_name(self) -> str: """返回 DoH 配置重载名称。""" return 'DoH' - def shutdown(self) -> None: - """恢复系统 DNS 并释放 DoH 查询线程池""" + def shutdown( + self, + timeout: float = _DOH_EXECUTOR_STOP_TIMEOUT_SECONDS, + ) -> bool: + """ + 恢复系统 DNS 并有限等待 DoH 查询线程池。 + + :param timeout: 等待已接受查询和 worker 终止的最长秒数 + :return: 查询线程池真实终止时返回 True，否则返回 False + """ global _executor, _doh_enabled with _executor_lock: _doh_enabled = False socket.getaddrinfo = _orig_getaddrinfo executor = _executor - _executor = None + converged = executor is None or executor.shutdown_bounded(timeout=timeout) + if converged and _executor is executor: + _executor = None with _doh_lock: _doh_cache.clear() - if executor: - executor.shutdown(wait=True) + return converged def _doh_query(resolver: str, host: str) -> Optional[str]: diff --git a/app/runtime/execution.py b/app/runtime/execution.py index e9aad5506..bc82deeaa 100644 --- a/app/runtime/execution.py +++ b/app/runtime/execution.py @@ -1,15 +1,105 @@ import asyncio import inspect +import threading import time +from concurrent.futures import Future, ThreadPoolExecutor, wait from contextvars import copy_context from functools import partial, wraps -from typing import Any, Callable, TypeVar +from typing import Any, Callable, TypeVar, cast from app.schemas.exception import ImmediateException from anyio.to_thread import run_sync TaskResult = TypeVar("TaskResult") +ExecutorResult = TypeVar("ExecutorResult") + + +class OwnedThreadPoolExecutor(ThreadPoolExecutor): + """ + 追踪已接受 Future，并提供可重试的有界关闭合同。 + + 该 owner 不取消排队任务，保持 ``ThreadPoolExecutor.shutdown(wait=True)`` + 的历史完成语义；区别仅在于调用方可以在预算耗尽后保留同一实例继续收敛。 + """ + + def __init__(self, max_workers: int | None = None) -> None: + """初始化线程池、提交准入状态和 Future owner 集合。""" + super().__init__(max_workers=max_workers) + self._ownership_lock = threading.RLock() + self._accepting = True + self._owned_futures: set[Future[Any]] = set() + + @property + def accepting(self) -> bool: + """返回执行器是否仍允许提交新任务。""" + with self._ownership_lock: + return self._accepting + + def submit( + self, + fn: Callable[..., ExecutorResult], + /, + *args: Any, + **kwargs: Any, + ) -> Future[ExecutorResult]: + """提交任务并在其达到终态前保留 owner。""" + with self._ownership_lock: + future = super().submit(fn, *args, **kwargs) + self._owned_futures.add(future) + future.add_done_callback(self._discard_future) + return future + + def shutdown( + self, + wait: bool = True, + *, + cancel_futures: bool = False, + ) -> None: + """封口提交准入并保留标准库 shutdown 的调用语义。""" + with self._ownership_lock: + self._accepting = False + # 先在锁内封口；真正等待必须在锁外进行，否则 worker 的完成回调无法释放 owner。 + super().shutdown(wait=False, cancel_futures=cancel_futures) + if wait: + super().shutdown(wait=True, cancel_futures=cancel_futures) + + def _discard_future(self, future: Future[Any]) -> None: + """任务达到终态后释放 owner 记录。""" + with self._ownership_lock: + self._owned_futures.discard(future) + + def shutdown_bounded(self, timeout: float) -> bool: + """ + 封口新提交并有限等待全部已接受任务。 + + :param timeout: 等待 Future 达到终态的最长秒数 + :return: 所有任务与 worker 均已终止时返回 True，否则返回 False + """ + deadline = time.monotonic() + max(0.0, timeout) + with self._ownership_lock: + self._accepting = False + # 不取消排队工作，保持历史 shutdown(wait=True) 的完成语义。 + super().shutdown(wait=False) + owned_futures = tuple(self._owned_futures) + if owned_futures: + _, pending_futures = wait( + owned_futures, + timeout=max(0.0, deadline - time.monotonic()), + ) + if pending_futures: + return False + # 标准库只提供无界 wait=True；Future 又会先标记完成再执行 done callback， + # 因此封口后读取稳定 worker 集合，复用同一 deadline 做有限 join。 + worker_threads = tuple(cast(set[threading.Thread], self._threads)) + current_thread = threading.current_thread() + for worker_thread in worker_threads: + if worker_thread is current_thread: + continue + worker_thread.join( + timeout=max(0.0, deadline - time.monotonic()), + ) + return all(not worker_thread.is_alive() for worker_thread in worker_threads) async def await_task_to_terminal( diff --git a/app/runtime/thread.py b/app/runtime/thread.py index 93cfa3d1f..7c9af59b6 100644 --- a/app/runtime/thread.py +++ b/app/runtime/thread.py @@ -1,10 +1,9 @@ -import threading -import time -from concurrent.futures import Future, ThreadPoolExecutor, wait +from concurrent.futures import Future from contextvars import copy_context from typing import Any, Callable, TypeVar, cast from app.foundation.singleton import Singleton +from app.runtime.execution import OwnedThreadPoolExecutor from app.runtime.settings import RuntimeSettingsCompat settings = RuntimeSettingsCompat() @@ -12,67 +11,8 @@ settings = RuntimeSettingsCompat() _Result = TypeVar("_Result") _THREAD_POOL_STOP_TIMEOUT_SECONDS = 10.0 - -class _OwnedThreadPoolExecutor(ThreadPoolExecutor): - """ - 追踪所有提交入口的共享执行器，包括旧调用方直接使用的 ``pool.submit``。 - """ - - def __init__(self, max_workers: int) -> None: - """初始化线程池及其 Future owner 集合。""" - super().__init__(max_workers=max_workers) - self._ownership_lock = threading.RLock() - self._owned_futures: set[Future[Any]] = set() - - def submit( - self, - fn: Callable[..., _Result], - /, - *args: Any, - **kwargs: Any, - ) -> Future[_Result]: - """提交任务并在其达到终态前保留 owner。""" - with self._ownership_lock: - future = super().submit(fn, *args, **kwargs) - self._owned_futures.add(future) - future.add_done_callback(self._discard_future) - return future - - def _discard_future(self, future: Future[Any]) -> None: - """任务达到终态后释放 owner 记录。""" - with self._ownership_lock: - self._owned_futures.discard(future) - - def shutdown_bounded(self, timeout: float) -> bool: - """ - 封口新提交并有限等待全部已接受任务。 - - :param timeout: 等待 Future 达到终态的最长秒数 - :return: 所有任务与 worker 均已终止时返回 True，否则返回 False - """ - deadline = time.monotonic() + max(0.0, timeout) - with self._ownership_lock: - # 不取消排队工作，保持历史 shutdown(wait=True) 的完成语义。 - super().shutdown(wait=False) - owned_futures = tuple(self._owned_futures) - if owned_futures: - _, pending_futures = wait( - owned_futures, - timeout=max(0.0, deadline - time.monotonic()), - ) - if pending_futures: - return False - # 标准库只提供无界 wait=True；Future 又会先标记完成再执行 done callback， - # 因此封口后读取稳定 worker 集合，复用同一 deadline 做有限 join。 - worker_threads = tuple(cast(set[threading.Thread], self._threads)) - current_thread = threading.current_thread() - for worker_thread in worker_threads: - if worker_thread is current_thread: - continue - worker_thread.join( - timeout=max(0.0, deadline - time.monotonic()), - ) - return all(not worker_thread.is_alive() for worker_thread in worker_threads) +# 阶段 64 曾在本模块引入该私有名；保留精确别名，避免测试或外部诊断代码失效。 +_OwnedThreadPoolExecutor = OwnedThreadPoolExecutor # strict mypy 跳过 foundation 实现导入，因此无法在本文件解析既有 Singleton 元类类型。 @@ -83,7 +23,7 @@ class ThreadHelper(metaclass=Singleton): # type: ignore[metaclass] def __init__(self) -> None: """按系统配置创建共享后台线程池。""" - self.pool = _OwnedThreadPoolExecutor(max_workers=settings.CONF.threadpool) + self.pool = OwnedThreadPoolExecutor(max_workers=settings.CONF.threadpool) def submit( self, @@ -99,7 +39,8 @@ class ThreadHelper(metaclass=Singleton): # type: ignore[metaclass] :return: future """ context = copy_context() - return self.pool.submit(context.run, func, *args, **kwargs) + # strict mypy 跳过 execution 实现导入，需要在兼容门面恢复 Future 泛型。 + return cast(Future[_Result], self.pool.submit(context.run, func, *args, **kwargs)) def shutdown( self, @@ -111,4 +52,4 @@ class ThreadHelper(metaclass=Singleton): # type: ignore[metaclass] :param timeout: 等待已接受任务达到终态的最长秒数 :return: 全部任务和 worker 均已终止时返回 True，否则返回 False """ - return self.pool.shutdown_bounded(timeout=timeout) + return bool(self.pool.shutdown_bounded(timeout=timeout)) diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 9d821029b..5474366b6 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -2,13 +2,13 @@ > 文档性质：当前架构复核、优秀 Python 后端实践对标、AI 可执行任务手册 > 适用仓库：`MoviePilot`，分支 `v3` -> 审计基线：`88262191`（2026-08-24） +> 审计基线：`e9053a65`（2026-08-24） > 审计范围：宿主后端；排除 `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 已统一插件索引请求与响应三态策略；阶段 52 已统一插件 Release 分页策略；阶段 53 已统一远端插件安装模式决策；阶段 54 已补齐同步安装成功后的临时回滚备份清理；阶段 55～56 已收口官方插件观察基线与报告保留策略；阶段 57 已统一进程级运行时 Facade 门禁并补齐 ModuleManager 边界；阶段 58 已消除 AgentTask 关闭回归的跨线程零时长等待竞态；阶段 59 已统一 Feishu 多实例长连接的 SDK 循环路由；阶段 60 已清除命令服务虚假的关停 owner 声明；阶段 61 已统一 Capability Runtime 同步/异步关闭的诚实收敛结果；阶段 62 已统一消息渠道长连接的多实例关闭收敛合同；阶段 63 已补齐应用消息队列线程的关闭收敛合同；阶段 64 已统一共享线程池的有界关闭 owner；阶段 65 已统一异步文件日志的单一有界写入和关闭 owner。 +> 实施进度：阶段 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 分页策略；阶段 53 已统一远端插件安装模式决策；阶段 54 已补齐同步安装成功后的临时回滚备份清理；阶段 55～56 已收口官方插件观察基线与报告保留策略；阶段 57 已统一进程级运行时 Facade 门禁并补齐 ModuleManager 边界；阶段 58 已消除 AgentTask 关闭回归的跨线程零时长等待竞态；阶段 59 已统一 Feishu 多实例长连接的 SDK 循环路由；阶段 60 已清除命令服务虚假的关停 owner 声明；阶段 61 已统一 Capability Runtime 同步/异步关闭的诚实收敛结果；阶段 62 已统一消息渠道长连接的多实例关闭收敛合同；阶段 63 已补齐应用消息队列线程的关闭收敛合同；阶段 64 已统一共享线程池的有界关闭 owner；阶段 65 已统一异步文件日志的单一有界写入和关闭 owner；阶段 66 已统一 DoH 与共享线程池的有界 executor owner。 > 当前 canonical 状态：API/Application 公共复杂度基线已清零，组合根外 `SystemConfigOper()` 构造和 Model/Oper 隐式事务均为 0；命名 Chain/Agent 数据端口、TaskRegistry owner、Module Contract V2、typed Event、Outbox durable intent、请求关联和插件运行时 getter 已形成当前路径。插件仓适配、未知第三方 fallback 和其它 E1/E3 副作用仍按风险持续治理。 -> 最新阶段：阶段 65 已统一异步文件日志的单一有界写入和关闭 owner。 +> 最新阶段：阶段 66 已统一 DoH 与共享线程池的有界 executor owner。 ## 当前复核结论（2026-08-24） @@ -17,7 +17,7 @@ ### 长期整改阶段 0：治理门禁恢复（2026-08-23） -- 宿主依赖基线已审查 TaskRegistry、有界后台 owner 与插件变更准入接入后的语义差异：当前为 `810` 个模块、`6560` 条内部导入边，12 组重点禁止边继续全部为 `0`，唯一非平凡 SCC 仍是隔离的 TMDB 移植包。 +- 宿主依赖基线已审查 TaskRegistry、有界后台 owner 与插件变更准入接入后的语义差异：当前为 `810` 个模块、`6562` 条内部导入边，12 组重点禁止边继续全部为 `0`，唯一非平凡 SCC 仍是隔离的 TMDB 移植包。 - 启动性能探针会在隔离生命周期中真实创建并释放 TaskRegistry；normal/safe 组件数分别为 `23`/`11`，CI 只读检查使用稳定的宿主模块集合和生命周期组件顺序，不再把 Python/平台模块数量当作硬合同。 - 官方插件快照覆盖 `plugins.v3`、`plugins.v2` 以及 V3 实际会从 `package.json` 回退加载的 31 个默认实现；`app/plugins/**` 仍只是宿主运行副本，不进入扫描。 - SDK 快照以各模块显式 `__all__` 为公开合同，能够记录赋值别名；`typing`、`__future__` 等实现期导入不再被误冻结，既有数据库备份门面已补精确导出清单。 @@ -707,6 +707,26 @@ 日志/生命周期/旧导入专项 83 项、架构合同 94 项、兼容调用链 102 项、Pylint 10.00/10、strict mypy 41 文件、全部宿主与质量 ratchet 及四分片全量 `5952 passed, 3 skipped` 均通过。 +### 长期整改阶段 66：DoH 与共享线程池有界 executor owner 统一（2026-08-24） + +- DoH 仍单独创建标准库 `ThreadPoolExecutor`，关闭时先清空全局句柄，再执行无界 `shutdown(wait=True)`； + 任一底层 DNS/HTTPS 调用未返回都会阻塞 `stop_modules()` 所在事件循环，外层生命周期预算无法介入， + 且已丢失的 executor 无法重试收敛。阻塞探针确认旧关闭 50ms 内不返回，同时 owner 已从全局移除。 +- 阶段 64 的 Future 追踪和有界 worker join 已抽取为 + `app.runtime.execution.OwnedThreadPoolExecutor` 唯一实现；`ThreadHelper` 与 DoH 共同复用，旧私有类名 + 保留精确别名。标准 `shutdown(wait=True)` 仍可用于旧 `.pool` 调用，但采用锁内封口、锁外等待，避免 + worker 完成回调与 owner 锁互锁。 +- DoH 关闭现在先恢复系统 DNS，再使用 10 秒默认预算有限等待；超时返回 `False` 并保留已封口 executor， + startup 继续收口其它资源并向上报告失败。释放阻塞查询后可在同一 owner 上重试成功，只有真实终止后 + 才清空句柄；配置重启也不得覆盖未收敛 owner，关闭期间完成的旧查询不会回填下一轮缓存。 +- `DohHelper()`、`enable_doh()`、无参数 `shutdown()`、socket 补丁、惰性建池、解析器并发、缓存命中及 + `app.helper.doh` 精确映射全部保留；新增 timeout 和布尔收敛结果为向后兼容扩展。未修改 SDK/Compat + 清单、插件 Hook 或插件仓，V1/V2/V3 插件导入和调用边界不变。 +- 阻塞查询、有限返回、owner 保留、拒绝替换、重试终态、标准 shutdown 兼容和 startup 失败传播均有 + 故障注入；DoH/线程池/生命周期专项 66 项、旧导入与插件兼容联合专项 174 项、架构合同 94 项、 + Pylint 10.00/10、strict mypy 41 文件、全部宿主与质量 ratchet 及四分片全量 + `5955 passed, 3 skipped` 均通过。 + ### 总体判断 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**： diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index ffa138da8..345cf2677 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -201,6 +201,9 @@ ModuleManager 与 startup 组合根继续关闭其余资源但必须向上返回 并向 startup 返回 `False`，不得用无界 `join()` 阻塞生命周期或把日志当作成功。 共享 `ThreadHelper` 必须追踪通过宿主 `submit()` 和旧兼容 `.pool.submit()` 接受的全部 Future；关闭时 先封口新任务，再有限等待且保留未终止 owner，结果由 startup 聚合，不得恢复无界 executor shutdown。 +`app.runtime.execution.OwnedThreadPoolExecutor` 是进程级同步执行器有界收敛的唯一事实源；新的专用 +线程池不得复制 Future 追踪、worker join 或重试关闭实现。DoH 查询线程池也必须复用该 owner：恢复系统 +DNS 后有限等待，超时保留原 executor 并向 startup 返回 `False`，真实收敛前不得创建替代线程池或回填缓存。 协程环境文件日志属于有界 E1 观测能力，只允许单一队列 writer；队列满时不得再以无界 executor 形成第二条异步写入路径。日志关闭必须有限等待 writer 与文件处理器，未收敛时 `LoggerManager` 保留原 owner 并让 lifespan 以关闭失败结束，不得先清空引用或用无界 `join()` 掩盖失败。 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 701b27e92..d6a4790a9 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6560, - "edge_sha256": "bdd34affb7c42a4cbcdc85e713d9b4ac591c3343559dd5032b4518e77eba721e", + "edge_count": 6562, + "edge_sha256": "08b6e0005a3c316a1de675193f7749ff8a93e8c1c052eb48339aa58b308ad6bf", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -109,6 +109,7 @@ "app.adapters.network.doh -> app.foundation", "app.adapters.network.doh -> app.foundation.singleton", "app.adapters.network.doh -> app.runtime", + "app.adapters.network.doh -> app.runtime.execution", "app.adapters.network.doh -> app.runtime.log", "app.adapters.network.doh -> app.runtime.reload", "app.adapters.network.doh -> app.runtime.settings", @@ -5810,6 +5811,7 @@ "app.runtime.thread -> app.foundation", "app.runtime.thread -> app.foundation.singleton", "app.runtime.thread -> app.runtime", + "app.runtime.thread -> app.runtime.execution", "app.runtime.thread -> app.runtime.settings", "app.scheduler -> app.adapters", "app.scheduler -> app.adapters.external", diff --git a/tests/test_doh_helper.py b/tests/test_doh_helper.py index 1e0b165ee..11a8dd89e 100644 --- a/tests/test_doh_helper.py +++ b/tests/test_doh_helper.py @@ -1,6 +1,9 @@ import socket +import threading +import time from app.adapters.network import doh +from app.runtime.execution import OwnedThreadPoolExecutor def test_doh_executor_is_lazy_and_shutdown_restores_socket(monkeypatch): @@ -13,15 +16,15 @@ def test_doh_executor_is_lazy_and_shutdown_restores_socket(monkeypatch): monkeypatch.setattr(doh, "_orig_getaddrinfo", lambda host, *args, **kwargs: []) try: - helper.shutdown() + assert helper.shutdown() is True assert doh._executor is None - doh.enable_doh(True) + assert doh.enable_doh(True) is True socket.getaddrinfo("example.com", None) executor = doh._executor - assert executor is not None + assert isinstance(executor, OwnedThreadPoolExecutor) - helper.shutdown() + assert helper.shutdown() is True assert doh._executor is None assert socket.getaddrinfo is doh._orig_getaddrinfo @@ -31,6 +34,52 @@ def test_doh_executor_is_lazy_and_shutdown_restores_socket(monkeypatch): socket.getaddrinfo = original_getaddrinfo +def test_doh_shutdown_is_bounded_and_retryable(monkeypatch): + """阻塞查询超时时保留同一 owner，释放后可重试并安全重新启用。""" + original_getaddrinfo = socket.getaddrinfo + helper = object.__new__(doh.DohHelper) + entered = threading.Event() + release = threading.Event() + future = None + monkeypatch.setattr(doh, "_orig_getaddrinfo", lambda host, *args, **kwargs: []) + + def blocked_query() -> None: + """模拟底层网络栈未按 DoH 请求超时返回的同步查询。""" + entered.set() + release.wait() + + try: + assert helper.shutdown(timeout=1) is True + assert doh.enable_doh(True) is True + with doh._executor_lock: + executor = doh._get_executor_locked() + future = executor.submit(blocked_query) + assert entered.wait(timeout=1) + + started_at = time.monotonic() + assert helper.shutdown(timeout=0.01) is False + assert time.monotonic() - started_at < 1 + assert doh._executor is executor + assert socket.getaddrinfo is doh._orig_getaddrinfo + assert executor.accepting is False + + # 未收敛 owner 不得被新 executor 覆盖，否则旧查询会脱离生命周期追踪。 + assert doh.enable_doh(True) is False + assert doh._executor is executor + + release.set() + future.result(timeout=1) + assert helper.shutdown(timeout=1) is True + assert doh._executor is None + assert doh.enable_doh(True) is True + finally: + release.set() + if future is not None: + future.result(timeout=1) + helper.shutdown(timeout=1) + socket.getaddrinfo = original_getaddrinfo + + def test_doh_config_reload_disables_and_closes_executor(monkeypatch): """热更新关闭 DoH 时恢复系统 DNS 并释放已创建的线程池""" original_getaddrinfo = socket.getaddrinfo @@ -41,8 +90,8 @@ def test_doh_config_reload_disables_and_closes_executor(monkeypatch): monkeypatch.setattr(doh, "_orig_getaddrinfo", lambda host, *args, **kwargs: []) try: - helper.shutdown() - doh.enable_doh(True) + assert helper.shutdown() is True + assert doh.enable_doh(True) is True socket.getaddrinfo("example.com", None) executor = doh._executor assert executor is not None @@ -83,7 +132,7 @@ def test_enable_doh_reuses_cached_host_resolution(monkeypatch): doh._doh_cache.clear() try: - doh.enable_doh(True) + assert doh.enable_doh(True) is True socket.getaddrinfo("example.com", None) socket.getaddrinfo("example.com", None) diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index 7abab2fae..03e3f6439 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/tests/test_lifecycle_shutdown.py @@ -1001,6 +1001,18 @@ def test_stop_modules_propagates_shared_thread_pool_nonconvergence(monkeypatch): _assert_completed_once(dependency) +def test_stop_modules_propagates_doh_nonconvergence(monkeypatch): + """DoH 查询线程未终止时必须由模块服务关闭结果向上暴露。""" + dependencies = _patch_module_shutdown_dependencies(monkeypatch) + dependencies["doh"].return_value = False + + converged = asyncio.run(modules_initializer.stop_modules()) + + assert converged is False + for dependency in dependencies.values(): + _assert_completed_once(dependency) + + def test_stop_modules_drains_web_agent_tasks_before_persistence(monkeypatch): """关闭时先收口 Web Agent，再关闭持久化准入和数据库任务。""" order = [] diff --git a/tests/test_thread_helper_shutdown.py b/tests/test_thread_helper_shutdown.py index f1b37c4ca..fdf5ccc78 100644 --- a/tests/test_thread_helper_shutdown.py +++ b/tests/test_thread_helper_shutdown.py @@ -101,3 +101,14 @@ def test_thread_helper_shutdown_waits_for_worker_after_future_completion(): callback_release.set() assert executor.shutdown_bounded(timeout=1) is True + + +def test_owned_executor_preserves_standard_shutdown_contract(): + """旧调用方直接使用 pool.shutdown(wait=True) 时不得与 owner 回调互锁。""" + executor = thread_module._OwnedThreadPoolExecutor(max_workers=1) + future = executor.submit(lambda: "done") + + executor.shutdown(wait=True) + + assert future.result(timeout=1) == "done" + assert executor.accepting is False