feat: 新增回收站任务清理handler，完善任务统计与批量操作能力
本次提交包含多维度功能增强： 1. 新增TaskRecycleCleanupHandler，实现过期软删除任务物理清理能力 2. 扩展任务统计模型，新增死信任务计数并完善统计逻辑 3. 新增异常任务聚合查询接口，支持按死信/连续失败/长期未执行分类返回 4. 实现任务批量暂停/恢复/软删除操作 5. 扩展运行日志与任务查询过滤条件，新增handler_name维度 6. 优化健康检查逻辑，支持动态健康窗口并返回配置快照 7. 完善数据映射与DTO定义，补充缺失字段与类型支持
This commit is contained in:
parent
371c554aa7
commit
36d0add930
@ -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,
|
||||
)
|
||||
|
||||
@ -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,
|
||||
)
|
||||
|
||||
|
||||
@ -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。
|
||||
|
||||
@ -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)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
|
||||
@ -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: ...
|
||||
|
||||
|
||||
@ -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",
|
||||
]
|
||||
|
||||
@ -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,
|
||||
},
|
||||
)
|
||||
@ -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)
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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,
|
||||
|
||||
Loading…
Reference in New Issue
Block a user