diff --git a/app/application/security/url.py b/app/application/security/url.py index bef0444db..0e2ebe274 100644 --- a/app/application/security/url.py +++ b/app/application/security/url.py @@ -950,6 +950,11 @@ _image_proxy_block_log_coalescer = EventCoalescer( ) +async def close_image_proxy_block_log_coalescer() -> None: + """刷新图片代理阻断摘要,并等待已经启动的窗口 flush 任务。""" + await _image_proxy_block_log_coalescer.close() + + async def _emit_image_proxy_block_warning( *, url: str, diff --git a/app/runtime/coalesce.py b/app/runtime/coalesce.py index ec774a610..ba4cc9988 100644 --- a/app/runtime/coalesce.py +++ b/app/runtime/coalesce.py @@ -97,6 +97,7 @@ class EventCoalescer: self._source = source self._buckets: Dict[Hashable, _BucketState] = {} self._is_flush_async = inspect.iscoroutinefunction(on_flush) + self._flush_tasks: set[asyncio.Task[None]] = set() @property def window_seconds(self) -> float: @@ -144,6 +145,12 @@ class EventCoalescer: if bucket.flush_handle is not None: bucket.flush_handle.cancel() await self._emit_summary_if_needed(key, bucket) + current_task = asyncio.current_task() + flush_tasks = tuple( + task for task in self._flush_tasks if task is not current_task + ) + if flush_tasks: + await asyncio.gather(*flush_tasks, return_exceptions=True) def _schedule_flush(self, key: Hashable) -> asyncio.TimerHandle: """ @@ -161,7 +168,9 @@ class EventCoalescer: `loop.call_later` 到期回调:从事件循环里把异步 flush 任务接力起来。 """ try: - asyncio.get_running_loop().create_task(self._flush_key(key)) + task = asyncio.get_running_loop().create_task(self._flush_key(key)) + self._flush_tasks.add(task) + task.add_done_callback(self._flush_tasks.discard) except RuntimeError as exc: # 事件循环已关闭等罕见路径:记录后丢弃,避免影响其它 bucket self._log_debug(f"flush 调度失败,已忽略 key={key!r}: {exc}") diff --git a/app/startup/initializers/modules.py b/app/startup/initializers/modules.py index c093b4cc0..a49243a77 100644 --- a/app/startup/initializers/modules.py +++ b/app/startup/initializers/modules.py @@ -77,6 +77,7 @@ from app.application.messaging.agent import ( from app.application.security.user import configure_user_lookups from app.application.security.auth import AuthService, configure_auth_service from app.application.security.passkeys import PasskeyService, configure_passkey_service +from app.application.security.url import close_image_proxy_block_log_coalescer from app.application.security.userconfig import ( UserConfigurationService, configure_user_configuration, @@ -594,6 +595,7 @@ async def stop_modules(): logger.error(f"关闭{name}失败:{err}") return True + await run_step("图片代理安全日志合并器", close_image_proxy_block_log_coalescer) await run_step("模块", lambda: ModuleManager().shutdown()) await run_step("事件消费", lambda: EventManager().stop_async()) await run_step("浏览器会话", close_browser_sessions) diff --git a/docs/adr/0007-background-action-reliability.md b/docs/adr/0007-background-action-reliability.md index 571f2c4da..4c928d015 100644 --- a/docs/adr/0007-background-action-reliability.md +++ b/docs/adr/0007-background-action-reliability.md @@ -52,7 +52,8 @@ Event Contract Registry 是 53 个事件的逐项机器清单。下表按相同 - `SubscribeAdded`、`SubscribeModified`、`SubscribeDeleted`:订阅业务行 commit 是业务完成点;事件、 通知和服务端上报必须由同事务 durable intent 驱动。ARCH-251 首选 `SubscribeAdded` pilot。 -- `DownloadAdded`:下载提交成功后,历史/通知不得仅依赖进程内回调;后续独立 pilot。 +- `DownloadAdded`:下载器确认接收后,下载历史与事件 intent 已在返回前原子提交;通知和模块后处理只在 + commit 后启动,事件由 Outbox 恢复投递。 - `TransferComplete`、`TransferFailed`:整理步骤本身属于 E3,但向事件消费者发布结果属于 E2。 ## 非 Event 后台机制映射 @@ -66,6 +67,8 @@ Event Contract Registry 是 53 个事件的逐项机器清单。下表按相同 - Slack、Telegram、Discord、飞书、QQBot、企业微信与 WeChatClawBot 的渠道回环统一经 `application.messaging.ingress` 进入同一个 API/TaskRegistry 主链;需要立即返回 SDK 回调的渠道把同步 HTTP 交给宿主共享线程池,模块关闭后由线程池生命周期等待,不再创建逐消息 daemon 线程。 +- 图片代理安全日志的窗口聚合属于 E1 观测;`EventCoalescer` 持有到期 flush task,模块关闭会取消未到期 + timer、刷新剩余摘要并等待已启动回调,不再把 `create_task` 留给事件循环隐式回收。 - 主仓不再新增或保留裸 FastAPI `BackgroundTasks`;若任务源于已提交的用户数据且不可从数据库重建, 必须提升为 E2,进入 Outbox 或持久任务表。 diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index b48ddffda..1554f3d81 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 已统一七种消息渠道的宿主回环与后台执行边界。 +> 实施进度:阶段 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 已补齐事件窗口聚合任务的生命周期所有权。 ## 当前复核结论(2026-08-24) @@ -15,7 +15,7 @@ ### 长期整改阶段 0:治理门禁恢复(2026-08-23) -- 宿主依赖基线已审查 TaskRegistry、有界后台 owner 与插件变更准入接入后的语义差异:当前为 `806` 个模块、`6528` 条内部导入边,12 组重点禁止边继续全部为 `0`,唯一非平凡 SCC 仍是隔离的 TMDB 移植包。 +- 宿主依赖基线已审查 TaskRegistry、有界后台 owner 与插件变更准入接入后的语义差异:当前为 `806` 个模块、`6529` 条内部导入边,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__` 等实现期导入不再被误冻结,既有数据库备份门面已补精确导出清单。 @@ -112,7 +112,7 @@ - 继续采用单进程控制面是正确选择,不建议现在拆成微服务;插件、调度器、工作流、事件和数据库共享进程内状态,拆分会放大部署、事务和兼容成本。 - `foundation/domain/runtime/adapters/application/chain/api/startup` 的职责方向基本成立;宿主架构基线、复杂度 ratchet、异步阻塞 ratchet 当前均通过。 -- 依赖图当前为 `806` 个 Python 模块、`6528` 条内部导入边;唯一非平凡 SCC 位于隔离的 TMDB 第三方移植包内部,不应为了指标归零重写。 +- 依赖图当前为 `806` 个 Python 模块、`6529` 条内部导入边;唯一非平凡 SCC 位于隔离的 TMDB 第三方移植包内部,不应为了指标归零重写。 - 当前主要风险已经从“目录和依赖失控”转移到运行时协议、后台副作用的可靠性和遗留兼容面。换言之,下一阶段重点应是**语义收口和可验证性**,而不是继续搬文件或机械拆大文件。 综合评价:架构方向可持续,生产可用性较高;可演进性仍处于中等水平。现阶段没有静态审计发现必须立即推倒重来的 P0 架构问题,但存在需要按 P1/P2 计划治理的真实债务。 @@ -134,6 +134,8 @@ 与业务删除原子暂存事件和统计 intent;旧类方法只作为插件 ABI 保留,不纳入宿主可靠性证明。 消息渠道回环也不再各自拥有 URL、HTTP 判定和逐消息线程:七种宿主渠道共享 ingress,三种立即返回 渠道交给生命周期持有的共享线程池;这仍是 E0 投递,不宣称跨进程恢复。 + 图片代理安全日志的窗口聚合也已补齐内部任务所有权:timer 到期创建的 flush task 由 coalescer 持有, + `stop_modules()` 会刷新未到期摘要并等待已启动回调;它属于 E1 观测,不扩大 TaskRegistry 或 Outbox 范围。 2. **Model/Base 的数据库装饰器和隐式会话 ABI 已全部清零。** 查询、写事务和 `legacy_*` 装饰器均为 `0`;所有 Model `db` 参数要求显式 Session,Base CRUD 仅在调用方事务内查询或 stage。可无会话构造的入口统一留在 Oper,经组合根事务执行器运行;插件 SDK 不再导出宿主 Model。后续重点转为减少 ORM 对象跨层流转,并保持 Model 隐式事务零回退。 Oper 内部的执行入口也已统一:最后一处 `AgentTaskOper` 直接 transaction runner 调用已迁入 diff --git a/mypy.ini b/mypy.ini index 6cbcaee20..4316ae754 100644 --- a/mypy.ini +++ b/mypy.ini @@ -12,6 +12,7 @@ files = app/domain/torrent.py, app/domain/meta/infopath.py, app/runtime/correlation.py, + app/runtime/coalesce.py, app/runtime/observability/__init__.py, app/runtime/event/contracts.py, app/runtime/event/errors.py, diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 4afdaa287..fa7572810 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": 6528, - "edge_sha256": "2e30c775e78f578d804f69972ab1348e1e9d0872b00bdbfcffad08398fae68ac", + "edge_count": 6529, + "edge_sha256": "daf8459c6194e28766d62c5f16a0e1bb64caf3f3430f51736b57da505e4401e1", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -6183,6 +6183,7 @@ "app.startup.initializers.modules -> app.application.security", "app.startup.initializers.modules -> app.application.security.auth", "app.startup.initializers.modules -> app.application.security.passkeys", + "app.startup.initializers.modules -> app.application.security.url", "app.startup.initializers.modules -> app.application.security.user", "app.startup.initializers.modules -> app.application.security.userconfig", "app.startup.initializers.modules -> app.application.server", diff --git a/tests/test_coalesce.py b/tests/test_coalesce.py index 1df9d3977..0e8fcee72 100644 --- a/tests/test_coalesce.py +++ b/tests/test_coalesce.py @@ -171,6 +171,28 @@ class EventCoalescerTest(IsolatedAsyncioTestCase): self.assertEqual(len(awaited), 1) self.assertEqual(awaited[0].count, 2) + async def test_close_waits_for_started_flush_task(self): + """窗口任务已经进入异步回调时,close 必须等待它真正结束。""" + started = asyncio.Event() + release = asyncio.Event() + + async def on_flush(_summary: CoalesceSummary) -> None: + """阻塞 flush 回调,暴露关闭与在途任务之间的时序。""" + started.set() + await release.wait() + + coalescer = EventCoalescer(_TEST_WINDOW, on_flush) + await coalescer.record("k", payload="a") + await coalescer.record("k", payload="b") + await asyncio.wait_for(started.wait(), timeout=_TEST_WAIT) + + close_task = asyncio.create_task(coalescer.close()) + await asyncio.sleep(0) + self.assertFalse(close_task.done()) + + release.set() + await asyncio.wait_for(close_task, timeout=_TEST_WAIT) + async def test_on_flush_exception_is_swallowed(self): """ on_flush 抛异常不能影响 coalescer 自身或上层调用方,仅 debug 记录。 diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index 952416fe5..c57b9ff4a 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/tests/test_lifecycle_shutdown.py @@ -1120,6 +1120,16 @@ def _patch_module_shutdown_dependencies(monkeypatch) -> dict: monkeypatch.setattr(modules_initializer, name, dependency) dependencies[name] = dependency + close_image_proxy_block_log_coalescer = AsyncMock() + monkeypatch.setattr( + modules_initializer, + "close_image_proxy_block_log_coalescer", + close_image_proxy_block_log_coalescer, + ) + dependencies["close_image_proxy_block_log_coalescer"] = ( + close_image_proxy_block_log_coalescer + ) + stop_managed_resources = AsyncMock() monkeypatch.setattr( modules_initializer, @@ -1158,6 +1168,15 @@ def test_browser_sessions_close_before_managed_resources(monkeypatch) -> None: assert calls == ["browser", "resources"] +def test_module_shutdown_waits_for_image_proxy_log_coalescer(monkeypatch) -> None: + """模块关闭必须等待图片安全日志的在途聚合任务收口。""" + dependencies = _patch_module_shutdown_dependencies(monkeypatch) + + asyncio.run(modules_initializer.stop_modules()) + + dependencies["close_image_proxy_block_log_coalescer"].assert_awaited_once_with() + + def test_shared_http_close_waits_for_real_lru_eviction(monkeypatch): """最终 HTTP 关闭必须等待真实 LRU 淘汰任务并消费其异常"""