From 138a48770c9e6547f652de4ad15a6748d01223ae Mon Sep 17 00:00:00 2001 From: jxxghp Date: Mon, 24 Aug 2026 06:55:50 +0800 Subject: [PATCH] refactor: unify durable event topics --- app/application/outbox.py | 42 ++++++++- app/application/subscription/delete.py | 3 +- app/application/subscription/identity.py | 8 +- app/application/subscription/mutation.py | 8 +- app/application/subscription/write.py | 4 +- app/chain/transfer.py | 5 +- app/db/adapters/chain.py | 8 +- app/startup/initializers/modules.py | 91 +++++++++++-------- .../backend-architecture-next-stage.md | 22 ++++- .../architecture/dependency-baseline.json | 7 +- tests/test_event_contracts.py | 38 ++++++-- 11 files changed, 172 insertions(+), 64 deletions(-) diff --git a/app/application/outbox.py b/app/application/outbox.py index ee5f3cfe7..b953fcccd 100644 --- a/app/application/outbox.py +++ b/app/application/outbox.py @@ -2,15 +2,43 @@ from __future__ import annotations -from collections.abc import Callable +from collections.abc import Callable, Mapping from dataclasses import dataclass from datetime import datetime, timedelta, timezone +from types import MappingProxyType from typing import Any, Protocol, TypeVar +from app.schemas.types import EventType + T = TypeVar("T") +SUBSCRIBE_ADDED_TOPIC = "subscribe.added" +SUBSCRIBE_MODIFIED_TOPIC = "subscribe.modified" +SUBSCRIBE_DELETED_TOPIC = "subscribe.deleted" +DOWNLOAD_ADDED_TOPIC = "download.added" +TRANSFER_COMPLETED_TOPIC = "transfer.completed" +TRANSFER_FAILED_TOPIC = "transfer.failed" + +DURABLE_EVENT_TOPICS: Mapping[EventType, str] = MappingProxyType({ + EventType.SubscribeAdded: SUBSCRIBE_ADDED_TOPIC, + EventType.SubscribeModified: SUBSCRIBE_MODIFIED_TOPIC, + EventType.SubscribeDeleted: SUBSCRIBE_DELETED_TOPIC, + EventType.DownloadAdded: DOWNLOAD_ADDED_TOPIC, + EventType.TransferComplete: TRANSFER_COMPLETED_TOPIC, + EventType.TransferFailed: TRANSFER_FAILED_TOPIC, +}) + + +def durable_event_topic(event_type: EventType) -> str: + """返回 durable-required 事件唯一登记的 outbox topic。""" + try: + return DURABLE_EVENT_TOPICS[event_type] + except KeyError as error: + raise ValueError(f"事件 {event_type.name} 未登记 durable topic") from error + + @dataclass(frozen=True, slots=True) class OutboxIntent: """与业务事务一起暂存的版本化副作用意图。""" @@ -33,6 +61,18 @@ class ClaimedOutboxMessage: attempt: int +def validate_durable_event_handlers( + handlers: Mapping[str, Callable[[ClaimedOutboxMessage], None]], +) -> None: + """拒绝缺少任一 durable-required 事件恢复 handler 的 dispatcher。""" + missing = set(DURABLE_EVENT_TOPICS.values()) - set(handlers) + if missing: + raise RuntimeError( + "Outbox dispatcher 缺少 durable 事件 handler: " + + ", ".join(sorted(missing)) + ) + + class OutboxRepository(Protocol): """outbox 写入、claim 和终态更新所需的最小端口。""" diff --git a/app/application/subscription/delete.py b/app/application/subscription/delete.py index 5a2177258..1149aaefd 100644 --- a/app/application/subscription/delete.py +++ b/app/application/subscription/delete.py @@ -12,6 +12,7 @@ from app.application.outbox import ( OutboxIntent, SyncOutboxTransaction, SyncUnitOfWork, + SUBSCRIBE_DELETED_TOPIC, ) from app.schemas.event import SubscribeDeletedEventData @@ -251,7 +252,7 @@ def _build_deletion_effects( report_payload=report_payload, event_intent=OutboxIntent( event_key=event_key, - topic="subscribe.deleted", + topic=SUBSCRIBE_DELETED_TOPIC, payload=event_payload, ), report_intent=OutboxIntent( diff --git a/app/application/subscription/identity.py b/app/application/subscription/identity.py index 063d4ae0d..3a2f49eeb 100644 --- a/app/application/subscription/identity.py +++ b/app/application/subscription/identity.py @@ -3,7 +3,11 @@ from datetime import datetime, timezone from typing import Any, Callable, Protocol -from app.application.outbox import AsyncOutboxTransaction, OutboxIntent +from app.application.outbox import ( + AsyncOutboxTransaction, + OutboxIntent, + SUBSCRIBE_DELETED_TOPIC, +) from app.application.subscription.delete import ( AsyncUnitOfWork, SubscribeDeletedPublisher, @@ -89,7 +93,7 @@ class DeleteSubscriptionsByIdentityCommand: await self._outbox.stage( OutboxIntent( event_key=event_payload["idempotency_key"], - topic="subscribe.deleted", + topic=SUBSCRIBE_DELETED_TOPIC, payload=event_payload, ), now, diff --git a/app/application/subscription/mutation.py b/app/application/subscription/mutation.py index 3aa96ed17..61afa86ff 100644 --- a/app/application/subscription/mutation.py +++ b/app/application/subscription/mutation.py @@ -7,7 +7,11 @@ from datetime import datetime, timezone from typing import Any, Protocol from uuid import uuid4 -from app.application.outbox import AsyncOutboxTransaction, OutboxIntent +from app.application.outbox import ( + AsyncOutboxTransaction, + OutboxIntent, + SUBSCRIBE_MODIFIED_TOPIC, +) from app.schemas.event import SubscribeModifiedEventData @@ -143,7 +147,7 @@ class SubscriptionMutationService: await self._outbox.stage( OutboxIntent( event_key=event_key, - topic="subscribe.modified", + topic=SUBSCRIBE_MODIFIED_TOPIC, payload=event_payload, ), datetime.now(timezone.utc), diff --git a/app/application/subscription/write.py b/app/application/subscription/write.py index ca34eec74..1c9a86ae7 100644 --- a/app/application/subscription/write.py +++ b/app/application/subscription/write.py @@ -19,7 +19,7 @@ from collections.abc import Awaitable, Callable from datetime import datetime, timezone from typing import Mapping, Optional, Protocol, Tuple -from app.application.outbox import OutboxIntent +from app.application.outbox import OutboxIntent, SUBSCRIBE_ADDED_TOPIC from app.domain.context import MediaInfo, MusicInfo from app.schemas.media import resolve_media_identity from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaType @@ -243,7 +243,7 @@ def _subscribe_added_intents( intents: list[OutboxIntent] = [ OutboxIntent( event_key=event_key, - topic="subscribe.added", + topic=SUBSCRIBE_ADDED_TOPIC, payload=event_payload, ), ] diff --git a/app/chain/transfer.py b/app/chain/transfer.py index 0132d514f..18cd56403 100755 --- a/app/chain/transfer.py +++ b/app/chain/transfer.py @@ -36,6 +36,7 @@ from app.application.history import (add_transfer_fail, add_transfer_success, clear_transfer_failures, describe_history_gate, evaluate_history_gate, is_skip_action, record_transfer_failure) +from app.application.outbox import TRANSFER_COMPLETED_TOPIC, TRANSFER_FAILED_TOPIC from app.runtime.log import logger from app.schemas.event import StorageOperSelectionEventData from app.schemas.transfer import TransferInfo @@ -456,7 +457,7 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo if durable_transfer_failed: event_payload = self._transfer_result_payload(task, transferinfo) history = self.durable_event_writer.transfer_result( - topic="transfer.failed", + topic=TRANSFER_FAILED_TOPIC, stage_history=lambda writer: add_transfer_fail( fileitem=task.fileitem, mode=transferinfo.transfer_type if transferinfo else "", @@ -569,7 +570,7 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo if durable_transfer_complete: event_payload = self._transfer_result_payload(task, transferinfo) history = self.durable_event_writer.transfer_result( - topic="transfer.completed", + topic=TRANSFER_COMPLETED_TOPIC, stage_history=lambda writer: add_transfer_success( fileitem=task.fileitem, mode=transferinfo.transfer_type if transferinfo else "", diff --git a/app/db/adapters/chain.py b/app/db/adapters/chain.py index 637882a30..c8b147afd 100644 --- a/app/db/adapters/chain.py +++ b/app/db/adapters/chain.py @@ -16,7 +16,11 @@ from app.application.chain.durable_events import ( transfer_result_event_key, ) from app.application.history import TransferHistoryRecord, TransferHistoryWriter -from app.application.outbox import DurableEventCommand, OutboxIntent +from app.application.outbox import ( + DurableEventCommand, + DOWNLOAD_ADDED_TOPIC, + OutboxIntent, +) from app.db.adapters.outbox import SqlAlchemyOutboxRepository from app.db.oper.downloadhistory import DownloadHistoryOper from app.db.oper.transferhistory import TransferHistoryOper @@ -88,7 +92,7 @@ class TransactionalChainDurableEventWriter(ChainDurableEventWriter): command.execute( intent=OutboxIntent( event_key=event_key, - topic="download.added", + topic=DOWNLOAD_ADDED_TOPIC, payload=snapshot_download_added(event_payload), ), stage_business=stage_business, diff --git a/app/startup/initializers/modules.py b/app/startup/initializers/modules.py index a49243a77..e69aa4680 100644 --- a/app/startup/initializers/modules.py +++ b/app/startup/initializers/modules.py @@ -83,7 +83,12 @@ from app.application.security.userconfig import ( configure_user_configuration, ) from app.application.history import configure_transfer_history_provider -from app.application.outbox import OutboxDispatcher, configure_outbox_dispatcher +from app.application.outbox import ( + OutboxDispatcher, + configure_outbox_dispatcher, + durable_event_topic, + validate_durable_event_handlers, +) from app.db.adapters.outbox import SqlAlchemyAsyncOutboxStager, SqlAlchemyOutboxRepository from app.application.site.query import SiteQueryService, configure_site_query_service from app.application.site.health import SiteHealthService, configure_site_health_service @@ -335,44 +340,58 @@ def _build_outbox_dispatcher() -> OutboxDispatcher: raise RuntimeError("订阅新增通知快照格式无效") CommandChain().post_message(Message.model_validate(snapshot)) + handlers = { + durable_event_topic( + EventType.SubscribeAdded + ): lambda message: EventManager().send_event( + EventType.SubscribeAdded, + message.payload, + ), + "subscribe.added.report": dispatch_subscribe_added_report, + "subscribe.added.notification": dispatch_subscribe_added_notification, + durable_event_topic( + EventType.SubscribeModified + ): lambda message: EventManager().send_event( + EventType.SubscribeModified, + message.payload, + ), + durable_event_topic( + EventType.SubscribeDeleted + ): lambda message: EventManager().send_event( + EventType.SubscribeDeleted, + message.payload, + ), + "subscribe.deleted.report": dispatch_subscribe_deleted_report, + "subscribe.complete": lambda message: EventManager().send_event( + EventType.SubscribeComplete, + message.payload, + ), + "subscribe.complete.report": dispatch_subscribe_complete_report, + "subscribe.complete.notification": dispatch_subscribe_notification, + durable_event_topic( + EventType.DownloadAdded + ): lambda message: EventManager().send_event( + EventType.DownloadAdded, + restore_download_added(message.payload), + ), + durable_event_topic( + EventType.TransferComplete + ): lambda message: EventManager().send_event( + EventType.TransferComplete, + restore_transfer_result(message.payload), + ), + durable_event_topic( + EventType.TransferFailed + ): lambda message: EventManager().send_event( + EventType.TransferFailed, + restore_transfer_result(message.payload), + ), + } + validate_durable_event_handlers(handlers) session = SessionFactory() return OutboxDispatcher( repository=SqlAlchemyOutboxRepository(session), - handlers={ - "subscribe.added": lambda message: EventManager().send_event( - EventType.SubscribeAdded, - message.payload, - ), - "subscribe.added.report": dispatch_subscribe_added_report, - "subscribe.added.notification": dispatch_subscribe_added_notification, - "subscribe.modified": lambda message: EventManager().send_event( - EventType.SubscribeModified, - message.payload, - ), - "subscribe.deleted": lambda message: EventManager().send_event( - EventType.SubscribeDeleted, - message.payload, - ), - "subscribe.deleted.report": dispatch_subscribe_deleted_report, - "subscribe.complete": lambda message: EventManager().send_event( - EventType.SubscribeComplete, - message.payload, - ), - "subscribe.complete.report": dispatch_subscribe_complete_report, - "subscribe.complete.notification": dispatch_subscribe_notification, - "download.added": lambda message: EventManager().send_event( - EventType.DownloadAdded, - restore_download_added(message.payload), - ), - "transfer.completed": lambda message: EventManager().send_event( - EventType.TransferComplete, - restore_transfer_result(message.payload), - ), - "transfer.failed": lambda message: EventManager().send_event( - EventType.TransferFailed, - restore_transfer_result(message.payload), - ), - }, + handlers=handlers, close=session.close, failure_observer=lambda dead: record_metric( "scheduler.job.dead_letter" if dead else "scheduler.job.retry", diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 70b60fc9f..8ee1b7ae8 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 运行时的获取路径。 +> 实施进度：阶段 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 事实源。 ## 当前复核结论（2026-08-24） @@ -354,13 +354,27 @@ - 依赖边集合按新门面路径刷新，模块数仍为 `806`、内部边仍为 `6546`；12 组禁止边 与唯一隔离 TMDB SCC 均未变化。 +### 长期整改阶段 34：durable-required 事件事实源收口（2026-08-24） + +- 当前 Event Contract 将订阅新增/修改/删除、下载添加、整理成功/失败六个事件标为 + `durable_required`；它们的宿主正式生产者早已与业务写入同事务暂存 Outbox intent，但 topic + 字符串在订阅、Chain adapter 和 startup dispatcher 中分散重复，当前复核结论仍误写成六个事件待实现。 +- `DURABLE_EVENT_TOPICS` 现在是六个 EventType 与版本化 topic 的单一映射；订阅命令、下载/整理 + writer 和恢复 dispatcher 共用该映射，不再各自维护同义字符串。启动恢复器也会拒绝缺失任一 + durable topic handler 的配置。 +- 事件契约测试保证全部 `durable_required` 事件与 topic 键集合一致、topic 不重复，并冻结恢复 + handler 完整性。原 topic、payload、幂等键、at-least-once 语义、无 writer 的测试/嵌入式兼容分支 + 以及插件 SDK/Compat 均未修改；第三方插件自行写库或发事件仍不在宿主事务边界内。 +- 单一映射新增 `app.application.outbox -> app.schemas.types` 与 `app.chain.transfer -> app.application.outbox` + 语义边，模块仍为 `806`、内部边为 `6549`；12 组禁止边与唯一隔离 TMDB SCC 均未变化。 + ### 总体判断 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**： - 继续采用单进程控制面是正确选择，不建议现在拆成微服务；插件、调度器、工作流、事件和数据库共享进程内状态，拆分会放大部署、事务和兼容成本。 - `foundation/domain/runtime/adapters/application/chain/api/startup` 的职责方向基本成立；宿主架构基线、复杂度 ratchet、异步阻塞 ratchet 当前均通过。 -- 依赖图当前为 `806` 个 Python 模块、`6546` 条内部导入边；唯一非平凡 SCC 位于隔离的 TMDB 第三方移植包内部，不应为了指标归零重写。 +- 依赖图当前为 `806` 个 Python 模块、`6549` 条内部导入边；唯一非平凡 SCC 位于隔离的 TMDB 第三方移植包内部，不应为了指标归零重写。 - 当前主要风险已经从“目录和依赖失控”转移到运行时协议、后台副作用的可靠性和遗留兼容面。换言之，下一阶段重点应是**语义收口和可验证性**，而不是继续搬文件或机械拆大文件。 综合评价：架构方向可持续，生产可用性较高；可演进性仍处于中等水平。现阶段没有静态审计发现必须立即推倒重来的 P0 架构问题，但存在需要按 P1/P2 计划治理的真实债务。 @@ -471,8 +485,8 @@ MoviePilot V3 当前不是“目录混乱、必须推倒重来”的状态。第 2. **V3 部署拓扑边界已完成。**全功能模式在 startup、launcher 和 Doctor 共同拒绝 `API_WORKERS > 1`，生产入口固定单 worker；开发 reload/监督模式使用 `app.factory:create_app` import-string factory，不再把 app 实例交给多进程 supervisor。旧配置键继续可解析，未来只有拆出 control role 后才重新评估全功能多 worker。 3. **事务所有权已完成装饰器层收口，但 ORM 对象跨层流转仍需治理。**正式 Model 查询/写装饰器均已清零，宿主 Oper 查询统一接收显式 Session；调用方仍需继续明确 ORM 对象生命周期、懒加载和业务提交后副作用边界。 4. **组合根之后仍存在全局服务定位，但配置和 API 数据主路径已收口。**canonical 未批准 Settings 导入与非组合根 `SystemConfigOper()` 构造均为 `0`；数据库基础设施 3 处和 startup 唯一构造点作为不可扩张边界登记。正式 FastAPI 依赖只读取 AppState `HostRuntime` 的命名领域，字符串 API 数据注册表仅允许 startup 注入和旧 Facade 转发；后续对象是 Singleton 与模块级 `configure/get` provider，不应再迁移已类型化 API 依赖。 -5. **模块与事件契约登记均已完成。**当前 212 个模块 spec 的宿主观察面已无 legacy aggregation，53 个事件全部绑定 typed payload，可见性、投递等级、错误行为和敏感字段均有基线，legacy event payload 为 `0`。后续重点是保持新增能力 ratchet、观察未知第三方 fallback 命中，以及 6 个 durable-required 事件的真实持久投递，不是重复创建契约或事件 DTO。 -6. **后台副作用缺少统一可靠性定义。**事件队列、APScheduler、FastAPI BackgroundTasks 和线程池任务的丢失、重试、幂等、关停语义各不相同；数据库提交与事件/上报之间仍有进程崩溃窗口。 +5. **模块与事件契约登记均已完成。**当前 212 个模块 spec 的宿主观察面已无 legacy aggregation，53 个事件全部绑定 typed payload，可见性、投递等级、错误行为和敏感字段均有基线，legacy event payload 为 `0`。六个 durable-required 事件的宿主正式生产者已通过业务同事务 Outbox 提供真实持久投递，事件与 topic 映射及恢复 handler 完整性已纳入 ratchet。后续重点是保持新增能力门禁和观察未知第三方 fallback 命中，不是重复创建契约、DTO 或 Outbox。 +6. **后台副作用已有统一可靠性分类，但其他 E1/E3 机制仍需逐项收口。** ADR-0007 已分类事件队列、APScheduler、进程内任务、Agent task 与 transfer pending 的完成点、恢复和失败表达；不能因六个关键事件已接 Outbox，就把仍需定时重建、持久任务表或人工恢复的其他 E1/E3 机制误报为全部完成。 7. **核心关联与健康边界已落地，指标导出仍未收口。**HTTP/SSE correlation ID 已传播到线程池、事件、工作流、子进程、外部请求和日志；`/health/live`、`/health/ready` 已由部署入口消费，事件/数据库队列深度及模块/事件耗时使用低基数指标登记。当前缺口是稳定 exporter、运维查询面和跨进程聚合，而不是重新实现 request ID 或健康路由。 8. **质量门禁已具备增量硬约束，但覆盖面仍需扩大。**push/PR 对变更 Python 文件执行 Pylint，CI 同时运行 host architecture、39 个 strict mypy 文件、复杂度、async 阻塞和 task owner ratchet；全仓 Pylint 仍是 advisory，strict 类型和复杂度拆分仍应随业务切片渐进扩展。 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 98c78e450..a4f7adc60 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": 6546, - "edge_sha256": "64bd673bda7224d371b91e229d23eeea6ac14655301357a209dff510e30e151b", + "edge_count": 6549, + "edge_sha256": "280799d1a7d3a993096834b5e0f961fcc82319d437b4d5f568c78d8a7bfa07a2", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -2699,6 +2699,8 @@ "app.application.notification -> app.schemas", "app.application.notification -> app.schemas.system", "app.application.notification -> app.schemas.types", + "app.application.outbox -> app.schemas", + "app.application.outbox -> app.schemas.types", "app.application.plugin.config -> app.schemas", "app.application.plugin.config -> app.schemas.exception", "app.application.plugin.folders -> app.application", @@ -3432,6 +3434,7 @@ "app.chain.transfer -> app.application.directory", "app.chain.transfer -> app.application.formatting", "app.chain.transfer -> app.application.history", + "app.chain.transfer -> app.application.outbox", "app.chain.transfer -> app.application.transfer", "app.chain.transfer -> app.chain", "app.chain.transfer -> app.chain._transfer", diff --git a/tests/test_event_contracts.py b/tests/test_event_contracts.py index 281ed084e..2271033f3 100644 --- a/tests/test_event_contracts.py +++ b/tests/test_event_contracts.py @@ -2,6 +2,12 @@ from unittest.mock import patch +import pytest + +from app.application.outbox import ( + DURABLE_EVENT_TOPICS, + validate_durable_event_handlers, +) from app.runtime.event.contracts import ( EVENT_CONTRACTS, EventDelivery, @@ -69,16 +75,28 @@ def test_invalid_typed_payload_is_diagnostic_only() -> None: def test_selected_user_side_effects_are_marked_durable_required() -> None: - """订阅、下载和整理完成事件必须明确暴露后续 durable pilot 要求。""" - for event_type in ( - EventType.SubscribeAdded, - EventType.SubscribeModified, - EventType.SubscribeDeleted, - EventType.DownloadAdded, - EventType.TransferComplete, - EventType.TransferFailed, - ): - assert get_event_contract(event_type).delivery is EventDelivery.DURABLE_REQUIRED + """所有 durable-required 事件必须一一登记唯一的恢复 topic。""" + durable_events = { + event_type + for event_type, contract in EVENT_CONTRACTS.items() + if contract.delivery is EventDelivery.DURABLE_REQUIRED + } + + assert set(DURABLE_EVENT_TOPICS) == durable_events + assert len(set(DURABLE_EVENT_TOPICS.values())) == len(durable_events) + + +def test_durable_event_dispatcher_requires_every_recovery_handler() -> None: + """恢复 dispatcher 缺少任一登记 topic 时必须在构造边界失败。""" + handlers = { + topic: lambda _message: None + for topic in DURABLE_EVENT_TOPICS.values() + } + validate_durable_event_handlers(handlers) + + handlers.pop(next(iter(DURABLE_EVENT_TOPICS.values()))) + with pytest.raises(RuntimeError, match="缺少 durable 事件 handler"): + validate_durable_event_handlers(handlers) def test_download_and_transfer_typed_contracts_accept_legacy_runtime_objects() -> None: