fix(subscribe): satisfy final governance gates

This commit is contained in:
jxxghp
2026-09-01 17:50:08 +08:00
parent 8054fbafe6
commit 857aa813e5
10 changed files with 234 additions and 234 deletions

View File

@@ -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 多季订阅使用的请求级原子批量写端口。"""

View File

@@ -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 "订阅搜索批次已结束或无法取消",

View File

@@ -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]:

View File

@@ -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,

View File

@@ -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,

View File

@@ -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) |

View File

@@ -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 等基线均为零 |

View File

@@ -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",

View File

@@ -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": [
"文件日志",

View File

@@ -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"]