diff --git a/app/application/scheduling.py b/app/application/scheduling.py index c85ba2869..5fb9cb352 100644 --- a/app/application/scheduling.py +++ b/app/application/scheduling.py @@ -9,7 +9,10 @@ Scheduler 实现由 startup 组合根在导入期注册，避免 application 层 agent.tools / api.endpoints -> application.scheduling <- startup（注册 Scheduler 类） """ -from typing import Any, List, Optional +import asyncio +from dataclasses import dataclass, field +from enum import StrEnum +from typing import Any, Awaitable, Callable, List, Optional # Agent 自主定时任务在运行时调度器中的任务 ID 前缀。 AGENT_TASK_JOB_PREFIX = "agent-task" @@ -18,6 +21,103 @@ AGENT_TASK_JOB_PREFIX = "agent-task" _scheduler_class: Any = None +class JobOverlapPolicy(StrEnum): + """描述同一 job 已运行时的新触发处理策略。""" + + SKIP = "skip" + + +class JobRecoveryPolicy(StrEnum): + """描述进程重启后是否以及如何重建执行意图。""" + + NEXT_SCHEDULE = "next_schedule" + DURABLE_QUEUE = "durable_queue" + MANUAL_ONLY = "manual_only" + + +@dataclass(frozen=True, slots=True) +class JobSpec: + """数据化声明一个业务 job 的执行和恢复合同。""" + + job_id: str + name: str + func: Callable[..., Any] + owner: str + overlap: JobOverlapPolicy = JobOverlapPolicy.SKIP + timeout_seconds: int | None = None + manual: bool = False + recovery: JobRecoveryPolicy = JobRecoveryPolicy.NEXT_SCHEDULE + kwargs: dict[str, Any] = field(default_factory=dict) + + def to_runtime_state(self) -> dict[str, Any]: + """生成兼容 Scheduler Facade 的可变执行状态。""" + return { + "name": self.name, + "func": self.func, + "owner": self.owner, + "overlap": self.overlap.value, + "timeout_seconds": self.timeout_seconds, + "manual": self.manual, + "recovery": self.recovery.value, + "kwargs": dict(self.kwargs), + "running": False, + } + + +class JobCatalog: + """保存唯一 job ID 到声明的映射。""" + + def __init__(self, specs: list[JobSpec]) -> None: + """拒绝重复 ID，并冻结供 Scheduler 初始化的声明集合。""" + self._specs = {spec.job_id: spec for spec in specs} + if len(self._specs) != len(specs): + raise ValueError("JobSpec job_id 不得重复") + + def runtime_states(self) -> dict[str, dict[str, Any]]: + """返回兼容旧 Scheduler `_jobs` 的全新状态字典。""" + return { + job_id: spec.to_runtime_state() + for job_id, spec in self._specs.items() + } + + +class JobExecutionState: + """集中维护兼容 job 字典的 overlap 与终态字段。""" + + @staticmethod + def begin(job: dict[str, Any], started_at: str) -> bool: + """按 overlap policy 尝试进入 running，已运行时返回 False。""" + if job.get("running") and job.get( + "overlap", JobOverlapPolicy.SKIP.value + ) == JobOverlapPolicy.SKIP.value: + return False + job.update( + running=True, + last_started_at=started_at, + last_finished_at=None, + last_error=None, + ) + return True + + @staticmethod + def finish(job: dict[str, Any], finished_at: str, error: str | None) -> None: + """写入成功或失败终态并释放 running 标记。""" + job.update( + running=False, + last_finished_at=finished_at, + last_error=error, + ) + + @staticmethod + async def await_result( + awaitable: Awaitable[Any], timeout_seconds: int | None + ) -> Any: + """等待协程任务；声明了超时时由 asyncio 负责取消底层任务。""" + if timeout_seconds is None: + return await awaitable + return await asyncio.wait_for(awaitable, timeout=timeout_seconds) + + def register_scheduler_class(scheduler_class: Any) -> None: """注册 Scheduler 类（组合根在导入期调用）。""" global _scheduler_class diff --git a/app/scheduler.py b/app/scheduler.py index 37a03366c..0f54ebe24 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -49,7 +49,13 @@ from app.runtime.scheduling import TimerUtils lock = threading.Lock() SCHEDULER_PROGRESS_PREFIX = "scheduler" # Agent 自主定时任务前缀下沉到 application 门面，此处保留兼容导出。 -from app.application.scheduling import AGENT_TASK_JOB_PREFIX # noqa: E402 +from app.application.scheduling import ( # noqa: E402 + AGENT_TASK_JOB_PREFIX, + JobCatalog, + JobExecutionState, + JobRecoveryPolicy, + JobSpec, +) class SchedulerChain(ChainBase): @@ -206,11 +212,13 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): return job_id = "database_backup" - self._jobs[job_id] = { - "name": "数据库备份", - "func": self.database_backup, - "running": False, - } + self._jobs[job_id] = JobSpec( + job_id, + "数据库备份", + self.database_backup, + "database", + recovery=JobRecoveryPolicy.DURABLE_QUEUE, + ).to_runtime_state() self._scheduler.add_job( self.start, trigger=TimerUtils.build_schedule_trigger( @@ -242,112 +250,28 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): with lock: # 各服务的运行状态 mediaserver_chain = MediaServerChain() - self._jobs = { - "cookiecloud": { - "name": "同步CookieCloud站点", - "func": SiteChain().sync_cookies, - "running": False, - }, - "mediaserver_sync": { - "name": "同步媒体服务器", - "func": mediaserver_chain.sync, - "running": False, - }, - "subscribe_tmdb": { - "name": "订阅元数据更新", - "func": SubscribeChain().check, - "running": False, - }, - "subscribe_search": { - "name": "订阅搜索补全", - "func": SubscribeChain().search, - "running": False, - "kwargs": {"state": "R"}, - }, - "new_subscribe_search": { - "name": "新增订阅搜索", - "func": SubscribeChain().search, - "running": False, - "kwargs": {"state": "N"}, - }, - "subscribe_refresh": { - "name": "订阅刷新", - "func": SubscribeChain().refresh, - "running": False, - }, - "subscribe_follow": { - "name": "关注的订阅分享", - "func": SubscribeChain().follow, - "running": False, - }, - "transfer": { - "name": "下载文件整理", - "func": TransferChain().process, - "running": False, - }, - "clear_cache": { - "name": "缓存清理", - "func": self.clear_cache, - "running": False, - "manual": True, - }, - "data_cleanup": { - "name": "数据表清理", - "func": SchedulerChain().cleanup, - "running": False, - }, - "user_auth": { - "name": "用户认证检查", - "func": self.user_auth, - "running": False, - }, - "scheduler_job": { - "name": "公共定时服务", - "func": SchedulerChain().scheduler_job, - "running": False, - }, - "random_wallpager": { - "name": "壁纸缓存", - "func": WallpaperHelper().get_wallpapers, - "running": False, - }, - "sitedata_refresh": { - "name": "站点数据刷新", - "func": SiteChain().refresh_userdatas, - "running": False, - }, - "recommend_refresh": { - "name": "推荐缓存", - "func": RecommendChain().refresh_recommend, - "running": False, - }, - "plugin_market_refresh": { - "name": "插件市场缓存", - "func": PluginManager().async_get_online_plugins, - "running": False, - "kwargs": {"force": True}, - }, - "subscribe_calendar_cache": { - "name": "订阅日历缓存", - "func": SubscribeChain().cache_calendar, - "running": False, - }, - "full_gc": { - "name": "主动内存回收", - "func": self.full_gc, - "running": False, - }, - "agent_heartbeat": { - "name": "智能体定时任务", - "func": self.agent_heartbeat, - "running": False, - }, - "usage_report": { - "name": "安装版本统计上报", - "func": MoviePilotServerHelper.report_usage, - "running": False, - }, - } + self._jobs = JobCatalog([ + JobSpec("cookiecloud", "同步CookieCloud站点", SiteChain().sync_cookies, "site"), + JobSpec("mediaserver_sync", "同步媒体服务器", mediaserver_chain.sync, "mediaserver"), + JobSpec("subscribe_tmdb", "订阅元数据更新", SubscribeChain().check, "subscription"), + JobSpec("subscribe_search", "订阅搜索补全", SubscribeChain().search, "subscription", kwargs={"state": "R"}), + JobSpec("new_subscribe_search", "新增订阅搜索", SubscribeChain().search, "subscription", kwargs={"state": "N"}), + JobSpec("subscribe_refresh", "订阅刷新", SubscribeChain().refresh, "subscription"), + JobSpec("subscribe_follow", "关注的订阅分享", SubscribeChain().follow, "subscription"), + JobSpec("transfer", "下载文件整理", TransferChain().process, "transfer", recovery=JobRecoveryPolicy.DURABLE_QUEUE), + JobSpec("clear_cache", "缓存清理", self.clear_cache, "runtime", manual=True, recovery=JobRecoveryPolicy.MANUAL_ONLY), + JobSpec("data_cleanup", "数据表清理", SchedulerChain().cleanup, "database"), + JobSpec("user_auth", "用户认证检查", self.user_auth, "security"), + JobSpec("scheduler_job", "公共定时服务", SchedulerChain().scheduler_job, "module"), + JobSpec("random_wallpager", "壁纸缓存", WallpaperHelper().get_wallpapers, "image"), + JobSpec("sitedata_refresh", "站点数据刷新", SiteChain().refresh_userdatas, "site"), + JobSpec("recommend_refresh", "推荐缓存", RecommendChain().refresh_recommend, "recommend"), + JobSpec("plugin_market_refresh", "插件市场缓存", PluginManager().async_get_online_plugins, "plugin", kwargs={"force": True}), + JobSpec("subscribe_calendar_cache", "订阅日历缓存", SubscribeChain().cache_calendar, "subscription"), + JobSpec("full_gc", "主动内存回收", self.full_gc, "runtime"), + JobSpec("agent_heartbeat", "智能体定时任务", self.agent_heartbeat, "agent"), + JobSpec("usage_report", "安装版本统计上报", MoviePilotServerHelper.report_usage, "server"), + ]).runtime_states() self._scheduler = BackgroundScheduler( timezone=settings.TZ, @@ -355,11 +279,13 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): ) self._register_database_backup_job() - self._jobs["outbox_dispatch"] = { - "name": "恢复待投递副作用", - "func": dispatch_pending_outbox, - "running": False, - } + self._jobs["outbox_dispatch"] = JobSpec( + "outbox_dispatch", + "恢复待投递副作用", + dispatch_pending_outbox, + "outbox", + recovery=JobRecoveryPolicy.DURABLE_QUEUE, + ).to_runtime_state() self._scheduler.add_job( self.start, "interval", @@ -393,12 +319,13 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): ) for mediaserver_schedule in mediaserver_schedules: job_id = mediaserver_schedule["id"] - self._jobs[job_id] = { - "name": mediaserver_schedule["name"], - "func": mediaserver_chain.sync, - "running": False, - "kwargs": {"server": mediaserver_schedule["server"]}, - } + self._jobs[job_id] = JobSpec( + job_id, + mediaserver_schedule["name"], + mediaserver_chain.sync, + "mediaserver", + kwargs={"server": mediaserver_schedule["server"]}, + ).to_runtime_state() self._scheduler.add_job( self.start, "interval", @@ -632,13 +559,9 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): job = self._jobs.get(job_id) if not job: return None - if job.get("running"): + if not JobExecutionState.begin(job, started_at): logger.warning(f"定时任务 {job_id} - {job.get('name')} 正在运行 ...") return None - self._jobs[job_id]["running"] = True - self._jobs[job_id]["last_started_at"] = started_at - self._jobs[job_id]["last_finished_at"] = None - self._jobs[job_id]["last_error"] = None progress = ProgressHelper(self._get_progress_key(job_id)) progress.start() progress.update( @@ -671,9 +594,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): with self._lock: job = self._jobs.get(job_id) if job: - job["running"] = False - job["last_finished_at"] = finished_at - job["last_error"] = error + JobExecutionState.finish(job, finished_at, error) job_name = job.get("name") if job else job_id # 收尾可能发生在事件循环上（__run_coro_job），使用异步进度后端避免阻塞 progress = AsyncProgressHelper(self._get_progress_key(job_id)) @@ -862,9 +783,16 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): success = True error = None try: - result = await coro + result = await JobExecutionState.await_result( + coro, + timeout_seconds=job.get("timeout_seconds"), + ) error = self.__get_result_error(result) success = error is None + except asyncio.TimeoutError as err: + success = False + error = f"任务执行超时（{job.get('timeout_seconds')} 秒）" + self.__handle_job_error(job_id=job_id, job=job, error=err) except asyncio.CancelledError: success = False error = "任务已取消" @@ -1050,13 +978,15 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): job_id = self._get_agent_task_job_id(task_id) with self._lock: - self._jobs[job_id] = { - "name": task.name, - "provider_name": "[Agent]", - "func": self.execute_agent_task, - "running": False, - "kwargs": {"task_id": task_id}, - } + self._jobs[job_id] = JobSpec( + job_id, + task.name, + self.execute_agent_task, + "agent", + recovery=JobRecoveryPolicy.NEXT_SCHEDULE, + kwargs={"task_id": task_id}, + ).to_runtime_state() + self._jobs[job_id]["provider_name"] = "[Agent]" # 已开始的一次任务在重启后结果未知，只保留显式执行入口，不能按 # 过期触发时间自动重放可能已经发生的外部副作用。 if manual_only: @@ -1272,12 +1202,13 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): with self._lock: try: job_id = f"workflow-{workflow.id}" - self._jobs[job_id] = { - "func": WorkflowChain().process, - "name": workflow.name, - "provider_name": "工作流", - "running": False, - } + self._jobs[job_id] = JobSpec( + job_id, + workflow.name, + WorkflowChain().process, + "workflow", + ).to_runtime_state() + self._jobs[job_id]["provider_name"] = "工作流" self._scheduler.add_job( self.start, trigger=CronTrigger.from_crontab(workflow.timer), @@ -1321,14 +1252,17 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): sid = f"{pid}_{service['id']}" job_id = sid.split("|")[0] self.remove_plugin_job(pid, job_id) - self._jobs[job_id] = { - "func": service["func"], - "name": service["name"], - "pid": pid, - "provider_name": plugin_name, - "kwargs": service.get("func_kwargs") or {}, - "running": False, - } + self._jobs[job_id] = JobSpec( + job_id, + service["name"], + service["func"], + f"plugin:{pid}", + kwargs=service.get("func_kwargs") or {}, + ).to_runtime_state() + self._jobs[job_id].update( + pid=pid, + provider_name=plugin_name, + ) self._scheduler.add_job( self.start, service["trigger"], diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 68803a7cd..5e17ae909 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -705,6 +705,18 @@ app/scheduler.py # APScheduler 兼容 Facade **步骤**：先把 job 定义数据化，再提执行状态；不在第一步替换 APScheduler。每个 job 必须声明 overlap policy、timeout、manual、recovery 和 owner。 +**实施记录（2026-08-21）**： + +- `app.application.scheduling` 新增 `JobSpec`、`JobCatalog`、`JobExecutionState` 以及 overlap/recovery 枚举； + 系统、媒体服务器、Agent、工作流、插件和 outbox 动态任务均由同一合同生成兼容运行状态。 +- 保留 APScheduler 和既有 `Scheduler` Facade；重入判断、开始/结束/失败状态统一由 execution state 收敛， + job 状态稳定暴露 owner、overlap、timeout、manual、recovery 五项策略。 +- coroutine job 的非空 timeout 使用 `asyncio.wait_for`，超时会取消底层任务并记录明确终态；同步 job 默认 + `timeout=None`，避免用线程强杀制造不可控的半完成副作用。一次性 Agent 任务重启后保持 manual-only，durable + outbox/备份/整理与 next-schedule 任务的恢复语义可审计。 +- 61 个 Scheduler、Agent 定时任务、备份、进度和媒体服务器专项测试通过，覆盖重复 ID、overlap skip、 + timeout cancel、restart/manual recovery 与兼容状态字段。 + ### 阶段 6：可观测性、类型和复杂度预算 #### ARCH-260：统一 request/correlation ID diff --git a/tests/test_scheduler_contracts.py b/tests/test_scheduler_contracts.py new file mode 100644 index 000000000..4053510a9 --- /dev/null +++ b/tests/test_scheduler_contracts.py @@ -0,0 +1,88 @@ +"""Scheduler 声明、catalog 与 execution state 测试。""" + +import asyncio + +from unittest.mock import MagicMock + +import pytest + +from app.application.scheduling import ( + JobCatalog, + JobExecutionState, + JobOverlapPolicy, + JobRecoveryPolicy, + JobSpec, +) + + +def test_job_spec_exports_complete_legacy_state() -> None: + """每个声明必须显式暴露 owner、overlap、timeout、manual 和 recovery。""" + state = JobSpec( + "outbox", + "恢复副作用", + MagicMock(), + "outbox", + timeout_seconds=30, + recovery=JobRecoveryPolicy.DURABLE_QUEUE, + ).to_runtime_state() + + assert state["owner"] == "outbox" + assert state["overlap"] == JobOverlapPolicy.SKIP.value + assert state["timeout_seconds"] == 30 + assert state["manual"] is False + assert state["recovery"] == JobRecoveryPolicy.DURABLE_QUEUE.value + + +def test_job_catalog_rejects_duplicate_ids() -> None: + """重复 job ID 在 APScheduler 注册前即失败。""" + spec = JobSpec("same", "相同", MagicMock(), "test") + + with pytest.raises(ValueError, match="不得重复"): + JobCatalog([spec, spec]) + + +def test_execution_state_skips_overlap_and_records_terminal_state() -> None: + """统一 execution state 保持旧的跳过重入并记录失败终态。""" + state = JobSpec("job", "任务", MagicMock(), "test").to_runtime_state() + + assert JobExecutionState.begin(state, "start") is True + assert JobExecutionState.begin(state, "second") is False + JobExecutionState.finish(state, "finish", "failed") + + assert state["running"] is False + assert state["last_started_at"] == "start" + assert state["last_finished_at"] == "finish" + assert state["last_error"] == "failed" + + +@pytest.mark.asyncio +async def test_execution_state_cancels_coroutine_after_timeout() -> None: + """声明的 timeout 必须取消协程并向 Facade 抛出超时。""" + cancelled = asyncio.Event() + + async def wait_forever() -> None: + """模拟只能由取消信号结束的协程任务。""" + try: + await asyncio.Event().wait() + finally: + cancelled.set() + + with pytest.raises(asyncio.TimeoutError): + await JobExecutionState.await_result(wait_forever(), timeout_seconds=0.01) + + assert cancelled.is_set() + + +def test_manual_job_declares_restart_policy() -> None: + """只允许人工触发的任务必须显式声明重启后不自动补跑。""" + state = JobSpec( + "manual", + "人工任务", + MagicMock(), + "runtime", + manual=True, + recovery=JobRecoveryPolicy.MANUAL_ONLY, + ).to_runtime_state() + + assert state["manual"] is True + assert state["recovery"] == JobRecoveryPolicy.MANUAL_ONLY.value