From e2814a129fb83c2f08935821b76e158c76d1448c Mon Sep 17 00:00:00 2001 From: jxxghp Date: Mon, 24 Aug 2026 08:16:54 +0800 Subject: [PATCH] fix: drain cancelled debounce tasks --- app/runtime/debounce.py | 34 ++++++++--- .../backend-architecture-next-stage.md | 10 +++- tests/test_debounce.py | 59 +++++++++++++++++++ 3 files changed, 95 insertions(+), 8 deletions(-) create mode 100644 tests/test_debounce.py diff --git a/app/runtime/debounce.py b/app/runtime/debounce.py index 6151148dd..893ca342b 100644 --- a/app/runtime/debounce.py +++ b/app/runtime/debounce.py @@ -184,9 +184,19 @@ class AsyncDebouncer(BaseDebouncer): """ super().__init__(*args, **kwargs) self.task: Optional[asyncio.Task] = None + self._retired_tasks: set[asyncio.Task[Any]] = set() self.lock = asyncio.Lock() self.is_cooling_down = False + def _cancel_active_task(self) -> None: + """取消当前任务并保留 owner，直到任务进入真实终态。""" + task = self.task + if task is None or task.done(): + return + task.cancel() + self._retired_tasks.add(task) + task.add_done_callback(self._retired_tasks.discard) + async def __call__(self, *args, **kwargs) -> None: """ 异步调用防抖函数。 @@ -205,8 +215,7 @@ class AsyncDebouncer(BaseDebouncer): self.log_info("前沿模式 (async): 立即执行协程。") await self.func(*args, **kwargs) - if self.task and not self.task.done(): - self.task.cancel() + self._cancel_active_task() self.is_cooling_down = True self.task = asyncio.create_task(self._end_cool_down()) @@ -226,7 +235,7 @@ class AsyncDebouncer(BaseDebouncer): 后沿模式的逻辑。 """ if self.task and not self.task.done(): - self.task.cancel() + self._cancel_active_task() self.log_debug("后沿模式 (async): 检测到新的调用，已取消旧任务。") self.task = asyncio.create_task(self._delayed_execute(*args, **kwargs)) @@ -246,12 +255,23 @@ class AsyncDebouncer(BaseDebouncer): async def cancel(self) -> None: """ - 取消任何挂起的调用，并重置状态。 + 取消并等待所有挂起调用进入终态，再重置状态。 """ async with self.lock: - if self.task and not self.task.done(): - self.task.cancel() - self.task = None + current_task = asyncio.current_task() + tasks = { + task + for task in (*self._retired_tasks, self.task) + if task is not None and task is not current_task + } + for task in tasks: + if not task.done(): + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + self._retired_tasks.difference_update(tasks) + self.task = None + if tasks: self.log_info("异步防抖器被手动取消。") self.is_cooling_down = False diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 874a012d1..5a1f60b89 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -6,7 +6,7 @@ > 审计范围：宿主后端；排除 `app/plugins/**` 运行时插件副本 > 规范优先级：`AGENTS.md` 与 `docs/rules/` 高于本文 > 相关文档：`docs/architecture-overview.md`、`docs/refactor/backend-architecture-governance.md`、`docs/refactor/backend-module-refactor-compatibility.md` -> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口；阶段 4 已统一七种消息渠道的宿主回环与后台执行边界；阶段 5 已补齐事件窗口聚合任务的生命周期所有权；阶段 6 已统一插件文件操作的取消完成语义；阶段 7 已统一插件协程补偿的终态等待；阶段 8 已统一宿主同步函数的异步线程池入口；阶段 9 已统一工作流运行时的宿主获取路径；阶段 10 已统一模块、插件与调度运行时的显式 getter 调用；阶段 11 已清除系统配置 getter 的 Oper 形别名；阶段 12 已完成工作流域的显式 Chain 数据端口迁移；阶段 13 已收口用户、交互与消息链的数据端口；阶段 14 已收口音乐订阅数据端口；阶段 15 已收口站点数据端口；阶段 16 已收口媒体服务器数据端口；阶段 17 已收口下载数据端口；阶段 18 已收口主订阅数据端口；阶段 19 已收口整理数据端口；阶段 20 已收口 Agent 数据端口；阶段 21 已收口监控历史端口；阶段 22 已统一服务配置应用边界；阶段 23 已补齐媒体服务器 API 遗留的类形配置读取路径；阶段 24 已清除 Scheduler 内部无 owner 的协程提交双轨；阶段 25 已补齐 TaskRegistry 跨线程 owner 并迁移整理 AI 接管；阶段 26 已统一 Agent 会话清理提交；阶段 27 已统一历史 AI 进度 owner；阶段 28 已托管旧插件订阅统计线程；阶段 29 已统一 Emby 系条目转换并清零重复代码白名单；阶段 30 已收口插件市场请求级子任务；阶段 31 已托管搜索 AI 推荐任务；阶段 32 已清除事件调度器绕过生命周期 owner 的投递回退；阶段 33 已统一宿主 Agent 运行时的获取路径；阶段 34 已统一 durable-required 事件与 Outbox topic 事实源；阶段 35 已统一 LLM provider 管理 API 的运行时解析路径；阶段 36 已统一 WebAgent 音频能力访问边界；阶段 37 已统一插件输入事件发布路径；阶段 38 已统一 WebAgent 通知事件监听与队列边界；阶段 39 已补齐搜索 SSE 断线时的上游任务清理。 +> 实施进度：阶段 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 已补齐异步防抖取消的终态所有权。 ## 当前复核结论（2026-08-24） @@ -418,6 +418,14 @@ - 回归测试锁定客户端在首个事件到达时断开后，上游 `finally` 已在流函数返回前执行；心跳、批量 append/replace、字幕签名、SSE payload 与缓存头均未修改，V1/V2/V3 插件合同不受影响。 +### 长期整改阶段 40：异步防抖取消终态收口（2026-08-24） + +- `AsyncDebouncer` 在后续调用替换旧任务时只发出取消并覆盖句柄，公开 `cancel()` 也在任务真正退出前 + 清空引用；现在被替换的任务保留为 retired owner，`cancel()` 会取消并等待活动任务和 retired 任务 + 全部进入终态后再返回。 +- 回归测试覆盖旧任务在取消清理中阻塞、后续任务同时存在的场景。`debounce()` 装饰器、leading/trailing + 触发规则、同步 `Debouncer`、`app.utils.debounce` 兼容映射和 V1/V2/V3 插件导入路径均未修改。 + ### 总体判断 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**： diff --git a/tests/test_debounce.py b/tests/test_debounce.py new file mode 100644 index 000000000..8d750f57e --- /dev/null +++ b/tests/test_debounce.py @@ -0,0 +1,59 @@ +"""同步与异步防抖生命周期测试。""" + +import asyncio + +import pytest + +from app.runtime.debounce import AsyncDebouncer + + +@pytest.mark.asyncio +async def test_async_debouncer_cancel_waits_for_replaced_task_cleanup() -> None: + """公开取消必须等待已被后续调用替换的任务完成取消清理。""" + first_started = asyncio.Event() + first_cancelled = asyncio.Event() + allow_first_cleanup = asyncio.Event() + + async def delayed_work(value: str) -> None: + """让首个调用在取消后停留，暴露 retired task 的终态时序。""" + if value != "first": + await asyncio.Event().wait() + return + first_started.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + first_cancelled.set() + await allow_first_cleanup.wait() + raise + + debouncer = AsyncDebouncer(delayed_work, interval=0) + await debouncer("first") + await first_started.wait() + + await debouncer("second") + await first_cancelled.wait() + cancel_task = asyncio.create_task(debouncer.cancel()) + await asyncio.sleep(0) + + assert cancel_task.done() is False + allow_first_cleanup.set() + await asyncio.wait_for(cancel_task, timeout=1) + assert debouncer.task is None + assert debouncer._retired_tasks == set() + + +@pytest.mark.asyncio +async def test_async_debouncer_cancel_is_idempotent_without_active_task() -> None: + """没有活动任务时重复取消应立即返回并保持冷却状态关闭。""" + + async def no_op() -> None: + """提供不会被实际调度的异步函数。""" + + debouncer = AsyncDebouncer(no_op, interval=1, leading=True) + + await debouncer.cancel() + await debouncer.cancel() + + assert debouncer.task is None + assert debouncer.is_cooling_down is False