diff --git a/backend/package/yuxi/scheduler/adapters/persistence/mappers.py b/backend/package/yuxi/scheduler/adapters/persistence/mappers.py index e89dccfa..b81e27d0 100644 --- a/backend/package/yuxi/scheduler/adapters/persistence/mappers.py +++ b/backend/package/yuxi/scheduler/adapters/persistence/mappers.py @@ -88,6 +88,7 @@ def orm_to_run_log(orm: ScheduledTaskRunLogORM | None) -> ScheduledTaskRunLogDC return ScheduledTaskRunLogDC( id=orm.id, task_id=orm.task_id, + handler_name=orm.handler_name, run_id=orm.run_id, triggered_by=orm.triggered_by, status=orm.status, @@ -124,12 +125,13 @@ def handler_summary_row_to_dataclass(row: Any) -> HandlerSummaryDC: """仓储层 ``HandlerSummary`` → core ``HandlerSummary`` dataclass，按字段名映射。 仓储层 ``HandlerSummary`` 是 dataclass，row 可能是 NamedTuple 或 dataclass 实例， - 统一按字段名读取并转换为 core dataclass。``active_task_count`` 在仓储层聚合 - 时提供，缺失时默认 0。 + 统一按字段名读取并转换为 core dataclass。``active_task_count`` / + ``dead_letter_count`` 在仓储层聚合时提供，缺失时默认 0。 """ return HandlerSummaryDC( name=row.name, task_count=int(row.task_count), active_task_count=int(row.active_task_count) if hasattr(row, "active_task_count") else 0, + dead_letter_count=int(row.dead_letter_count) if hasattr(row, "dead_letter_count") else 0, last_active_at=row.last_active_at, ) diff --git a/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_run_log_daily_repo.py b/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_run_log_daily_repo.py index 44687bb5..08061966 100644 --- a/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_run_log_daily_repo.py +++ b/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_run_log_daily_repo.py @@ -48,6 +48,7 @@ class SqlAlchemyRunLogDailyRepository: handler_name: str, result: str, *, + increment: int = 1, commit: bool = True, ) -> None: """委托给 ScheduledTaskRunLogDailyRepository.upsert_daily_stat。""" @@ -55,6 +56,7 @@ class SqlAlchemyRunLogDailyRepository: stat_date, handler_name, result, + increment=increment, commit=commit, ) diff --git a/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_run_log_repo.py b/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_run_log_repo.py index b84e1018..c82ffdac 100644 --- a/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_run_log_repo.py +++ b/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_run_log_repo.py @@ -95,7 +95,8 @@ class SqlAlchemyRunLogRepository: async def list_by_status( self, *, - status: str, + status: str | None = None, + handler_name: str | None = None, started_after: datetime | None = None, started_before: datetime | None = None, page: int = 1, @@ -104,6 +105,7 @@ class SqlAlchemyRunLogRepository: """委托给 ScheduledTaskRunLogRepository.list_by_status。""" orms, total = await self._repo.list_by_status( status=status, + handler_name=handler_name, started_after=started_after, started_before=started_before, page=page, @@ -120,11 +122,15 @@ class SqlAlchemyRunLogRepository: """委托给 ScheduledTaskRunLogRepository.count_running。""" return await self._repo.count_running() + async def get_latest_started_at(self, *, exclude_skipped: bool = True) -> datetime | None: + """委托给 ScheduledTaskRunLogRepository.get_latest_started_at。""" + return await self._repo.get_latest_started_at(exclude_skipped=exclude_skipped) + async def reclaim_stale_runs( self, *, stale_before: datetime, - error_message: str = "reclaimed by scheduler restart", + error_message: str = "reclaimed as stale run", commit: bool = True, ) -> list[str]: """委托给 ScheduledTaskRunLogRepository.reclaim_stale_runs。 diff --git a/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_task_repo.py b/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_task_repo.py index 51670ea1..c66a3b7e 100644 --- a/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_task_repo.py +++ b/backend/package/yuxi/scheduler/adapters/persistence/sqlalchemy_task_repo.py @@ -85,10 +85,13 @@ class SqlAlchemyTaskRepository: owner_scope: str | None = None, owner_id: str | None = None, handler_name: str | None = None, + keyword: str | None = None, enabled: bool | None = None, status: str | None = None, created_after: datetime | None = None, created_before: datetime | None = None, + sort_by: str | None = None, + sort_order: str | None = None, page: int = 1, page_size: int = 20, ) -> tuple[list[ScheduledTaskDC], int]: @@ -97,10 +100,13 @@ class SqlAlchemyTaskRepository: owner_scope=owner_scope, owner_id=owner_id, handler_name=handler_name, + keyword=keyword, enabled=enabled, status=status, created_after=created_after, created_before=created_before, + sort_by=sort_by, + sort_order=sort_order, page=page, page_size=page_size, ) @@ -121,6 +127,26 @@ class SqlAlchemyTaskRepository: """委托给 ScheduledTaskRepository.delete。""" return await self._repo.delete(task_id, commit=commit) + async def batch_pause( + self, + task_ids: list[str], + *, + updated_by: str | None = None, + commit: bool = True, + ) -> int: + """委托给 ScheduledTaskRepository.batch_pause。""" + return await self._repo.batch_pause(task_ids, updated_by=updated_by, commit=commit) + + async def batch_soft_delete( + self, + task_ids: list[str], + *, + updated_by: str | None = None, + commit: bool = True, + ) -> int: + """委托给 ScheduledTaskRepository.batch_soft_delete。""" + return await self._repo.batch_soft_delete(task_ids, updated_by=updated_by, commit=commit) + # ------------------------------------------------------------------ # 业务查询 # ------------------------------------------------------------------ @@ -206,9 +232,7 @@ class SqlAlchemyTaskRepository: commit: bool = True, ) -> ScheduledTaskDC | None: """委托给 ScheduledTaskRepository.resume。""" - orm = await self._repo.resume( - task_id, next_run_at=next_run_at, updated_by=updated_by, commit=commit - ) + orm = await self._repo.resume(task_id, next_run_at=next_run_at, updated_by=updated_by, commit=commit) return orm_to_task(orm) async def mark_dead_letter( @@ -267,9 +291,7 @@ class SqlAlchemyTaskRepository: commit: bool = True, ) -> ScheduledTaskDC | None: """委托给 ScheduledTaskRepository.finalize_one_shot_task。""" - orm = await self._repo.finalize_one_shot_task( - task_id, updated_by=updated_by, commit=commit - ) + orm = await self._repo.finalize_one_shot_task(task_id, updated_by=updated_by, commit=commit) return orm_to_task(orm) # ------------------------------------------------------------------ diff --git a/backend/package/yuxi/scheduler/core/models.py b/backend/package/yuxi/scheduler/core/models.py index 0e658cfb..113fe3e4 100644 --- a/backend/package/yuxi/scheduler/core/models.py +++ b/backend/package/yuxi/scheduler/core/models.py @@ -63,6 +63,7 @@ class ScheduledTaskRunLog: task_id: str run_id: str + handler_name: str | None = None triggered_by: str = "auto" status: str = "running" # running / success / failure / timeout / skipped error_message: str | None = None @@ -82,12 +83,13 @@ class ScheduledTaskRunLog: class HandlerSummary: """handler 聚合摘要（与仓储层 dataclass 同名同字段，core 层独立声明）。 - 用于 Admin API 展示各 handler 的任务数、活跃任务数与最近执行时间。 + 用于 Admin API 展示各 handler 的任务数、活跃任务数、死信任务数与最近执行时间。 """ name: str task_count: int active_task_count: int = 0 + dead_letter_count: int = 0 last_active_at: Any = None diff --git a/backend/package/yuxi/scheduler/core/ports.py b/backend/package/yuxi/scheduler/core/ports.py index a87780dc..156540e1 100644 --- a/backend/package/yuxi/scheduler/core/ports.py +++ b/backend/package/yuxi/scheduler/core/ports.py @@ -177,6 +177,14 @@ class ScheduledTaskRepositoryPort(Protocol): handler_name: str | None = None, ) -> dict[str, int]: ... + async def list_anomalies( + self, + *, + consecutive_error_threshold: int = 3, + stale_days: int = 7, + limit: int = 5, + ) -> dict[str, list[ScheduledTask]]: ... + # 并发控制 async def acquire_for_run( @@ -280,7 +288,8 @@ class ScheduledTaskRunLogRepositoryPort(Protocol): async def list_by_status( self, *, - status: str, + status: str | None = None, + handler_name: str | None = None, started_after: datetime | None = None, started_before: datetime | None = None, page: int = 1, @@ -291,11 +300,13 @@ class ScheduledTaskRunLogRepositoryPort(Protocol): async def count_running(self) -> int: ... + async def get_latest_started_at(self, *, exclude_skipped: bool = True) -> datetime | None: ... + async def reclaim_stale_runs( self, *, stale_before: datetime, - error_message: str = "reclaimed by scheduler restart", + error_message: str = "reclaimed as stale run", commit: bool = True, ) -> list[str]: ... @@ -316,6 +327,7 @@ class ScheduledTaskRunLogDailyRepositoryPort(Protocol): handler_name: str, result: str, *, + increment: int = 1, commit: bool = True, ) -> None: ... diff --git a/backend/package/yuxi/scheduler/framework/handlers/__init__.py b/backend/package/yuxi/scheduler/framework/handlers/__init__.py index d096cf9d..4dd735ed 100644 --- a/backend/package/yuxi/scheduler/framework/handlers/__init__.py +++ b/backend/package/yuxi/scheduler/framework/handlers/__init__.py @@ -13,6 +13,7 @@ handler（``external_systems`` 等）的区别： - ``RunLogCleanupHandler``：清理过期的执行明细日志（PRD §FR-ST-11） - ``IdempotencyCleanupHandler``：清理过期的幂等记录 +- ``TaskRecycleCleanupHandler``：清理过期的回收站任务（超期自动硬删除） 依赖方向：仅依赖 ``core/contracts``（``TaskHandler`` 协议）+ ``adapters/persistence`` （``create_repositories``）+ ``utils``。不依赖 ``use_cases`` / ``repositories`` / @@ -26,8 +27,12 @@ from yuxi.scheduler.framework.handlers.idempotency_cleanup_handler import ( from yuxi.scheduler.framework.handlers.run_log_cleanup_handler import ( RunLogCleanupHandler, ) +from yuxi.scheduler.framework.handlers.task_recycle_cleanup_handler import ( + TaskRecycleCleanupHandler, +) __all__ = [ "IdempotencyCleanupHandler", "RunLogCleanupHandler", + "TaskRecycleCleanupHandler", ] diff --git a/backend/package/yuxi/scheduler/framework/handlers/task_recycle_cleanup_handler.py b/backend/package/yuxi/scheduler/framework/handlers/task_recycle_cleanup_handler.py new file mode 100644 index 00000000..1e5e02df --- /dev/null +++ b/backend/package/yuxi/scheduler/framework/handlers/task_recycle_cleanup_handler.py @@ -0,0 +1,94 @@ +"""清理过期回收站任务的维护型 handler。 + +由 scheduler 自身在 worker 启动时注册到 ``HandlerRegistry``，通过 +``scheduled_tasks`` 表配置触发周期。按保留天数物理删除已软删除且超过 +保留期的任务记录，回收存储空间。 + +设计要点（见设计方案 §14.1）： + +- **与 ``RunLogCleanupHandler`` 结构对齐**：差异仅在清理对象。 +- **会话隔离**：通过构造函数注入 ``session_factory``，每次执行创建独立 db + 会话，避免 worker 进程长生命周期会话导致的连接失效问题。 +- **保留策略可配置**：保留天数从 ``payload.retention_days`` 读取，未传入时 + 使用构造函数默认值（对齐 ``config.scheduler_task_recycle_retention_days``）。 +- **物理删除**：通过仓储的 ``hard_delete_before`` 方法执行，物理删除 + ``deleted_at`` 早于 cutoff 的软删除记录。 +- **异常隔离**：handler 内部捕获所有异常并转换为 ``TaskResult(success=False)``， + 不向调度域传播业务异常（对齐 INV-CROSS-4）。 +""" + +from __future__ import annotations + +from collections.abc import Callable +from contextlib import AbstractAsyncContextManager +from datetime import timedelta + +from sqlalchemy.ext.asyncio import AsyncSession + +from yuxi.scheduler.adapters.persistence import create_repositories +from yuxi.scheduler.core.contracts import TaskContext, TaskResult +from yuxi.utils.datetime_utils import utc_now_naive +from yuxi.utils.logging_config import logger + +# 默认保留天数（与 config.scheduler_task_recycle_retention_days 默认值对齐） +_DEFAULT_RETENTION_DAYS = 7 + + +class TaskRecycleCleanupHandler: + """清理过期回收站任务的维护型 handler。 + + 通过 ``scheduled_tasks`` 表配置触发，按保留天数物理删除 + ``scheduled_tasks`` 表中已软删除且 ``deleted_at`` 早于 cutoff 的记录。 + + payload 参数： + retention_days: 保留天数，未传入时使用构造函数默认值。 + + 依赖注入： + session_factory: 调用返回 ``AsyncSession`` 的异步上下文管理器工厂， + 通常为 ``pg_manager.get_async_session_context``。每次执行创建独立 + 会话，避免长生命周期会话问题。 + default_retention_days: 默认保留天数，未在 payload 指定时使用。 + """ + + name = "task_recycle_cleanup" + description = "清理过期的回收站任务（超期自动硬删除）" + + def __init__( + self, + session_factory: Callable[[], AbstractAsyncContextManager[AsyncSession]], + *, + default_retention_days: int = _DEFAULT_RETENTION_DAYS, + ) -> None: + self._session_factory = session_factory + self._default_retention_days = default_retention_days + + async def execute(self, ctx: TaskContext) -> TaskResult: + retention_days = int(ctx.payload.get("retention_days", self._default_retention_days)) + cutoff = utc_now_naive() - timedelta(days=retention_days) + + try: + async with self._session_factory() as db: + repos = create_repositories(db) + deleted_count = await repos.task.hard_delete_before(cutoff) + except Exception as e: + logger.exception( + "scheduler_task_recycle_cleanup_failed", + extra={"retention_days": retention_days, "cutoff": cutoff}, + ) + return TaskResult(success=False, error=str(e)) + + logger.info( + "scheduler_task_recycle_cleanup", + extra={ + "deleted_count": deleted_count, + "retention_days": retention_days, + "cutoff": cutoff, + }, + ) + return TaskResult( + success=True, + output={ + "deleted_count": deleted_count, + "retention_days": retention_days, + }, + ) diff --git a/backend/package/yuxi/scheduler/use_cases/dto/scheduler.py b/backend/package/yuxi/scheduler/use_cases/dto/scheduler.py index 55452ff2..e9de670c 100644 --- a/backend/package/yuxi/scheduler/use_cases/dto/scheduler.py +++ b/backend/package/yuxi/scheduler/use_cases/dto/scheduler.py @@ -301,12 +301,41 @@ class ListTasksInput(BaseModel): owner_scope: str | None = Field(default=None, max_length=64) owner_id: str | None = Field(default=None, max_length=128) handler_name: str | None = Field(default=None, max_length=128) + keyword: str | None = Field( + default=None, + max_length=128, + description="关键词模糊搜索，匹配 task_id 或 handler_name", + ) enabled: bool | None = None status: str | None = Field( default=None, pattern=r"^(active|paused|dead_letter)$", description="任务状态过滤", ) + sort_by: str | None = Field( + default=None, + pattern=r"^(created_at|next_run_at|last_run_at|consecutive_errors|updated_at)$", + description="排序字段", + ) + sort_order: str | None = Field( + default=None, + pattern=r"^(asc|desc)$", + description="排序方向，默认 desc", + ) + + +class BatchTaskOperationOutput(BaseModel): + """批量任务操作输出 DTO。 + + ``success_count`` 为实际受影响行数，``skipped_count`` 为未匹配或状态 + 不允许的任务数（``task_ids`` 总数减去 ``success_count``）。 + """ + + model_config = ConfigDict(frozen=True) + + success_count: int = Field(..., description="实际操作成功的任务数") + skipped_count: int = Field(..., description="未匹配或状态不允许的任务数") + failed: list[str] = Field(default_factory=list, description="未成功的 task_id 列表") class ListTasksOutput(BaseModel): @@ -378,6 +407,7 @@ class RunLogOutput(BaseModel): model_config = ConfigDict(frozen=True) task_id: str + handler_name: str | None = None run_id: str triggered_by: str = "auto" status: str = "running" @@ -413,7 +443,8 @@ class ListRunLogsOutput(BaseModel): class ListAllRunLogsInput(BaseModel): """跨任务执行日志查询输入 DTO。 - ``task_id`` 与 ``status`` 均可选，但不可同时为空（由用例层校验）， + ``task_id`` / ``status`` / ``handler_name`` / 时间范围均可选， + 但至少需指定 ``task_id``、``status`` 或时间范围之一（由用例层校验）， 避免无过滤的全表扫描。``start_date`` / ``end_date`` 为 ISO 格式字符串， 由用例层解析为 ``datetime`` 后传入仓储层。 """ @@ -427,6 +458,12 @@ class ListAllRunLogsInput(BaseModel): pattern=r"^(running|success|failure|timeout|skipped)$", description="执行状态过滤", ) + handler_name: str | None = Field( + default=None, + min_length=1, + max_length=128, + description="handler 名称过滤（跨任务按 handler 维度查询）", + ) start_date: str | None = Field( default=None, description="起始时间（ISO 格式字符串），按 started_at 过滤", @@ -453,9 +490,9 @@ class HandlerSummaryOutput(BaseModel): model_config = ConfigDict(frozen=True) handler_name: str - description: str = "" task_count: int = 0 active_task_count: int = 0 + dead_letter_count: int = 0 last_active_at: str | None = None @@ -520,12 +557,51 @@ class ListUpcomingOutput(BaseModel): items: list[TaskOutput] = Field(default_factory=list) +class ListAnomaliesOutput(BaseModel): + """工作台异常任务聚合输出 DTO。 + + 聚合三类异常任务供工作台一次性拉取，避免前端依赖活跃任务列表前 N 条 + 做前端过滤导致的漏报问题： + - ``dead_letter``：死信任务（status=dead_letter，按 updated_at 降序） + - ``consecutive_failure``：连续失败任务（consecutive_errors >= 3， + 非 dead_letter，按 consecutive_errors 降序） + - ``stale``：长期未执行任务（active 状态，last_run_at 早于 7 天前， + 按 last_run_at 升序） + """ + + model_config = ConfigDict(frozen=True) + + dead_letter: list[TaskOutput] = Field(default_factory=list) + consecutive_failure: list[TaskOutput] = Field(default_factory=list) + stale: list[TaskOutput] = Field(default_factory=list) + + +class SchedulerConfigOutput(BaseModel): + """调度器配置快照（只读），用于运维中心展示当前生效配置。 + + 字段值由 ``scheduler_service.get_health`` 从 ``app_config`` 读取并填充， + 确保前端展示与后端实际配置一致，避免硬编码默认值导致的信息不同步。 + """ + + model_config = ConfigDict(frozen=True) + + enabled: bool = Field(description="定时任务总开关") + tick_interval_seconds: int = Field(description="tick 频率（秒）") + task_timeout_seconds: int = Field(description="单任务超时（秒）") + max_consecutive_errors: int = Field(description="死信阈值（连续失败次数）") + backoff_schedule: list[int] = Field(description="指数退避序列（秒）") + stagger_max_seconds: int = Field(description="整点抖动上限（秒）") + run_log_retention_days: int = Field(description="run_logs 保留天数") + idempotency_retention_hours: int = Field(description="幂等记录保留小时数") + min_cron_interval_minutes: int = Field(description="cron 最小间隔（分钟）") + + class GetHealthOutput(BaseModel): """调度器健康检查输出 DTO。 - 对齐 PRD §FR-ST-07 健康检查端点响应：``status`` 表示 worker 是否在最近 - 5 分钟内执行 tick；``last_tick_at`` 为最近 tick 时间；``active_task_count`` - / ``dead_letter_count`` / ``running_count`` 为关键状态计数。 + 对齐 PRD §FR-ST-07 健康检查端点响应：``status`` 表示 worker 活跃度健康 + 判断结果；``last_tick_at`` 为最近一次任务活动时间（不限状态）；``config`` + 为当前生效的调度器配置快照。 """ model_config = ConfigDict(frozen=True) @@ -533,12 +609,13 @@ class GetHealthOutput(BaseModel): status: str = Field( default="healthy", pattern=r"^(healthy|unhealthy)$", - description="健康状态：healthy（最近 5 分钟内有 tick）/ unhealthy", + description="健康状态：healthy / unhealthy", ) - last_tick_at: str | None = Field(default=None, description="最近 tick 时间（ISO 格式字符串）") + last_tick_at: str | None = Field(default=None, description="最近一次任务活动时间（ISO 格式字符串）") active_task_count: int = 0 dead_letter_count: int = 0 running_count: int = 0 + config: SchedulerConfigOutput = Field(description="当前生效的调度器配置快照") # --------------------------------------------------------------------------- @@ -564,7 +641,8 @@ class DailyStatOutput(BaseModel): """日聚合统计输出 DTO。字段对齐 ``DailyStat`` dataclass，补充派生字段。 ``total_count`` / ``success_rate`` 为派生字段，由 mapper 计算： - ``total_count = success_count + failure_count + timeout_count``， + ``total_count = success_count + failure_count + timeout_count + dead_letter_count`` + （dead_letter 视为失败终态，纳入分母）， ``success_rate = success_count / total_count if total_count > 0 else 0.0``。 """ @@ -596,9 +674,9 @@ class ListDailyStatsOutput(BaseModel): class ReclaimStaleRunsInput(BaseModel): """回收僵尸执行输入 DTO。 - 用于 scheduler 重启后回收 ``status=running`` 但实际已超时的执行记录， - 将其标记为 ``failure`` 并写入回收错误信息。``timeout_seconds`` 为可选参数， - 未传入时由用例层从配置取默认值。 + 用于回收 ``status=running`` 但实际已超时的执行记录，将其标记为 ``timeout`` + 并写入回收错误信息。``timeout_seconds`` 为可选参数，未传入时由用例层从配置 + 取默认值（``scheduler_task_timeout_seconds``）。 """ model_config = ConfigDict(frozen=True) diff --git a/backend/package/yuxi/scheduler/use_cases/mappers.py b/backend/package/yuxi/scheduler/use_cases/mappers.py index 8e26da71..fdbc6da6 100644 --- a/backend/package/yuxi/scheduler/use_cases/mappers.py +++ b/backend/package/yuxi/scheduler/use_cases/mappers.py @@ -26,10 +26,12 @@ from yuxi.scheduler.use_cases.dto.scheduler import ( DailyStatOutput, GetHealthOutput, HandlerSummaryOutput, + ListAnomaliesOutput, ListDailyStatsOutput, ListHandlerSummaryOutput, ListUpcomingOutput, RunLogOutput, + SchedulerConfigOutput, TaskOutput, TriggerTaskOutput, ) @@ -41,6 +43,7 @@ __all__ = [ "to_daily_stat_output", "to_get_health_output", "to_handler_summary_output", + "to_list_anomalies_output", "to_list_daily_stats_output", "to_run_log_output", "to_task_output", @@ -99,6 +102,7 @@ def to_run_log_output(log: ScheduledTaskRunLog) -> RunLogOutput: duration_seconds = round(delta.total_seconds(), 3) return RunLogOutput( task_id=log.task_id, + handler_name=log.handler_name, run_id=log.run_id, triggered_by=log.triggered_by, status=log.status, @@ -120,14 +124,13 @@ def to_run_log_output(log: ScheduledTaskRunLog) -> RunLogOutput: def to_handler_summary_output(summary: HandlerSummary) -> HandlerSummaryOutput: """``HandlerSummary`` dataclass → ``HandlerSummaryOutput`` DTO。 - ``description`` 字段仓储层不聚合（DB 无此列），由调用方（如 worker 进程 - 持有 ``HandlerRegistry`` 时）按需补充；API 进程无注册表时保持空字符串。 - ``active_task_count`` 已由仓储层聚合，此处直接透传。 + ``active_task_count`` / ``dead_letter_count`` 已由仓储层聚合，此处直接透传。 """ return HandlerSummaryOutput( handler_name=summary.name, task_count=summary.task_count, active_task_count=summary.active_task_count, + dead_letter_count=summary.dead_letter_count, last_active_at=format_utc_datetime(summary.last_active_at), ) @@ -152,10 +155,12 @@ def to_get_health_output( dead_letter_count: int, running_count: int, healthy: bool, + config: SchedulerConfigOutput, ) -> GetHealthOutput: """构造 ``GetHealthOutput`` DTO。 - ``healthy`` 由调用方按"最近 tick 是否在 5 分钟内"判断后传入。 + ``healthy`` 由调用方按 worker 活跃度判断后传入；``config`` 为从 + ``app_config`` 读取的当前生效配置快照。 """ return GetHealthOutput( status="healthy" if healthy else "unhealthy", @@ -163,6 +168,7 @@ def to_get_health_output( active_task_count=active_task_count, dead_letter_count=dead_letter_count, running_count=running_count, + config=config, ) @@ -186,9 +192,10 @@ def to_daily_stat_output(stat: DailyStat) -> DailyStatOutput: """``DailyStat`` dataclass → ``DailyStatOutput`` DTO。 计算派生字段 ``total_count`` / ``success_rate``，``stat_date`` 格式化为 - ISO 日期字符串（``YYYY-MM-DD``）。 + ISO 日期字符串（``YYYY-MM-DD``）。``dead_letter_count`` 视为失败终态， + 纳入 ``total_count`` 分母以反映真实成功率。 """ - total_count = stat.success_count + stat.failure_count + stat.timeout_count + total_count = stat.success_count + stat.failure_count + stat.timeout_count + stat.dead_letter_count success_rate = stat.success_count / total_count if total_count > 0 else 0.0 return DailyStatOutput( stat_date=stat.stat_date.isoformat(), @@ -212,6 +219,15 @@ def to_list_upcoming_output(tasks: list[ScheduledTask]) -> ListUpcomingOutput: return ListUpcomingOutput(items=[to_task_output(t) for t in tasks]) +def to_list_anomalies_output(groups: dict[str, list[ScheduledTask]]) -> ListAnomaliesOutput: + """异常任务分组映射 → ``ListAnomaliesOutput`` DTO。""" + return ListAnomaliesOutput( + dead_letter=[to_task_output(t) for t in groups.get("dead_letter", [])], + consecutive_failure=[to_task_output(t) for t in groups.get("consecutive_failure", [])], + stale=[to_task_output(t) for t in groups.get("stale", [])], + ) + + def to_list_handler_summary_output( summaries: list[HandlerSummary], ) -> ListHandlerSummaryOutput: diff --git a/backend/package/yuxi/scheduler/use_cases/services/scheduler_service.py b/backend/package/yuxi/scheduler/use_cases/services/scheduler_service.py index 8e8ecdbd..3098808c 100644 --- a/backend/package/yuxi/scheduler/use_cases/services/scheduler_service.py +++ b/backend/package/yuxi/scheduler/use_cases/services/scheduler_service.py @@ -47,6 +47,7 @@ from yuxi.scheduler.exceptions import ( TaskStatusTransitionError, ) from yuxi.scheduler.use_cases.dto.scheduler import ( + BatchTaskOperationOutput, CountByStatusInput, CountByStatusOutput, CreateTaskInput, @@ -56,6 +57,7 @@ from yuxi.scheduler.use_cases.dto.scheduler import ( GetTaskInput, HardDeleteTaskInput, ListAllRunLogsInput, + ListAnomaliesOutput, ListDailyStatsInput, ListDailyStatsOutput, ListDeletedTasksInput, @@ -72,6 +74,7 @@ from yuxi.scheduler.use_cases.dto.scheduler import ( RestoreTaskInput, ResumeTaskInput, RunLogOutput, + SchedulerConfigOutput, TaskOutput, TriggerTaskInput, TriggerTaskOutput, @@ -80,6 +83,7 @@ from yuxi.scheduler.use_cases.dto.scheduler import ( from yuxi.scheduler.use_cases.mappers import ( to_count_by_status_output, to_get_health_output, + to_list_anomalies_output, to_list_daily_stats_output, to_list_handler_summary_output, to_list_upcoming_output, @@ -93,6 +97,7 @@ from yuxi.scheduler.use_cases.utils.cron_calc import calc_next_run_at from yuxi.scheduler.use_cases.utils.stagger import calc_stagger_seconds from yuxi.utils.datetime_utils import utc_now_naive from yuxi.utils.logging_config import logger +from yuxi.utils.trace_context import get_trace_id if TYPE_CHECKING: from arq import ArqRedis @@ -101,8 +106,8 @@ if TYPE_CHECKING: from yuxi.scheduler.framework.runtime import RuntimeServices -# tick 健康检查的"最近活跃"窗口（分钟） -_HEALTH_RECENT_WINDOW_MINUTES = 5 +# 健康检查窗口下限（秒）：有活跃任务时，最近活动超过此窗口则判定不健康 +_HEALTH_WINDOW_MIN_SECONDS = 1800 # 日聚合统计的自然日时区（与 ORM stat_date 注释约定一致） _DAILY_STAT_TZ = ZoneInfo("Asia/Shanghai") @@ -457,8 +462,11 @@ class SchedulerService(SchedulerServicePort): owner_scope=input_dto.owner_scope, owner_id=input_dto.owner_id, handler_name=input_dto.handler_name, + keyword=input_dto.keyword, enabled=input_dto.enabled, status=input_dto.status, + sort_by=input_dto.sort_by, + sort_order=input_dto.sort_order, page=input_dto.page, page_size=input_dto.page_size, ) @@ -496,9 +504,7 @@ class SchedulerService(SchedulerServicePort): details={"task_id": input_dto.task_id}, ) - update_data: dict[str, Any] = input_dto.model_dump( - exclude_unset=True, exclude={"task_id", "idempotency_key"} - ) + update_data: dict[str, Any] = input_dto.model_dump(exclude_unset=True, exclude={"task_id", "idempotency_key"}) if not update_data: return to_task_output(task) @@ -545,15 +551,11 @@ class SchedulerService(SchedulerServicePort): min_interval_minutes=self._config.scheduler_min_cron_interval_minutes, ) effective_tz = update_data.get("tz", task.tz) - update_data["next_run_at"] = calc_next_run_at( - update_data["cron_expression"], tz=effective_tz - ) + update_data["next_run_at"] = calc_next_run_at(update_data["cron_expression"], tz=effective_tz) # run_at 变更时重算 next_run_at（at 类型），校验未来时间 if "run_at" in update_data and task.schedule_kind == "at": - run_at_dt = self._parse_iso_to_utc_naive( - update_data["run_at"], field_name="run_at" - ) + run_at_dt = self._parse_iso_to_utc_naive(update_data["run_at"], field_name="run_at") if run_at_dt <= utc_now_naive(): raise SchedulerValidationError("run_at 必须为未来时间") update_data["next_run_at"] = run_at_dt @@ -583,9 +585,7 @@ class SchedulerService(SchedulerServicePort): operation="update", operator=input_dto.updated_by, ) - updated = await self._repos.task.update( - input_dto.task_id, update_data, commit=False - ) + updated = await self._repos.task.update(input_dto.task_id, update_data, commit=False) if updated is not None: await self._store_idempotency_response( input_dto.idempotency_key, @@ -641,6 +641,158 @@ class SchedulerService(SchedulerServicePort): extra={"task_id": input_dto.task_id, "operator": input_dto.updated_by}, ) + # ================================================================== + # 批量操作 + # ================================================================== + + async def batch_pause_tasks( + self, + task_ids: list[str], + *, + operator: str | None = None, + ) -> BatchTaskOperationOutput: + """批量暂停任务（仅 ``active`` 状态被转换）。 + + 单事务单 SQL 完成，``paused`` / ``dead_letter`` 状态的任务跳过。 + 不支持幂等键（批量操作的幂等由调用方按业务场景处理）。 + """ + if not task_ids: + return BatchTaskOperationOutput(success_count=0, skipped_count=0, failed=[]) + async with self._uow: + affected = await self._repos.task.batch_pause( + task_ids, + updated_by=operator, + commit=False, + ) + failed = list(set(task_ids) - set(await self._resolve_affected_task_ids(task_ids, affected))) + logger.info( + "scheduler_batch_paused", + extra={ + "total": len(task_ids), + "affected": affected, + "operator": operator, + "trace_id": get_trace_id(), + }, + ) + return BatchTaskOperationOutput( + success_count=affected, + skipped_count=len(task_ids) - affected, + failed=failed, + ) + + async def batch_resume_tasks( + self, + task_ids: list[str], + *, + operator: str | None = None, + ) -> BatchTaskOperationOutput: + """批量恢复任务（``paused`` → ``active``，``dead_letter`` → ``active``）。 + + 由于每个任务需按其 ``cron_expression`` / ``run_at`` 重算 ``next_run_at``， + 在单事务内逐个调用 ``resume`` / ``reset_from_dead_letter``。 + 非法状态的任务跳过并记录到 ``failed``。 + """ + if not task_ids: + return BatchTaskOperationOutput(success_count=0, skipped_count=0, failed=[]) + success_count = 0 + failed: list[str] = [] + async with self._uow: + for task_id in task_ids: + existing = await self._repos.task.get_by_task_id(task_id, for_update=True) + if existing is None: + failed.append(task_id) + continue + next_run_at = self._calc_next_run_at(existing) + if existing.status == "paused": + task = await self._repos.task.resume( + task_id, + next_run_at=next_run_at, + updated_by=operator, + commit=False, + ) + elif existing.status == "dead_letter": + task = await self._repos.task.reset_from_dead_letter( + task_id, + next_run_at=next_run_at, + updated_by=operator, + commit=False, + ) + else: + failed.append(task_id) + continue + if task is None: + failed.append(task_id) + else: + success_count += 1 + logger.info( + "scheduler_batch_resumed", + extra={ + "total": len(task_ids), + "affected": success_count, + "operator": operator, + "trace_id": get_trace_id(), + }, + ) + return BatchTaskOperationOutput( + success_count=success_count, + skipped_count=len(task_ids) - success_count, + failed=failed, + ) + + async def batch_delete_tasks( + self, + task_ids: list[str], + *, + operator: str | None = None, + ) -> BatchTaskOperationOutput: + """批量软删除任务。 + + 单事务单 SQL 完成，已软删除的任务跳过。 + """ + if not task_ids: + return BatchTaskOperationOutput(success_count=0, skipped_count=0, failed=[]) + async with self._uow: + affected = await self._repos.task.batch_soft_delete( + task_ids, + updated_by=operator, + commit=False, + ) + logger.info( + "scheduler_batch_deleted", + extra={ + "total": len(task_ids), + "affected": affected, + "operator": operator, + "trace_id": get_trace_id(), + }, + ) + return BatchTaskOperationOutput( + success_count=affected, + skipped_count=len(task_ids) - affected, + failed=[], + ) + + async def _resolve_affected_task_ids( + self, + task_ids: list[str], + expected_count: int, + ) -> list[str]: + """查询实际被批量暂停影响的 task_id 列表（用于 failed 推断）。 + + 简化实现：当 affected == len(task_ids) 时直接返回全部，否则查询当前 + ``paused`` 状态的 task_id 交集。 + """ + if expected_count >= len(task_ids): + return list(task_ids) + # 查询当前已暂停的任务（刚被批量暂停的目标） + affected_tasks, _ = await self._repos.task.list( + status="paused", + page=1, + page_size=100, + ) + affected_ids = {t.task_id for t in affected_tasks} + return [tid for tid in task_ids if tid in affected_ids] + # ================================================================== # 状态机 # ================================================================== @@ -879,8 +1031,9 @@ class SchedulerService(SchedulerServicePort): async def list_all_run_logs(self, input_dto: ListAllRunLogsInput) -> ListRunLogsOutput: """跨任务列出执行日志（FR-ST-09）。 - ``task_id`` 非空时按任务维度查询；``task_id`` 为空时按 ``status`` 跨任务 - 查询。两者不可同时为空，否则抛 ``SchedulerValidationError``。 + ``task_id`` 非空时按任务维度查询；否则按 ``status`` / ``handler_name`` / + 时间范围跨任务查询。至少需指定 ``task_id``、``status`` 或时间范围之一， + 避免无过滤的全表扫描。``handler_name`` 为可选过滤维度。 """ started_after: datetime | None = None started_before: datetime | None = None @@ -898,9 +1051,11 @@ class SchedulerService(SchedulerServicePort): page=input_dto.page, page_size=input_dto.page_size, ) - elif input_dto.status: + elif input_dto.status or started_after is not None or started_before is not None: + # status 可为 None（仅靠时间范围 / handler_name 过滤的"浏览全部"场景） logs, total = await self._repos.run_log.list_by_status( status=input_dto.status, + handler_name=input_dto.handler_name, started_after=started_after, started_before=started_before, page=input_dto.page, @@ -908,7 +1063,7 @@ class SchedulerService(SchedulerServicePort): ) else: raise SchedulerValidationError( - "task_id 与 status 不可同时为空", + "至少需要指定 task_id、status 或时间范围之一", details={"reason": "至少需要一个过滤条件避免全表扫描"}, ) @@ -950,6 +1105,15 @@ class SchedulerService(SchedulerServicePort): ) return to_list_upcoming_output(tasks) + async def list_anomalies(self) -> ListAnomaliesOutput: + """列出工作台异常任务聚合（死信 / 连续失败 / 长期未执行）。 + + 委托 ``task_repo.list_anomalies`` 在数据库层精准过滤，避免前端依赖 + 活跃任务列表前 N 条做前端过滤导致的漏报问题。 + """ + groups = await self._repos.task.list_anomalies() + return to_list_anomalies_output(groups) + async def list_daily_stats(self, input_dto: ListDailyStatsInput) -> ListDailyStatsOutput: """列出日聚合统计（FR-ST-09）。 @@ -1029,6 +1193,7 @@ class SchedulerService(SchedulerServicePort): await self._repos.run_log.create( { "task_id": task.task_id, + "handler_name": task.handler_name, "run_id": run_id, "triggered_by": "auto", "status": "skipped", @@ -1153,6 +1318,7 @@ class SchedulerService(SchedulerServicePort): await self._repos.run_log.create( { "task_id": task_id, + "handler_name": task.handler_name, "run_id": run_id, "triggered_by": triggered_by, "status": "running", @@ -1397,44 +1563,70 @@ class SchedulerService(SchedulerServicePort): async def get_health(self) -> GetHealthOutput: """获取调度器健康状态（FR-ST-07）。 - 检测 worker 是否在最近活跃窗口内有执行记录。v1 通过查询最近 run_log - 的 ``started_at`` 近似判断；v2 评估独立 tick 心跳表。 + 健康判断逻辑（基于 worker 活跃度，非固定 5 分钟窗口）： + + - 有 running 执行 → 健康（worker 正在执行任务） + - 无活跃任务（active=0）→ 健康（worker 无事可做，不报警） + - 有活跃任务时，检查最近一次任务活动是否在健康窗口内； + 窗口取 ``max(task_timeout_seconds * 2, _HEALTH_WINDOW_MIN_SECONDS)``， + 覆盖 tick 周期与任务超时，避免 cron 间隔较大时的误报。 + + ``last_tick_at`` 取最近一条非 skipped 执行日志的 ``started_at``（不限 + 窗口），表示"最近一次任务活动时间"。``config`` 返回当前生效配置快照， + 供前端展示，避免硬编码默认值导致的信息不同步。 """ - running_count = await self._repos.run_log.count_running() - status_dist = await self._repos.task.count_by_status() - - # 查询最近活跃窗口内的执行记录（近似判断 worker 是否在运行） now = utc_now_naive() - window_start = now - timedelta(minutes=_HEALTH_RECENT_WINDOW_MINUTES) - recent_logs, recent_total = await self._repos.run_log.list_by_status( - status="success", - started_after=window_start, - page=1, - page_size=1, + running_count, status_dist, last_activity_at = await asyncio.gather( + self._repos.run_log.count_running(), + self._repos.task.count_by_status(), + self._repos.run_log.get_latest_started_at(), ) - last_tick_at = recent_logs[0].started_at if recent_logs else None + active_task_count = status_dist.get("active", 0) + dead_letter_count = status_dist.get("dead_letter", 0) - # 若最近窗口内有成功记录，或当前有运行中任务，则视为健康 - healthy = recent_total > 0 or running_count > 0 + health_window = timedelta( + seconds=max(self._config.scheduler_task_timeout_seconds * 2, _HEALTH_WINDOW_MIN_SECONDS) + ) + if running_count > 0 or active_task_count == 0: + healthy = True + elif last_activity_at is not None and (now - last_activity_at) <= health_window: + healthy = True + else: + healthy = False + + config = SchedulerConfigOutput( + enabled=self._config.scheduler_enabled, + tick_interval_seconds=self._config.scheduler_tick_interval_seconds, + task_timeout_seconds=self._config.scheduler_task_timeout_seconds, + max_consecutive_errors=self._config.scheduler_max_consecutive_errors, + backoff_schedule=list(self._config.scheduler_backoff_schedule), + stagger_max_seconds=self._config.scheduler_stagger_max_seconds, + run_log_retention_days=self._config.scheduler_run_log_retention_days, + idempotency_retention_hours=self._config.scheduler_idempotency_retention_hours, + min_cron_interval_minutes=self._config.scheduler_min_cron_interval_minutes, + ) return to_get_health_output( - last_tick_at=last_tick_at, - active_task_count=status_dist.get("active", 0), - dead_letter_count=status_dist.get("dead_letter", 0), + last_tick_at=last_activity_at, + active_task_count=active_task_count, + dead_letter_count=dead_letter_count, running_count=running_count, healthy=healthy, + config=config, ) async def reclaim_stale_runs(self, input_dto: ReclaimStaleRunsInput) -> ReclaimStaleRunsOutput: """回收僵尸执行（运维恢复）。 - scheduler 重启后回收 ``status=running`` 但实际已超时的执行记录， - 将其标记为 ``timeout`` 并写入回收错误信息。同时按 handler_name 分组 - 补录 ``timeout_count`` 到日聚合统计表，保证统计准确（此前 - ``timeout_count`` 字段无写入路径，永远为 0）。 + 回收 ``status=running`` 但实际已超时的执行记录，将其标记为 ``timeout`` + 并写入回收错误信息。同时按 handler_name 分组补录 ``timeout_count`` 到 + 日聚合统计表，保证统计准确（此前 ``timeout_count`` 字段无写入路径， + 永远为 0）。 - reclaim 与补录在同一事务内完成，保证原子性：补录失败时 reclaim - 也回滚，避免出现"已标记 timeout 但未计入聚合"的中间状态。 + reclaim 与补录在同一事务内完成，保证原子性：补录失败时 reclaim 也 + 回滚，避免出现"已标记 timeout 但未计入聚合"的中间状态。补录使用 + ``increment`` 参数按 handler 分组单次 upsert ``+N``，避免逐条 upsert + 产生 N 次 SQL。 """ timeout_seconds = ( input_dto.timeout_seconds @@ -1445,19 +1637,20 @@ class SchedulerService(SchedulerServicePort): async with self._uow: handler_names = await self._repos.run_log.reclaim_stale_runs( - stale_before=stale_before, commit=False + stale_before=stale_before, + error_message="reclaimed by admin operation", + commit=False, ) if handler_names: today = self._daily_stat_date(utc_now_naive()) - # 按 handler_name 分组计数，逐条 upsert 补录 timeout_count + # 按 handler_name 分组计数，单次 upsert 增量补录 timeout_count timeout_counts: dict[str, int] = {} for name in handler_names: timeout_counts[name] = timeout_counts.get(name, 0) + 1 for name, count in timeout_counts.items(): - for _ in range(count): - await self._repos.run_log_daily.upsert_daily_stat( - today, name, "timeout", commit=False - ) + await self._repos.run_log_daily.upsert_daily_stat( + today, name, "timeout", increment=count, commit=False + ) reclaimed_count = len(handler_names) if reclaimed_count > 0: @@ -1535,11 +1728,16 @@ class SchedulerService(SchedulerServicePort): operation="restore", operator=input_dto.updated_by, ) - await self._repos.task.restore_by_id( + restored_rows = await self._repos.task.restore_by_id( deleted_task.id, updated_by=input_dto.updated_by, commit=False, ) + if restored_rows == 0: + raise SchedulerConflictError( + f"任务 {input_dto.task_id!r} 恢复失败：可能已被其他请求恢复或 task_id 冲突", + details={"task_id": input_dto.task_id}, + ) if was_active: await self._repos.task.pause( input_dto.task_id,