diff --git a/app/application/messaging/subscribe.py b/app/application/messaging/subscribe.py index f06222fc5..f9bc7d750 100644 --- a/app/application/messaging/subscribe.py +++ b/app/application/messaging/subscribe.py @@ -23,10 +23,18 @@ class SubscribeInteractionActions(Protocol): 声明订阅交互需要调用的业务动作。 """ - def refresh(self): + def refresh(self) -> Any: """执行订阅刷新。""" ... + def check(self) -> Any: + """执行订阅元数据检查。""" + ... + + def search(self, **kwargs: Any) -> Any: + """按消息入口参数执行订阅搜索。""" + ... + class SubscribeInteractionRepository(Protocol): """订阅消息交互所需的同步数据端口。""" @@ -37,14 +45,11 @@ class SubscribeInteractionRepository(Protocol): def get(self, subscribe_id: int) -> Optional[Any]: """按 ID 返回订阅。""" - def delete(self, subscribe_id: int) -> Any: - """删除订阅。""" - - def check(self): + def check(self) -> Any: """执行订阅元数据检查。""" ... - def search(self, **kwargs): + def search(self, **kwargs: Any) -> Any: """执行订阅搜索。""" ... @@ -62,15 +67,15 @@ class SubscribeInteractionHandler: messenger: MessageGateway, actions: SubscribeInteractionActions, repository: SubscribeInteractionRepository, - report_deleted: Callable[[dict], Any], - ): + delete_subscription: Callable[[int], bool], + ) -> None: """ 注入消息投递接口和订阅业务动作。 """ self._messenger = messenger self._actions = actions self._repository = repository - self._report_deleted = report_deleted + self._delete_subscription = delete_subscription def remote_list( self, @@ -727,15 +732,10 @@ class SubscribeInteractionHandler: if not subscribe: missing.append(str(subscribe_id)) continue + if not self._delete_subscription(subscribe_id): + missing.append(str(subscribe_id)) + continue deleted.append(subscribe.name) - self._repository.delete(subscribe_id) - self._report_deleted( - { - "media_source": subscribe.media_source, - "media_id": subscribe.media_id, - "season": subscribe.season, - } - ) if not deleted and missing: return False, f"未找到订阅：{', '.join(missing)}" diff --git a/app/application/subscription/delete.py b/app/application/subscription/delete.py index fe18c8e8e..5a2177258 100644 --- a/app/application/subscription/delete.py +++ b/app/application/subscription/delete.py @@ -1,13 +1,18 @@ """订阅删除应用用例及其依赖端口。""" -from contextlib import AbstractAsyncContextManager +from contextlib import AbstractAsyncContextManager, AbstractContextManager from dataclasses import dataclass from datetime import datetime, timezone import inspect from typing import Any, Awaitable, Callable, Mapping, Protocol, cast from uuid import uuid4 -from app.application.outbox import AsyncOutboxTransaction, OutboxIntent +from app.application.outbox import ( + AsyncOutboxTransaction, + OutboxIntent, + SyncOutboxTransaction, + SyncUnitOfWork, +) from app.schemas.event import SubscribeDeletedEventData @@ -43,6 +48,20 @@ class SubscribeDeletionRepository(Protocol): ... +class SyncSubscribeDeletionRepository(Protocol): + """同步消息入口执行订阅删除所需的最小数据访问端口。""" + + def get_candidate_sync( + self, + subscribe_id: int, + ) -> SubscribeDeletionCandidate | None: + """读取订阅及删除事件所需的稳定快照。""" + ... + + def stage_delete_sync(self, subscribe_id: int) -> None: + """把已读取的订阅登记为待删除，但不自行提交事务。""" + ... + class AsyncUnitOfWork(Protocol): """订阅写用例使用的异步事务端口。""" @@ -57,6 +76,18 @@ class AsyncUnitOfWork(Protocol): SubscribeDeletedPublisher = Callable[[dict[str, Any]], Awaitable[None]] SubscribeDeletedReporter = Callable[[Mapping[str, object]], object | Awaitable[object]] +SyncSubscribeDeletedPublisher = Callable[[dict[str, Any]], None] +SyncSubscribeDeletedReporter = Callable[[Mapping[str, object]], object] + + +@dataclass(frozen=True) +class _SubscribeDeletionEffects: + """同步和异步入口共用的删除事件、统计与 outbox 意图。""" + + event_payload: dict[str, Any] + report_payload: dict[str, object] + event_intent: OutboxIntent + report_intent: OutboxIntent class DeleteSubscribeCommand: @@ -89,36 +120,23 @@ class DeleteSubscribeCommand: 提交后的事件与上报保持原有顺序，任一副作用失败都会继续向调用方抛出。 """ candidate = await self._repository.get_candidate(subscribe_id) - if not self._can_delete(candidate, actor): + if not can_delete_subscribe(candidate, actor): return False assert candidate is not None - await self._repository.stage_delete(subscribe_id) - event_payload = build_subscribe_deleted_payload( + effects = _build_deletion_effects( subscribe_id, candidate.event_payload, ) - event_key = event_payload["idempotency_key"] - report_key = f"{event_key}:report" try: + await self._repository.stage_delete(subscribe_id) if self._outbox: await self._outbox.stage( - OutboxIntent( - event_key=event_key, - topic="subscribe.deleted", - payload=event_payload, - ), + effects.event_intent, datetime.now(timezone.utc), ) await self._outbox.stage( - OutboxIntent( - event_key=report_key, - topic="subscribe.deleted.report", - payload={ - "idempotency_key": report_key, - "subscribe_info": dict(candidate.event_payload), - }, - ), + effects.report_intent, datetime.now(timezone.utc), ) await self._unit_of_work.commit() @@ -126,37 +144,125 @@ class DeleteSubscribeCommand: await self._unit_of_work.rollback() raise - await self._publish_deleted(event_payload) + await self._publish_deleted(effects.event_payload) if self._outbox: await self._outbox.complete_by_event_key( - event_key, + effects.event_intent.event_key, datetime.now(timezone.utc), ) # 上报适配器会自行白名单过滤公开字段；传完整删除前快照可保留音乐实体维度， # 避免 Agent 与 API 入口收敛后丢失 music_type / total_tracks。 - report_result = self._report_deleted(dict(candidate.event_payload)) + report_result = self._report_deleted(effects.report_payload) if inspect.isawaitable(report_result): report_result = await report_result if report_result is False: raise RuntimeError("订阅删除统计上报未确认") if self._outbox: await self._outbox.complete_by_event_key( - report_key, + effects.report_intent.event_key, datetime.now(timezone.utc), ) return True - @staticmethod - def _can_delete( - candidate: SubscribeDeletionCandidate | None, + +class SyncDeleteSubscribeCommand: + """为同步消息入口执行同一订阅删除事务与 durable 副作用协议。""" + + def __init__( + self, + repository: SyncSubscribeDeletionRepository, + unit_of_work: SyncUnitOfWork, + publish_deleted: SyncSubscribeDeletedPublisher, + report_deleted: SyncSubscribeDeletedReporter, + outbox: SyncOutboxTransaction | None = None, + ) -> None: + """注入同步数据访问、事务与提交后副作用端口。""" + self._repository = repository + self._unit_of_work = unit_of_work + self._publish_deleted = publish_deleted + self._report_deleted = report_deleted + self._outbox = outbox + + def execute( + self, + subscribe_id: int, actor: SubscribeDeletionActor, ) -> bool: - """判断用户是否拥有目标订阅的删除权限。""" - if candidate is None: + """同步删除当前用户可访问的订阅，并保持事件和统计的可靠投递顺序。""" + candidate = self._repository.get_candidate_sync(subscribe_id) + if not can_delete_subscribe(candidate, actor): return False - if actor.is_superuser: - return True - return bool(candidate.username) and candidate.username == actor.username + assert candidate is not None + + effects = _build_deletion_effects( + subscribe_id, + candidate.event_payload, + ) + try: + self._repository.stage_delete_sync(subscribe_id) + if self._outbox: + now = datetime.now(timezone.utc) + self._outbox.stage(effects.event_intent, now) + self._outbox.stage(effects.report_intent, now) + self._unit_of_work.commit() + except Exception: + self._unit_of_work.rollback() + raise + + self._publish_deleted(effects.event_payload) + if self._outbox: + self._outbox.complete_by_event_key( + effects.event_intent.event_key, + datetime.now(timezone.utc), + ) + if self._report_deleted(effects.report_payload) is False: + raise RuntimeError("订阅删除统计上报未确认") + if self._outbox: + self._outbox.complete_by_event_key( + effects.report_intent.event_key, + datetime.now(timezone.utc), + ) + return True + + +def can_delete_subscribe( + candidate: SubscribeDeletionCandidate | None, + actor: SubscribeDeletionActor, +) -> bool: + """判断用户是否拥有目标订阅的删除权限。""" + if candidate is None: + return False + if actor.is_superuser: + return True + return bool(candidate.username) and candidate.username == actor.username + + +def _build_deletion_effects( + subscribe_id: int, + subscribe_info: Mapping[str, object], +) -> _SubscribeDeletionEffects: + """一次性构造两种执行风格共用的事件、上报和 durable intent。""" + event_payload = build_subscribe_deleted_payload(subscribe_id, subscribe_info) + event_key = event_payload["idempotency_key"] + report_key = f"{event_key}:report" + report_payload = dict(subscribe_info) + return _SubscribeDeletionEffects( + event_payload=event_payload, + report_payload=report_payload, + event_intent=OutboxIntent( + event_key=event_key, + topic="subscribe.deleted", + payload=event_payload, + ), + report_intent=OutboxIntent( + event_key=report_key, + topic="subscribe.deleted.report", + payload={ + "idempotency_key": report_key, + "subscribe_info": report_payload, + }, + ), + ) def build_subscribe_deleted_payload( @@ -176,7 +282,9 @@ def build_subscribe_deleted_payload( DeleteSubscribeScope = Callable[[], AbstractAsyncContextManager[DeleteSubscribeCommand]] +SyncDeleteSubscribeScope = Callable[[], AbstractContextManager[SyncDeleteSubscribeCommand]] _configured_delete_scope: DeleteSubscribeScope | None = None +_configured_sync_delete_scope: SyncDeleteSubscribeScope | None = None def configure_delete_subscribe_scope(provider: DeleteSubscribeScope) -> None: @@ -190,3 +298,16 @@ def get_delete_subscribe_scope() -> AbstractAsyncContextManager[DeleteSubscribeC if _configured_delete_scope is None: raise RuntimeError("订阅删除事务作用域尚未配置") return _configured_delete_scope() + + +def configure_sync_delete_subscribe_scope(provider: SyncDeleteSubscribeScope) -> None: + """由启动组合根登记同步消息入口使用的订阅删除事务作用域。""" + global _configured_sync_delete_scope + _configured_sync_delete_scope = provider + + +def get_sync_delete_subscribe_scope() -> AbstractContextManager[SyncDeleteSubscribeCommand]: + """返回一次独占同步会话的订阅删除命令作用域。""" + if _configured_sync_delete_scope is None: + raise RuntimeError("同步订阅删除事务作用域尚未配置") + return _configured_sync_delete_scope() diff --git a/app/chain/subscribe.py b/app/chain/subscribe.py index 887f9bb0e..f8fc352fa 100644 --- a/app/chain/subscribe.py +++ b/app/chain/subscribe.py @@ -48,6 +48,10 @@ from app.application.messaging.message import MessageTemplateHelper from app.application.mediaserver import MediaServerHelper from app.application.subscription.write import add_subscribe, async_add_subscribe from app.application.subscription.complete import get_subscription_completion_scope +from app.application.subscription.delete import ( + SubscribeDeletionActor, + get_sync_delete_subscribe_scope, +) from app.application.subscription.contract import ( build_subscribe_meta as _build_subscribe_meta, subscribe_media_key, @@ -3171,9 +3175,18 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): messenger=self, actions=self, repository=SubscribeOper(), - report_deleted=MoviePilotServerHelper.sub_done_async, + delete_subscription=self._delete_subscription, ) + @staticmethod + def _delete_subscription(subscribe_id: int) -> bool: + """通过统一同步命令删除订阅，保留消息入口原有的全局管理权限。""" + with get_sync_delete_subscribe_scope() as command: + return command.execute( + subscribe_id, + SubscribeDeletionActor(username="", is_superuser=True), + ) + def remote_delete(self, arg_str: str, channel: NotificationChannel, userid: Union[str, int] = None, source: Optional[str] = None): """ @@ -3189,28 +3202,18 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): save_history=False)) return arg_strs = str(arg_str).split() - subscribeoper = SubscribeOper() for arg_str in arg_strs: arg_str = arg_str.strip() if not arg_str.isdigit(): continue subscribe_id = int(arg_str) - subscribe = subscribeoper.get(subscribe_id) - if not subscribe: + if not self._delete_subscription(subscribe_id): self.post_message(_SchemaMessage( channel=channel, source=source, title=f"订阅编号 {subscribe_id} 不存在！", userid=userid, save_history=False)) return - # 删除订阅 - subscribeoper.delete(subscribe_id) - # 统计订阅 - MoviePilotServerHelper.sub_done_async({ - "media_source": subscribe.media_source, - "media_id": subscribe.media_id, - "season": subscribe.season, - }) # 重新发送消息 self.remote_list(channel=channel, userid=userid, source=source) diff --git a/app/db/oper/subscribe.py b/app/db/oper/subscribe.py index 209858171..b1e18717a 100644 --- a/app/db/oper/subscribe.py +++ b/app/db/oper/subscribe.py @@ -319,22 +319,36 @@ class SubscribeOper(DbOper): ) async def get_candidate( - self, - subscribe_id: int, + self, + subscribe_id: int, ) -> Optional[SubscribeDeletionCandidate]: """读取订阅删除用例需要的权限字段与完整事件快照。""" subscribe = await self.async_get(subscribe_id) + return self._deletion_candidate(subscribe_id, subscribe) + + def get_candidate_sync( + self, + subscribe_id: int, + ) -> Optional[SubscribeDeletionCandidate]: + """同步读取订阅删除用例需要的权限字段与完整事件快照。""" + return self._deletion_candidate(subscribe_id, self.get(subscribe_id)) + + @staticmethod + def _deletion_candidate( + subscribe_id: int, + subscribe: Optional[Subscribe], + ) -> Optional[SubscribeDeletionCandidate]: + """把 ORM 行投影为同步和异步删除命令共用的稳定快照。""" if not subscribe: return None values = subscribe.__dict__ - event_payload = { - column.name: values.get(column.name) - for column in subscribe.__table__.columns - } return SubscribeDeletionCandidate( subscribe_id=subscribe_id, username=subscribe.username, - event_payload=event_payload, + event_payload={ + column.name: values.get(column.name) + for column in subscribe.__table__.columns + }, ) async def list_candidates_by_identity( @@ -484,6 +498,12 @@ class SubscribeOper(DbOper): sqlalchemy_delete(Subscribe).where(Subscribe.id == sid) ) + def stage_delete_sync(self, sid: int) -> None: + """同步登记订阅删除但不提交，由 Application UnitOfWork 控制事务边界。""" + if not isinstance(self._db, Session): + raise RuntimeError("同步订阅删除需要调用方提供 Session") + self._db.execute(sqlalchemy_delete(Subscribe).where(Subscribe.id == sid)) + async def async_update(self, sid: int, payload: dict) -> Optional[Subscribe]: """ 异步更新订阅。 diff --git a/app/startup/composition/subscription.py b/app/startup/composition/subscription.py index 3c15e8cd0..47cf44db7 100644 --- a/app/startup/composition/subscription.py +++ b/app/startup/composition/subscription.py @@ -1,5 +1,6 @@ """订阅事务作用域及提交后回调的组合装配。""" +from collections.abc import AsyncIterator, Iterator from contextlib import asynccontextmanager, contextmanager from typing import Any @@ -10,7 +11,9 @@ from app.application.subscription.complete import ( ) from app.application.subscription.delete import ( DeleteSubscribeCommand, + SyncDeleteSubscribeCommand, configure_delete_subscribe_scope, + configure_sync_delete_subscribe_scope, ) from app.application.subscription.mutation import ( SubscriptionMutationService, @@ -38,13 +41,18 @@ async def _publish_deleted(payload: dict[str, Any]) -> None: await EventManager().async_send_event(EventType.SubscribeDeleted, payload) +def _publish_deleted_sync(payload: dict[str, Any]) -> None: + """为同步消息入口发布事务已提交的订阅删除事件。""" + EventManager().send_event(EventType.SubscribeDeleted, payload) + + def _publish_completed(payload: dict[str, Any]) -> None: """发布已提交的订阅完成事件。""" EventManager().send_event(EventType.SubscribeComplete, payload) @contextmanager -def subscription_completion_scope(): +def subscription_completion_scope() -> Iterator[CompleteSubscriptionCommand]: """为同步完成链创建独占 Session、UoW 与 durable outbox。""" session = SessionFactory() try: @@ -59,7 +67,7 @@ def subscription_completion_scope(): @asynccontextmanager -async def subscription_mutation_scope(): +async def subscription_mutation_scope() -> AsyncIterator[SubscriptionMutationService]: """为非 HTTP 入口创建独占订阅修改会话、UoW 与 outbox。""" async with async_session_scope() as session: yield SubscriptionMutationService( @@ -72,7 +80,7 @@ async def subscription_mutation_scope(): @asynccontextmanager -async def delete_subscribe_scope(): +async def delete_subscribe_scope() -> AsyncIterator[DeleteSubscribeCommand]: """为非 HTTP 入口创建独占订阅删除会话、UoW 与 outbox。""" async with async_session_scope() as session: yield DeleteSubscribeCommand( @@ -84,8 +92,25 @@ async def delete_subscribe_scope(): ) +@contextmanager +def sync_delete_subscribe_scope() -> Iterator[SyncDeleteSubscribeCommand]: + """为同步消息入口创建独占 Session、UoW 与 durable outbox。""" + session = SessionFactory() + try: + yield SyncDeleteSubscribeCommand( + repository=SubscribeOper(session), + unit_of_work=SqlAlchemyUnitOfWork(session), + publish_deleted=_publish_deleted_sync, + report_deleted=MoviePilotServerHelper.sub_done_durable, + outbox=SqlAlchemyOutboxRepository(session), + ) + finally: + session.close() + + def configure_transactional_subscription_scopes() -> None: """登记 Agent 等非 HTTP 入口复用的订阅事务作用域。""" configure_subscription_mutation_scope(subscription_mutation_scope) configure_delete_subscribe_scope(delete_subscribe_scope) + configure_sync_delete_subscribe_scope(sync_delete_subscribe_scope) configure_subscription_completion_scope(subscription_completion_scope) diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 1663ff2a5..eaead1554 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 边界。 +> 实施进度：阶段 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，宿主不再调用裸线程统计入口。 ## 当前复核结论（2026-08-24） @@ -15,7 +15,7 @@ ### 长期整改阶段 0：治理门禁恢复（2026-08-23） -- 宿主依赖基线已审查 TaskRegistry、有界后台 owner 与插件变更准入接入后的语义差异：当前为 `805` 个模块、`6502` 条内部导入边，12 组重点禁止边继续全部为 `0`，唯一非平凡 SCC 仍是隔离的 TMDB 移植包。 +- 宿主依赖基线已审查 TaskRegistry、有界后台 owner 与插件变更准入接入后的语义差异：当前为 `805` 个模块、`6503` 条内部导入边，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__` 等实现期导入不再被误冻结，既有数据库备份门面已补精确导出清单。 @@ -80,13 +80,25 @@ - 兼容边界不变：未知第三方自定义方法继续走开放 legacy fallback；旧插件签名和结果不匹配仍只诊断、 不拒绝加载。已登记方法的名称、kwargs、插件优先级、同步/异步入口和异常隔离 ABI 均保留。 +### 长期整改阶段 3：订阅删除生产路径统一（2026-08-24） + +- `/subscribes` 交互删除与 `/subscribe_delete` 远程命令不再直接调用 `SubscribeOper.delete` 后启动 + `sub_done_async` 裸线程；两个同步入口统一委托 `app.application.subscription.delete` 的权限、事务、 + 事件、统计与 outbox 协议，和 API、Agent 的异步删除路径共享候选快照、授权规则及 durable intent。 +- 启动组合根为同步消息入口提供独占 Session、UoW 和 outbox；订阅行与 `subscribe.deleted`、 + `subscribe.deleted.report` 在同一事务提交，暂存或提交失败都会回滚，成功后仍按事件、统计顺序收口。 +- 消息交互层只保留 ID 解析、名称展示和结果提示，不再拥有写库或统计实现；事务内目标已消失时按既有 + “未找到”语义返回，避免读取与删除竞态被误报为成功。 +- 插件兼容边界保持不变：`MoviePilotServerHelper.sub_reg_async` 与 `sub_done_async` 的类方法、签名和返回值 + 继续保留给旧插件；仅宿主生产调用清零，因此没有改动插件仓、SDK/Compat 映射或事件 payload。 + ### 总体判断 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**： - 继续采用单进程控制面是正确选择，不建议现在拆成微服务；插件、调度器、工作流、事件和数据库共享进程内状态，拆分会放大部署、事务和兼容成本。 - `foundation/domain/runtime/adapters/application/chain/api/startup` 的职责方向基本成立；宿主架构基线、复杂度 ratchet、异步阻塞 ratchet 当前均通过。 -- 依赖图当前为 `805` 个 Python 模块、`6502` 条内部导入边；唯一非平凡 SCC 位于隔离的 TMDB 第三方移植包内部，不应为了指标归零重写。 +- 依赖图当前为 `805` 个 Python 模块、`6503` 条内部导入边；唯一非平凡 SCC 位于隔离的 TMDB 第三方移植包内部，不应为了指标归零重写。 - 当前主要风险已经从“目录和依赖失控”转移到运行时协议、后台副作用的可靠性和遗留兼容面。换言之，下一阶段重点应是**语义收口和可验证性**，而不是继续搬文件或机械拆大文件。 综合评价：架构方向可持续，生产可用性较高；可演进性仍处于中等水平。现阶段没有静态审计发现必须立即推倒重来的 P0 架构问题，但存在需要按 P1/P2 计划治理的真实债务。 @@ -104,6 +116,8 @@ `shield` 不再让网络请求逃逸生命周期预算，仓库级并发合并、缓存键和 V1/V2/V3 返回兼容保持不变。 请求作用域的结构化并发不进入全局登记器：传统 WebAgent SSE 的 collection 子任务改由生成器 `finally` 取消并等待清理，断线和 ASGI 取消均不会留下请求级 task。 + 订阅删除的宿主生产者也已完成 durable 分级：消息交互和远程删除不再调用裸线程统计入口，而是 + 与业务删除原子暂存事件和统计 intent；旧类方法只作为插件 ABI 保留，不纳入宿主可靠性证明。 2. **Model/Base 的数据库装饰器和隐式会话 ABI 已全部清零。** 查询、写事务和 `legacy_*` 装饰器均为 `0`；所有 Model `db` 参数要求显式 Session，Base CRUD 仅在调用方事务内查询或 stage。可无会话构造的入口统一留在 Oper，经组合根事务执行器运行；插件 SDK 不再导出宿主 Model。后续重点转为减少 ORM 对象跨层流转，并保持 Model 隐式事务零回退。 Oper 内部的执行入口也已统一：最后一处 `AgentTaskOper` 直接 transaction runner 调用已迁入 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 6bd635b87..5f06e98dd 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": 6502, - "edge_sha256": "fe24e90415afe0789875f5e6d936e9b7d3536b3bfd0383d585ea35cc4be76386", + "edge_count": 6503, + "edge_sha256": "57c326a81dbba07909871a97df1c73416850205c3d385a15afe3a07e3e44c00c", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -3324,6 +3324,7 @@ "app.chain.subscribe -> app.application.subscription", "app.chain.subscribe -> app.application.subscription.complete", "app.chain.subscribe -> app.application.subscription.contract", + "app.chain.subscribe -> app.application.subscription.delete", "app.chain.subscribe -> app.application.subscription.query", "app.chain.subscribe -> app.application.subscription.write", "app.chain.subscribe -> app.application.torrent", diff --git a/tests/fixtures/architecture/runtime-contract-baseline.json b/tests/fixtures/architecture/runtime-contract-baseline.json index 66af0168a..b7e04c4f3 100644 --- a/tests/fixtures/architecture/runtime-contract-baseline.json +++ b/tests/fixtures/architecture/runtime-contract-baseline.json @@ -2221,7 +2221,7 @@ }, { "caller": "app.startup.composition.subscription", - "count": 1 + "count": 2 }, { "caller": "app.startup.initializers.modules", @@ -2344,7 +2344,7 @@ ] } }, - "producer_count": 77 + "producer_count": 78 }, "module_method_specs": { "anilist_credits": { diff --git a/tests/test_subscribe_delete_command.py b/tests/test_subscribe_delete_command.py index 1811d632d..6b08eec97 100644 --- a/tests/test_subscribe_delete_command.py +++ b/tests/test_subscribe_delete_command.py @@ -1,11 +1,13 @@ """订阅删除应用用例的事务、权限与副作用时序测试。""" -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, MagicMock import pytest +from sqlalchemy.orm import Session from app.application.subscription.delete import ( DeleteSubscribeCommand, + SyncDeleteSubscribeCommand, SubscribeDeletionActor, SubscribeDeletionCandidate, ) @@ -16,10 +18,11 @@ from app.db.oper.subscribe import SubscribeOper class _Repository: """记录订阅删除用例数据访问顺序的仓储替身。""" - def __init__(self, candidate, calls): - """保存候选订阅和共享调用序列。""" + def __init__(self, candidate, calls, delete_error=None): + """保存候选订阅、共享调用序列和可选删除异常。""" self.candidate = candidate self.calls = calls + self.delete_error = delete_error async def get_candidate(self, subscribe_id): """返回预设候选订阅。""" @@ -29,6 +32,8 @@ class _Repository: async def stage_delete(self, subscribe_id): """记录待删除的订阅编号。""" self.calls.append(("delete", subscribe_id)) + if self.delete_error: + raise self.delete_error class _UnitOfWork: @@ -69,6 +74,62 @@ class _Outbox: self.calls.append(("outbox_complete", event_key)) +class _SyncRepository: + """记录同步订阅删除的数据访问顺序。""" + + def __init__(self, candidate, calls, delete_error=None): + """保存候选订阅、共享调用序列和可选删除异常。""" + self.candidate = candidate + self.calls = calls + self.delete_error = delete_error + + def get_candidate_sync(self, subscribe_id): + """返回预设候选订阅。""" + self.calls.append(("get", subscribe_id)) + return self.candidate + + def stage_delete_sync(self, subscribe_id): + """记录同步待删除编号并按需失败。""" + self.calls.append(("delete", subscribe_id)) + if self.delete_error: + raise self.delete_error + + +class _SyncUnitOfWork: + """记录同步删除命令的提交与回滚。""" + + def __init__(self, calls, commit_error=None): + """保存共享调用序列与可选提交异常。""" + self.calls = calls + self.commit_error = commit_error + + def commit(self): + """记录提交并按需抛出异常。""" + self.calls.append(("commit",)) + if self.commit_error: + raise self.commit_error + + def rollback(self): + """记录回滚。""" + self.calls.append(("rollback",)) + + +class _SyncOutbox: + """记录同步删除 intent 的暂存与完成顺序。""" + + def __init__(self, calls): + """保存共享调用序列。""" + self.calls = calls + + def stage(self, intent, _now): + """记录同步暂存的 intent。""" + self.calls.append(("outbox_stage", intent)) + + def complete_by_event_key(self, event_key, _completed_at): + """记录同步完成的 intent。""" + self.calls.append(("outbox_complete", event_key)) + + def _candidate(username="alice"): """构造带完整事件身份字段的订阅删除候选。""" return SubscribeDeletionCandidate( @@ -92,6 +153,7 @@ def _command( event_error=None, report_error=None, outbox=None, + delete_error=None, ): """构造可观察事件与上报失败的订阅删除用例。""" async def publish(payload): @@ -108,7 +170,7 @@ def _command( return True return DeleteSubscribeCommand( - repository=_Repository(candidate, calls), + repository=_Repository(candidate, calls, delete_error), unit_of_work=_UnitOfWork(calls, commit_error), publish_deleted=publish, report_deleted=report, @@ -138,6 +200,26 @@ def _async_report_command(candidate, calls, result=True, error=None, outbox=None ) +def _sync_command(candidate, calls, commit_error=None, delete_error=None, outbox=None): + """构造可观察事务和副作用顺序的同步订阅删除命令。""" + def publish(payload): + """记录同步删除事件。""" + calls.append(("event", payload["subscribe_id"], payload)) + + def report(payload): + """记录同步删除统计。""" + calls.append(("report", payload)) + return True + + return SyncDeleteSubscribeCommand( + repository=_SyncRepository(candidate, calls, delete_error), + unit_of_work=_SyncUnitOfWork(calls, commit_error), + publish_deleted=publish, + report_deleted=report, + outbox=outbox, + ) + + @pytest.mark.asyncio async def test_owner_delete_commits_before_event_and_report(): """owner 删除成功时必须先提交，再按原顺序发送事件和上报。""" @@ -202,6 +284,26 @@ async def test_commit_failure_rolls_back_without_event_or_report(): assert [call[0] for call in calls] == ["get", "delete", "commit", "rollback"] +@pytest.mark.asyncio +async def test_delete_stage_failure_rolls_back_without_effects(): + """异步暂存失败必须显式回滚，且不得写 intent 或发送成功副作用。""" + calls = [] + command = _command( + _candidate(), + calls, + delete_error=RuntimeError("delete failed"), + outbox=_Outbox(calls), + ) + + with pytest.raises(RuntimeError, match="delete failed"): + await command.execute( + 7, + SubscribeDeletionActor(username="alice", is_superuser=False), + ) + + assert [call[0] for call in calls] == ["get", "delete", "rollback"] + + @pytest.mark.asyncio async def test_event_failure_happens_after_commit_and_stops_report(): """事件失败保持原有传播语义，但事务必须已经提交且不得继续上报。""" @@ -385,3 +487,65 @@ async def test_repository_stage_delete_does_not_commit(): session.execute.assert_awaited_once() session.commit.assert_not_awaited() + + +def test_sync_delete_uses_same_durable_effect_order(): + """同步消息入口必须复用异步删除命令的事务、事件和统计顺序。""" + calls = [] + command = _sync_command(_candidate(), calls, outbox=_SyncOutbox(calls)) + + assert command.execute( + 7, + SubscribeDeletionActor(username="", is_superuser=True), + ) is True + + assert [call[0] for call in calls] == [ + "get", "delete", "outbox_stage", "outbox_stage", "commit", + "event", "outbox_complete", "report", "outbox_complete", + ] + assert calls[2][1].topic == "subscribe.deleted" + assert calls[3][1].topic == "subscribe.deleted.report" + assert calls[7][1] == _candidate().event_payload + + +@pytest.mark.parametrize("failure", ["delete", "commit"]) +def test_sync_delete_rolls_back_transaction_failures(failure): + """同步暂存或提交失败时必须回滚，且不得发送删除成功副作用。""" + calls = [] + error = RuntimeError(f"{failure} failed") + command = _sync_command( + _candidate(), + calls, + delete_error=error if failure == "delete" else None, + commit_error=error if failure == "commit" else None, + ) + + with pytest.raises(RuntimeError, match=f"{failure} failed"): + command.execute( + 7, + SubscribeDeletionActor(username="", is_superuser=True), + ) + + assert calls[-1] == ("rollback",) + assert all(call[0] not in {"event", "report"} for call in calls) + + +def test_sync_repository_candidate_and_delete_share_caller_session(monkeypatch): + """同步仓储投影和删除都使用组合根传入的同一个 Session 且不提交。""" + subscribe = Subscribe( + id=7, + username="alice", + name="测试订阅", + media_source="tmdb", + media_id="123", + season=2, + ) + session = MagicMock(spec=Session) + oper = SubscribeOper(session) + monkeypatch.setattr(oper, "get", lambda subscribe_id: subscribe) + + candidate = oper.get_candidate_sync(7) + oper.stage_delete_sync(7) + + assert candidate is not None + assert candidate.event_payload["media_id"] == "123" diff --git a/tests/test_subscribe_interaction_delete.py b/tests/test_subscribe_interaction_delete.py new file mode 100644 index 000000000..ce69097ab --- /dev/null +++ b/tests/test_subscribe_interaction_delete.py @@ -0,0 +1,52 @@ +"""订阅消息交互删除动作的应用边界测试。""" + +from types import SimpleNamespace + +from app.application.messaging.subscribe import SubscribeInteractionHandler + + +class _Repository: + """只提供消息展示所需读取能力的订阅仓储替身。""" + + def __init__(self, subscribes): + """按订阅 ID 保存测试快照。""" + self._subscribes = subscribes + + def get(self, subscribe_id): + """返回指定订阅快照。""" + return self._subscribes.get(subscribe_id) + + +def test_interaction_delete_delegates_each_existing_id_to_application_action(): + """消息层只负责解析和展示，删除必须委托统一 Application 动作。""" + deleted_ids = [] + handler = SubscribeInteractionHandler( + messenger=SimpleNamespace(), + actions=SimpleNamespace(), + repository=_Repository({ + 7: SimpleNamespace(name="电影订阅"), + 8: SimpleNamespace(name="剧集订阅"), + }), + delete_subscription=lambda subscribe_id: deleted_ids.append(subscribe_id) or True, + ) + + success, message = handler._delete_subscribes("7 8 9") + + assert success is True + assert deleted_ids == [7, 8] + assert message == "已删除 2 个订阅：电影订阅, 剧集订阅；未找到：9" + + +def test_interaction_delete_reports_transaction_race_as_missing(): + """读取后事务内目标消失时不得误报已删除。""" + handler = SubscribeInteractionHandler( + messenger=SimpleNamespace(), + actions=SimpleNamespace(), + repository=_Repository({7: SimpleNamespace(name="电影订阅")}), + delete_subscription=lambda _subscribe_id: False, + ) + + success, message = handler._delete_subscribes("7") + + assert success is False + assert message == "未找到订阅：7"