diff --git a/app/api/dependencies/subscription.py b/app/api/dependencies/subscription.py index 9faa5d308..27748d861 100644 --- a/app/api/dependencies/subscription.py +++ b/app/api/dependencies/subscription.py @@ -79,9 +79,7 @@ def get_delete_subscribe_command( repository_port: SubscriptionStagingPort = Depends(get_subscription_repository), unit_of_work: object = Depends(get_subscription_transaction), outbox: AsyncOutboxStager = Depends(get_subscription_outbox), - dispatch_store: AsyncOutboxDispatchStore = Depends( - get_subscription_outbox_store - ), + dispatch_store: AsyncOutboxDispatchStore = Depends(get_subscription_outbox_store), ) -> DeleteSubscribeCommand: """组装请求级订阅删除用例及其具体适配器。""" return DeleteSubscribeCommand( @@ -109,9 +107,7 @@ def get_delete_subscriptions_by_identity_command( repository_port: SubscriptionStagingPort = Depends(get_subscription_repository), unit_of_work: object = Depends(get_subscription_transaction), outbox: AsyncOutboxStager = Depends(get_subscription_outbox), - dispatch_store: AsyncOutboxDispatchStore = Depends( - get_subscription_outbox_store - ), + dispatch_store: AsyncOutboxDispatchStore = Depends(get_subscription_outbox_store), ) -> DeleteSubscriptionsByIdentityCommand: """组装请求级按媒体身份删除订阅用例。""" return DeleteSubscriptionsByIdentityCommand( @@ -151,6 +147,7 @@ def get_search_subscriptions_command( runtime: HostRuntime = Depends(get_host_runtime), ) -> SearchSubscriptionsCommand: """组装手工订阅搜索用例，并把调度延迟到响应后的后台任务。""" + def schedule_search( subscribe_ids: tuple[int, ...] | None, state: str | None, @@ -184,14 +181,28 @@ def get_subscription_query_service( def get_subscription_execution_status_service( db: AsyncSession = Depends(get_async_session), runtime: HostRuntime = Depends(get_host_runtime), + task_registry: TaskRegistry = Depends(get_background_task_registry), ) -> SubscriptionExecutionStatusService: - """组装请求级订阅执行状态投影服务。""" + """组装请求级订阅执行状态投影与取消服务。""" factory = runtime.subscription.execution_status_repository if factory is None: raise RuntimeError("订阅执行状态仓储未注册") repository = factory(db) + search_repository = get_subscription_search_repository(runtime) + registry = resolve_background_task_registry(task_registry) + + async def request_cancel(batch_id: str) -> bool: + """在线程 owner 内提交同步队列写入，并等待其真实终态。""" + task = registry.create_sync( + search_repository.request_cancel, + batch_id, + owner="api.subscribe.execution.cancel", + ) + return bool(await task) + return SubscriptionExecutionStatusService( - repository=cast(SubscriptionExecutionReadRepository, repository) + repository=cast(SubscriptionExecutionReadRepository, repository), + request_cancel=request_cancel, ) @@ -206,14 +217,10 @@ def get_subscription_search_repository( def get_subscription_mutation_service( repository_port: SessionSubscriptionPort = Depends(get_subscription_repository), - history_repository: SubscriptionHistoryStagingPort = Depends( - get_subscription_history_repository - ), + history_repository: SubscriptionHistoryStagingPort = Depends(get_subscription_history_repository), unit_of_work: object = Depends(get_subscription_transaction), outbox: AsyncOutboxStager = Depends(get_subscription_outbox), - dispatch_store: AsyncOutboxDispatchStore = Depends( - get_subscription_outbox_store - ), + dispatch_store: AsyncOutboxDispatchStore = Depends(get_subscription_outbox_store), ) -> SubscriptionMutationService: """组装异步订阅写服务。""" return SubscriptionMutationService( @@ -230,9 +237,7 @@ def get_servarr_subscription_batch_writer( repository_port: SubscriptionStagingPort = Depends(get_subscription_repository), unit_of_work: object = Depends(get_subscription_transaction), outbox: AsyncOutboxStager = Depends(get_subscription_outbox), - dispatch_store: AsyncOutboxDispatchStore = Depends( - get_subscription_outbox_store - ), + dispatch_store: AsyncOutboxDispatchStore = Depends(get_subscription_outbox_store), runtime: HostRuntime = Depends(get_host_runtime), ) -> SubscriptionBatchWritePort: """组装 Servarr 多季订阅使用的请求级原子批量写端口。""" diff --git a/app/api/endpoints/subexecution.py b/app/api/endpoints/subexecution.py index 9aace7321..52cad52fa 100644 --- a/app/api/endpoints/subexecution.py +++ b/app/api/endpoints/subexecution.py @@ -8,14 +8,11 @@ from app.api.dependencies.auth import get_current_active_user_async from app.api.dependencies.subscription import ( get_subscription_execution_status_service, get_subscription_query_service, - get_subscription_search_repository, ) from app.api.principal import ApiPrincipal from app.api.response import ResponseAPIRouter -from app.application.subscription.execution import SubscriptionSearchRepository from app.application.subscription.query import SubscriptionQueryService from app.application.subscription.status import SubscriptionExecutionStatusService -from app.runtime.execution import run_in_threadpool from app.schemas.response import Response as _SchemaResponse from app.schemas.subscribe import SubscriptionBatchStatus as _SchemaSubscriptionBatchStatus @@ -40,9 +37,7 @@ async def _accessible_subscription_ids( ) async def list_subscription_execution_batches( limit: int = 10, - status_service: SubscriptionExecutionStatusService = Depends( - get_subscription_execution_status_service - ), + status_service: SubscriptionExecutionStatusService = Depends(get_subscription_execution_status_service), query: SubscriptionQueryService = Depends(get_subscription_query_service), current_user: ApiPrincipal = Depends(get_current_active_user_async), ) -> Any: @@ -62,9 +57,7 @@ async def list_subscription_execution_batches( ) async def get_subscription_execution_batch( batch_id: str, - status_service: SubscriptionExecutionStatusService = Depends( - get_subscription_execution_status_service - ), + status_service: SubscriptionExecutionStatusService = Depends(get_subscription_execution_status_service), query: SubscriptionQueryService = Depends(get_subscription_query_service), current_user: ApiPrincipal = Depends(get_current_active_user_async), ) -> Any: @@ -86,13 +79,8 @@ async def get_subscription_execution_batch( ) async def cancel_subscription_execution_batch( batch_id: str, - status_service: SubscriptionExecutionStatusService = Depends( - get_subscription_execution_status_service - ), + status_service: SubscriptionExecutionStatusService = Depends(get_subscription_execution_status_service), query: SubscriptionQueryService = Depends(get_subscription_query_service), - search_repository: SubscriptionSearchRepository = Depends( - get_subscription_search_repository - ), current_user: ApiPrincipal = Depends(get_current_active_user_async), ) -> Any: """在权限校验后请求取消尚未越过下载副作用边界的任务。""" @@ -103,7 +91,7 @@ async def cancel_subscription_execution_batch( ) if batch is None: return _SchemaResponse(success=False, message="订阅搜索批次不存在") - cancelled = await run_in_threadpool(search_repository.request_cancel, batch_id) + cancelled = await status_service.request_cancel(batch_id) return _SchemaResponse( success=bool(cancelled), message="" if cancelled else "订阅搜索批次已结束或无法取消", diff --git a/app/application/subscription/status.py b/app/application/subscription/status.py index f45ac2956..7d202f824 100644 --- a/app/application/subscription/status.py +++ b/app/application/subscription/status.py @@ -1,5 +1,6 @@ """订阅执行状态的业务投影与访问范围治理。""" +from collections.abc import Awaitable, Callable from dataclasses import dataclass from typing import Optional, Protocol @@ -108,9 +109,20 @@ class SubscriptionExecutionStatusService: "reconcile_required", } - def __init__(self, repository: SubscriptionExecutionReadRepository) -> None: - """保存请求会话绑定的状态读取端口。""" + def __init__( + self, + repository: SubscriptionExecutionReadRepository, + request_cancel: Optional[Callable[[str], Awaitable[bool]]] = None, + ) -> None: + """保存请求会话绑定的状态读取端口和可选取消用例。""" self._repository = repository + self._request_cancel = request_cancel + + async def request_cancel(self, batch_id: str) -> bool: + """通过组合根提供的执行边界请求取消搜索批次。""" + if self._request_cancel is None: + raise RuntimeError("订阅搜索批次取消能力未注册") + return bool(await self._request_cancel(batch_id)) async def for_subscriptions( self, @@ -245,8 +257,7 @@ class SubscriptionExecutionStatusService: created_at=batch.created_at, updated_at=batch.updated_at, error=cls._safe_error(batch.last_error), - can_cancel=batch.state in {"queued", "running", "cancelling"} - and not batch.cancel_requested, + can_cancel=batch.state in {"queued", "running", "cancelling"} and not batch.cancel_requested, ) @staticmethod @@ -257,9 +268,7 @@ class SubscriptionExecutionStatusService: """超级用户不限制；普通用户必须拥有批次内全部订阅。""" if accessible_subscription_ids is None: return True - return bool(tasks) and all( - task.subscription_id in accessible_subscription_ids for task in tasks - ) + return bool(tasks) and all(task.subscription_id in accessible_subscription_ids for task in tasks) @staticmethod def _safe_error(error: Optional[str]) -> Optional[str]: diff --git a/app/chain/search/provider.py b/app/chain/search/provider.py index a95bd98c1..7ba598c5d 100644 --- a/app/chain/search/provider.py +++ b/app/chain/search/provider.py @@ -38,7 +38,7 @@ def _site_request_interval(site: SiteIndexer) -> float: """读取站点管理中已有的单次访问间隔配置。""" try: return max(0.0, float(site.get("limit_seconds") or 0)) - except (TypeError, ValueError): + except TypeError, ValueError: return 0.0 @@ -47,12 +47,7 @@ def _reserve_site_request(site: SiteIndexer) -> float: interval = _site_request_interval(site) if interval <= 0: return 0.0 - site_key = str( - site.get("id") - or site.get("domain") - or site.get("url") - or site.get("name") - ) + site_key = str(site.get("id") or site.get("domain") or site.get("url") or site.get("name")) now = time.monotonic() with _site_request_schedule_lock: request_at = max(now, _site_next_request_at.get(site_key, now)) @@ -77,7 +72,7 @@ async def _async_wait_for_site_request(site: SiteIndexer) -> None: def _search_site_page( - owner: "SearchProviderOwner", + owner: "_SearchProviderSyncOwner", *, site: SiteIndexer, keyword: str, diff --git a/app/chain/subscribe/create.py b/app/chain/subscribe/create.py index 8b0f567c6..febb5390b 100644 --- a/app/chain/subscribe/create.py +++ b/app/chain/subscribe/create.py @@ -1,6 +1,6 @@ """订阅创建与批量写入编排""" -from collections.abc import Sequence +from collections.abc import Awaitable, Callable, Sequence from typing import Any, Dict, Optional, Tuple, cast from app.application.subscription.write import ( @@ -26,6 +26,32 @@ from app.schemas.types import ( ) +def _build_post_commit_callback( + owner: Any, + context: _SubscribePostCommitContext, +) -> Callable[[int], bool]: + """冻结同步订阅提交后的副作用上下文。""" + + def after_commit(subscribe_id: int) -> bool: + """把同步提交后的副作用委托给通知 owner。""" + return cast(bool, owner._SubscribeChain__post_subscribe_added(subscribe_id, context)) + + return after_commit + + +def _build_async_post_commit_callback( + owner: Any, + context: _SubscribePostCommitContext, +) -> Callable[[int], Awaitable[bool]]: + """冻结异步订阅提交后的副作用上下文。""" + + async def after_commit(subscribe_id: int) -> bool: + """把异步提交后的副作用委托给通知 owner。""" + return cast(bool, await owner._SubscribeChain__async_post_subscribe_added(subscribe_id, context)) + + return after_commit + + class SubscribeCreateOwner(_SubscribeOwnerBase): """订阅创建与批量写入编排，作为 SubscribeChain 的单一职责实现 owner。""" @@ -408,17 +434,12 @@ class SubscribeCreateOwner(_SubscribeOwnerBase): context, self._SubscribeChain__build_subscribe_notification(context), ) - - def _after_commit(subscribe_id: int) -> bool: - """把同步提交后的副作用委托给单一顺序实现。""" - return cast(bool, self._SubscribeChain__post_subscribe_added(subscribe_id, post_commit_context)) - sid, err_msg = add_subscribe( mediainfo=context.mediainfo, subscribe_oper=self.subscription_repository, season=context.season, username=context.username, - after_commit=_after_commit, + after_commit=_build_post_commit_callback(self, post_commit_context), notification=post_commit_context.notification, occurrence_id=post_commit_context.occurrence_id, **context.options, @@ -437,23 +458,12 @@ class SubscribeCreateOwner(_SubscribeOwnerBase): context, self._SubscribeChain__build_subscribe_notification(context), ) - - async def _after_commit(subscribe_id: int) -> bool: - """把异步提交后的副作用委托给单一顺序实现。""" - return cast( - bool, - await self._SubscribeChain__async_post_subscribe_added( - subscribe_id, - post_commit_context, - ), - ) - sid, err_msg = await async_add_subscribe( mediainfo=context.mediainfo, subscribe_oper=self.subscription_repository, season=context.season, username=context.username, - after_commit=_after_commit, + after_commit=_build_async_post_commit_callback(self, post_commit_context), notification=post_commit_context.notification, occurrence_id=post_commit_context.occurrence_id, **context.options, @@ -644,23 +654,10 @@ class SubscribeCreateOwner(_SubscribeOwnerBase): notification, ) - async def after_commit( - subscribe_id: int, - frozen: _SubscribePostCommitContext = post_commit_context, - ) -> bool: - """按各季准备阶段冻结的上下文复用单条新增副作用。""" - return cast( - bool, - await self._SubscribeChain__async_post_subscribe_added( - subscribe_id, - frozen, - ), - ) - request = build_subscription_create_request( context.mediainfo, notification=notification, - after_commit=after_commit, + after_commit=_build_async_post_commit_callback(self, post_commit_context), occurrence_id=post_commit_context.occurrence_id, season=context.season, username=context.username, diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 0db30930d..a9c6fd598 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -755,7 +755,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 936 | -| 内部导入边 | 7,835 | +| 内部导入边 | 7,832 | | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 53（债务已清零，53 条精确 containment） | diff --git a/docs/architecture/optimization-checklist.md b/docs/architecture/optimization-checklist.md index 267444baa..522235169 100644 --- a/docs/architecture/optimization-checklist.md +++ b/docs/architecture/optimization-checklist.md @@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 936 / 7,835 | `dependency-baseline.json` 当前快照 | +| 宿主 Python 模块 / 内部依赖边 | 936 / 7,832 | `dependency-baseline.json` 当前快照 | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 02c65b25e..d2d39d7ce 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1089,8 +1089,8 @@ "runtime_only": true } }, - "edge_count": 7835, - "edge_sha256": "491127de88d2727e18db050dba310fd4df5a0fe32c67f4976c9f6d1145e2e53f", + "edge_count": 7832, + "edge_sha256": "3922707ade2b154adcb26bea725f9070ab9ccca109dd4ad9aa0c932cbbad9b74", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -2641,11 +2641,8 @@ "app.api.endpoints.subexecution -> app.api.response", "app.api.endpoints.subexecution -> app.application", "app.api.endpoints.subexecution -> app.application.subscription", - "app.api.endpoints.subexecution -> app.application.subscription.execution", "app.api.endpoints.subexecution -> app.application.subscription.query", "app.api.endpoints.subexecution -> app.application.subscription.status", - "app.api.endpoints.subexecution -> app.runtime", - "app.api.endpoints.subexecution -> app.runtime.execution", "app.api.endpoints.subexecution -> app.schemas", "app.api.endpoints.subexecution -> app.schemas.response", "app.api.endpoints.subexecution -> app.schemas.subscribe", diff --git a/tests/fixtures/architecture/startup-performance-baseline.json b/tests/fixtures/architecture/startup-performance-baseline.json index d9cb92f8b..94153e728 100644 --- a/tests/fixtures/architecture/startup-performance-baseline.json +++ b/tests/fixtures/architecture/startup-performance-baseline.json @@ -1,41 +1,41 @@ { "schema_version": 2, - "generated_at": "2026-08-30T04:51:04.908046+00:00", + "generated_at": "2026-09-01T09:42:57.922376+00:00", "platform": "macOS-26.5.2-arm64-arm-64bit-Mach-O", "python": "3.14.3", "repeat": 3, "targets": { "app.startup.lifecycle": { - "loaded_app_module_count": 499, - "max_ms": 1138.629, - "median_ms": 1108.684, - "min_ms": 1097.855, + "loaded_app_module_count": 510, + "max_ms": 1115.407, + "median_ms": 1106.606, + "min_ms": 1098.166, "samples_ms": [ - 1108.684, - 1138.629, - 1097.855 + 1115.407, + 1098.166, + 1106.606 ] }, "app.factory": { - "loaded_app_module_count": 511, - "max_ms": 1118.665, - "median_ms": 1117.279, - "min_ms": 1111.113, + "loaded_app_module_count": 522, + "max_ms": 1140.393, + "median_ms": 1134.83, + "min_ms": 1119.004, "samples_ms": [ - 1111.113, - 1117.279, - 1118.665 + 1119.004, + 1140.393, + 1134.83 ] }, "app.main": { - "loaded_app_module_count": 513, - "max_ms": 1172.187, - "median_ms": 1161.652, - "min_ms": 1155.282, + "loaded_app_module_count": 524, + "max_ms": 1180.675, + "median_ms": 1165.915, + "min_ms": 1161.859, "samples_ms": [ - 1155.282, - 1161.652, - 1172.187 + 1180.675, + 1165.915, + 1161.859 ] } }, @@ -47,22 +47,22 @@ { "mode": "normal", "enabled_component_count": 32, - "startup_ms": 0.714, - "full_lifespan_ms": 1.828, + "startup_ms": 0.716, + "full_lifespan_ms": 1.819, "stage_ms": { "文件日志": 0.067, - "后台任务登记器": 0.036, - "数据库准备": 0.027, - "HTTP 基础能力": 0.027, - "站点访问端口": 0.023, - "Chain 外部端口": 0.022, - "Chain 网络端口": 0.021, + "后台任务登记器": 0.035, + "数据库准备": 0.03, + "HTTP 基础能力": 0.022, + "站点访问端口": 0.021, + "Chain 外部端口": 0.02, + "Chain 网络端口": 0.019, "领域依赖装配": 0.019, - "数据库引擎预热": 0.021, - "数据库连接预算": 0.021, - "路由": 0.022, - "插件运行时装配": 0.02, - "模块服务": 0.019, + "数据库引擎预热": 0.016, + "数据库连接预算": 0.019, + "路由": 0.021, + "插件运行时装配": 0.019, + "模块服务": 0.023, "插件服务装配": 0.02, "消息队列": 0.02, "插件备份恢复": 0.02, @@ -70,7 +70,7 @@ "定时器": 0.02, "监控器": 0.02, "待处理整理回放": 0.02, - "命令服务": 0.019, + "命令服务": 0.02, "工作流": 0.02, "插件同步与启动收尾": 0.016 }, @@ -85,32 +85,32 @@ { "mode": "normal", "enabled_component_count": 32, - "startup_ms": 0.797, - "full_lifespan_ms": 1.851, + "startup_ms": 0.729, + "full_lifespan_ms": 1.879, "stage_ms": { - "文件日志": 0.08, - "后台任务登记器": 0.047, - "数据库准备": 0.034, - "HTTP 基础能力": 0.025, - "站点访问端口": 0.025, - "Chain 外部端口": 0.019, + "文件日志": 0.07, + "后台任务登记器": 0.036, + "数据库准备": 0.028, + "HTTP 基础能力": 0.024, + "站点访问端口": 0.023, + "Chain 外部端口": 0.023, "Chain 网络端口": 0.02, - "领域依赖装配": 0.023, - "数据库引擎预热": 0.019, + "领域依赖装配": 0.022, + "数据库引擎预热": 0.02, "数据库连接预算": 0.02, - "路由": 0.023, - "插件运行时装配": 0.019, - "模块服务": 0.02, - "插件服务装配": 0.022, - "消息队列": 0.018, - "插件备份恢复": 0.02, - "插件": 0.021, - "定时器": 0.021, + "路由": 0.022, + "插件运行时装配": 0.021, + "模块服务": 0.021, + "插件服务装配": 0.019, + "消息队列": 0.023, + "插件备份恢复": 0.021, + "插件": 0.019, + "定时器": 0.02, "监控器": 0.021, "待处理整理回放": 0.021, - "命令服务": 0.021, - "工作流": 0.021, - "插件同步与启动收尾": 0.021 + "命令服务": 0.02, + "工作流": 0.02, + "插件同步与启动收尾": 0.02 }, "threads_before": 1, "threads_started": 1, @@ -123,32 +123,32 @@ { "mode": "normal", "enabled_component_count": 32, - "startup_ms": 0.834, - "full_lifespan_ms": 1.917, + "startup_ms": 0.758, + "full_lifespan_ms": 1.909, "stage_ms": { - "文件日志": 0.113, - "后台任务登记器": 0.041, - "数据库准备": 0.029, - "HTTP 基础能力": 0.026, + "文件日志": 0.068, + "后台任务登记器": 0.037, + "数据库准备": 0.028, + "HTTP 基础能力": 0.029, "站点访问端口": 0.025, - "Chain 外部端口": 0.022, - "Chain 网络端口": 0.021, + "Chain 外部端口": 0.023, + "Chain 网络端口": 0.023, "领域依赖装配": 0.022, - "数据库引擎预热": 0.02, + "数据库引擎预热": 0.022, "数据库连接预算": 0.021, - "路由": 0.022, - "插件运行时装配": 0.021, - "模块服务": 0.021, - "插件服务装配": 0.021, - "消息队列": 0.02, - "插件备份恢复": 0.02, - "插件": 0.019, - "定时器": 0.02, - "监控器": 0.017, - "待处理整理回放": 0.025, - "命令服务": 0.02, + "路由": 0.023, + "插件运行时装配": 0.022, + "模块服务": 0.024, + "插件服务装配": 0.022, + "消息队列": 0.021, + "插件备份恢复": 0.021, + "插件": 0.021, + "定时器": 0.021, + "监控器": 0.021, + "待处理整理回放": 0.021, + "命令服务": 0.021, "工作流": 0.021, - "插件同步与启动收尾": 0.02 + "插件同步与启动收尾": 0.022 }, "threads_before": 1, "threads_started": 1, @@ -159,8 +159,8 @@ "database_connections_started": 0 } ], - "median_startup_ms": 0.797, - "median_full_lifespan_ms": 1.851, + "median_startup_ms": 0.729, + "median_full_lifespan_ms": 1.879, "enabled_component_count": 32, "enabled_components": [ "文件日志", @@ -202,25 +202,25 @@ { "mode": "safe", "enabled_component_count": 20, - "startup_ms": 0.593, - "full_lifespan_ms": 1.266, + "startup_ms": 0.552, + "full_lifespan_ms": 1.178, "stage_ms": { - "文件日志": 0.07, - "后台任务登记器": 0.036, - "数据库准备": 0.029, - "HTTP 基础能力": 0.025, - "站点访问端口": 0.024, - "Chain 外部端口": 0.023, - "Chain 网络端口": 0.022, - "领域依赖装配": 0.022, - "数据库引擎预热": 0.019, - "数据库连接预算": 0.021, - "路由": 0.023, - "插件运行时装配": 0.021, - "模块服务": 0.021, - "插件服务装配": 0.021, - "消息队列": 0.021, - "插件同步与启动收尾": 0.019 + "文件日志": 0.065, + "后台任务登记器": 0.032, + "数据库准备": 0.028, + "HTTP 基础能力": 0.024, + "站点访问端口": 0.02, + "Chain 外部端口": 0.021, + "Chain 网络端口": 0.019, + "领域依赖装配": 0.021, + "数据库引擎预热": 0.021, + "数据库连接预算": 0.02, + "路由": 0.021, + "插件运行时装配": 0.016, + "模块服务": 0.018, + "插件服务装配": 0.018, + "消息队列": 0.019, + "插件同步与启动收尾": 0.018 }, "threads_before": 1, "threads_started": 1, @@ -233,23 +233,23 @@ { "mode": "safe", "enabled_component_count": 20, - "startup_ms": 0.596, - "full_lifespan_ms": 1.265, + "startup_ms": 0.558, + "full_lifespan_ms": 1.2, "stage_ms": { - "文件日志": 0.07, - "后台任务登记器": 0.037, - "数据库准备": 0.031, - "HTTP 基础能力": 0.027, - "站点访问端口": 0.023, - "Chain 外部端口": 0.022, - "Chain 网络端口": 0.02, - "领域依赖装配": 0.021, - "数据库引擎预热": 0.02, - "数据库连接预算": 0.021, - "路由": 0.023, - "插件运行时装配": 0.021, - "模块服务": 0.021, - "插件服务装配": 0.021, + "文件日志": 0.067, + "后台任务登记器": 0.036, + "数据库准备": 0.025, + "HTTP 基础能力": 0.024, + "站点访问端口": 0.02, + "Chain 外部端口": 0.02, + "Chain 网络端口": 0.019, + "领域依赖装配": 0.019, + "数据库引擎预热": 0.015, + "数据库连接预算": 0.019, + "路由": 0.021, + "插件运行时装配": 0.02, + "模块服务": 0.02, + "插件服务装配": 0.02, "消息队列": 0.02, "插件同步与启动收尾": 0.02 }, @@ -264,25 +264,25 @@ { "mode": "safe", "enabled_component_count": 20, - "startup_ms": 0.594, - "full_lifespan_ms": 1.236, + "startup_ms": 0.579, + "full_lifespan_ms": 1.211, "stage_ms": { - "文件日志": 0.068, - "后台任务登记器": 0.036, - "数据库准备": 0.028, + "文件日志": 0.062, + "后台任务登记器": 0.042, + "数据库准备": 0.027, "HTTP 基础能力": 0.028, - "站点访问端口": 0.024, - "Chain 外部端口": 0.023, - "Chain 网络端口": 0.022, - "领域依赖装配": 0.022, - "数据库引擎预热": 0.022, + "站点访问端口": 0.021, + "Chain 外部端口": 0.022, + "Chain 网络端口": 0.021, + "领域依赖装配": 0.021, + "数据库引擎预热": 0.019, "数据库连接预算": 0.021, "路由": 0.022, - "插件运行时装配": 0.021, - "模块服务": 0.021, - "插件服务装配": 0.021, - "消息队列": 0.021, - "插件同步与启动收尾": 0.021 + "插件运行时装配": 0.02, + "模块服务": 0.023, + "插件服务装配": 0.02, + "消息队列": 0.02, + "插件同步与启动收尾": 0.02 }, "threads_before": 1, "threads_started": 1, @@ -293,8 +293,8 @@ "database_connections_started": 0 } ], - "median_startup_ms": 0.594, - "median_full_lifespan_ms": 1.265, + "median_startup_ms": 0.558, + "median_full_lifespan_ms": 1.2, "enabled_component_count": 20, "enabled_components": [ "文件日志", diff --git a/tests/test_subscription_execution_status.py b/tests/test_subscription_execution_status.py index e7ef83b82..6e340b551 100644 --- a/tests/test_subscription_execution_status.py +++ b/tests/test_subscription_execution_status.py @@ -100,9 +100,7 @@ def test_execution_status_exposes_site_wait_and_cancel_capability(): repository = _Repository() repository.tasks[1] = _task(1, phase="waiting_site_budget") - statuses = asyncio.run( - SubscriptionExecutionStatusService(repository).for_subscriptions((1,)) - ) + statuses = asyncio.run(SubscriptionExecutionStatusService(repository).for_subscriptions((1,))) assert statuses[1].state == "waiting_site_budget" assert statuses[1].current_site_id == 9 @@ -115,9 +113,7 @@ def test_reconciliation_state_overrides_newer_search_terminal(): repository.tasks[2] = _task(2, state="failed", phase="failed") repository.downloads[2] = _download(2, "reconcile_required") - statuses = asyncio.run( - SubscriptionExecutionStatusService(repository).for_subscriptions((2,)) - ) + statuses = asyncio.run(SubscriptionExecutionStatusService(repository).for_subscriptions((2,))) assert statuses[2].state == "reconcile_required" assert statuses[2].requires_reconciliation is True @@ -130,9 +126,7 @@ def test_failed_search_exposes_safe_retryable_error(): repository = _Repository() repository.tasks[3] = _task(3, state="failed", phase="failed") - statuses = asyncio.run( - SubscriptionExecutionStatusService(repository).for_subscriptions((3,)) - ) + statuses = asyncio.run(SubscriptionExecutionStatusService(repository).for_subscriptions((3,))) assert statuses[3].state == "failed" assert statuses[3].can_retry is True @@ -145,15 +139,30 @@ def test_batch_requires_complete_subscription_access(): repository.tasks = {1: _task(1), 2: _task(2)} service = SubscriptionExecutionStatusService(repository) - hidden = asyncio.run( - service.get_batch("batch-1", accessible_subscription_ids={1}) - ) - visible = asyncio.run( - service.get_batch("batch-1", accessible_subscription_ids={1, 2}) - ) + hidden = asyncio.run(service.get_batch("batch-1", accessible_subscription_ids={1})) + visible = asyncio.run(service.get_batch("batch-1", accessible_subscription_ids={1, 2})) assert hidden is None assert visible is not None assert visible.current_subscription_id == 1 assert visible.processed_count == 0 assert visible.can_cancel is True + + +def test_request_cancel_uses_injected_execution_boundary(): + """取消必须通过组合根注入的异步执行边界并返回真实结果。""" + repository = _Repository() + requested: list[str] = [] + + async def request_cancel(batch_id: str) -> bool: + """记录测试请求并模拟队列接受取消。""" + requested.append(batch_id) + return True + + service = SubscriptionExecutionStatusService( + repository, + request_cancel=request_cancel, + ) + + assert asyncio.run(service.request_cancel("batch-1")) is True + assert requested == ["batch-1"]