From f2918b928585de5432ab6327b3fd0725bd52d919 Mon Sep 17 00:00:00 2001 From: JontyLee <31982854+JontyLee@users.noreply.github.com> Date: Mon, 26 Oct 2026 10:34:42 +0800 Subject: [PATCH] =?UTF-8?q?fix(scheduler):=20=E8=AE=A9=E5=A3=B0=E6=98=8E?= =?UTF-8?q?=E4=B8=BA=E4=B8=8D=E9=99=90=E6=97=B6=E7=9A=84=E5=8D=8F=E7=A8=8B?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E9=A2=84=E7=AE=97=E7=9C=9F=E6=AD=A3=E7=94=9F?= =?UTF-8?q?=E6=95=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit JobSpec.timeout_seconds 此前无法表达“显式不限时”：所有未声明预算的协程任务 一律补齐 SCHEDULER_JOB_TIMEOUT，订阅搜索队列因此被 3600 秒总时长判死，而它的 托管协程消费者无法靠取消收尾，只会把正常长轮次记成失败。现在收敛为三态语义 （正数为上限、0 为显式不限时、缺省继承默认），并修正 asyncio 超时异常 str() 为空导致的日志、系统通知与 SystemError 事件文本缺失。 --- app/application/scheduling.py | 3 +- app/scheduler/catalog.py | 2 + app/scheduler/execution.py | 32 ++++++-- docs/agent-execution.md | 3 +- tests/test_scheduler_contracts.py | 122 +++++++++++++++++++++++++++++- 5 files changed, 151 insertions(+), 11 deletions(-) diff --git a/app/application/scheduling.py b/app/application/scheduling.py index 3497625ad..3b38a60ad 100644 --- a/app/application/scheduling.py +++ b/app/application/scheduling.py @@ -54,7 +54,8 @@ class JobSpec: :param func: 调度触发时调用的业务函数 :param owner: 任务生命周期所属的业务组件标识 :param overlap: 前一次仍在执行时的重入策略，默认跳过新触发 - :param timeout_seconds: 声明的执行超时秒数，None 表示不声明超时 + :param timeout_seconds: 执行超时预算：正数为总时长上限，0 表示显式不限时， + None 表示未声明并由全局默认补齐 :param manual: 是否允许通过管理界面手动触发 :param recovery: 进程重启后的恢复策略，默认等待下一次计划触发 :param kwargs: 每次调用业务函数时使用的关键字参数 diff --git a/app/scheduler/catalog.py b/app/scheduler/catalog.py index 91a525ade..878ea0a72 100644 --- a/app/scheduler/catalog.py +++ b/app/scheduler/catalog.py @@ -69,6 +69,8 @@ def _subscription_search_job_specs( "恢复订阅搜索队列", services.resume_subscribe_search, "subscription", + # 单轮耗时由队列自身限额决定，取消又杀不掉同步消费者，因此不叠加总时长预算。 + timeout_seconds=0, recovery=JobRecoveryPolicy.DURABLE_QUEUE, ), ) diff --git a/app/scheduler/execution.py b/app/scheduler/execution.py index 62092f34f..413a6c7c9 100644 --- a/app/scheduler/execution.py +++ b/app/scheduler/execution.py @@ -50,10 +50,16 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase): def _assign_job_generation(self, job_id: str, job: dict[str, Any]) -> None: """ - 把注册 generation 写入可变运行时状态。 + 把注册 generation 写入可变运行时状态，并补齐协程任务的执行预算。 + + 声明为 0 的任务保留不限时：托管协程消费者无法靠取消真正停止，给它总时长上限 + 只会把正常长轮次记成失败，真实收尾仍然要等消费者自己结束。 """ self._registry.assign_generation(job_id, job) - if inspect.iscoroutinefunction(job.get("func")) and job.get("timeout_seconds") is None: + declared = job.get("timeout_seconds") + if declared == 0: + job["timeout_seconds"] = None + elif inspect.iscoroutinefunction(job.get("func")) and declared is None: timeout = int(get_runtime_setting("SCHEDULER_JOB_TIMEOUT", 3600)) job["timeout_seconds"] = timeout if timeout > 0 else None @@ -129,14 +135,26 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase): return job @staticmethod - def _handle_job_error(job_id: str, job: dict[str, Any], error: Exception) -> None: + def _handle_job_error( + job_id: str, + job: dict[str, Any], + error: Exception, + detail: Optional[str] = None, + ) -> None: """ 记录定时任务执行异常并发送系统错误事件。 + + :param job_id: 调度任务标识 + :param job: 任务运行时状态 + :param error: 捕获到的异常，用于取异常类型与堆栈 + :param detail: 面向日志、通知和事件的错误文本；省略时取异常文本， + 用于补足 asyncio 超时这类 str() 为空的标准异常 """ - logger.error(f"定时任务 {job.get('name')} 执行失败：{str(error)} - {traceback.format_exc()}") + message = detail or str(error) + logger.error(f"定时任务 {job.get('name')} 执行失败：{message} - {traceback.format_exc()}") _message_helper_factory().put( title=f"{job.get('name')} 执行失败", - message=str(error), + message=message, role="system", level="error", ) @@ -146,7 +164,7 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase): "type": "scheduler", "scheduler_id": job_id, "scheduler_name": job.get("name"), - "error": str(error), + "error": message, "traceback": traceback.format_exc(), }, ) @@ -196,7 +214,7 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase): 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) + self._handle_job_error(job_id=job_id, job=job, error=err, detail=error) except asyncio.CancelledError: success = False error = "任务已取消" diff --git a/docs/agent-execution.md b/docs/agent-execution.md index 2edeae7e9..0d006fcb7 100644 --- a/docs/agent-execution.md +++ b/docs/agent-execution.md @@ -52,7 +52,8 @@ APScheduler 3 的标准线程池没有任意数值 `priority` 的调度能力。 - `AGENT_LLM_TIMEOUT`：单次根 Agent 模型请求总预算，默认 180 秒，覆盖流式与非流式生成 以及该请求的供应商重试；`0` 表示不限时。工具执行时间由原 `LLM_TOOL_TIMEOUT` 管理。 - `SCHEDULER_JOB_TIMEOUT`：协程调度任务的默认执行预算，默认 3600 秒，等待避让不计入执行时间； - `0` 表示不限时。显式 `JobSpec.timeout_seconds` 优先于此默认值。 + `0` 表示不限时。显式 `JobSpec.timeout_seconds` 优先于此默认值：正数是总时长上限，`0` + 表示该任务显式不限时，缺省才继承此默认值。 超时通过协程取消收口，不能杀死同步线程或撤销已经发生的外部副作用。 数据库和线程调用仍等待实际操作收尾后释放容量，因而收尾时间可能超过超时预算。 diff --git a/tests/test_scheduler_contracts.py b/tests/test_scheduler_contracts.py index 527b4f239..87934511c 100644 --- a/tests/test_scheduler_contracts.py +++ b/tests/test_scheduler_contracts.py @@ -1,9 +1,11 @@ """ -Scheduler 声明、catalog 与 execution state 测试。 +Scheduler 声明、catalog、execution state 与协程执行预算语义测试。 """ import asyncio -from unittest.mock import MagicMock +import threading +from typing import Any +from unittest.mock import MagicMock, patch import pytest @@ -14,6 +16,9 @@ from app.application.scheduling import ( JobRecoveryPolicy, JobSpec, ) +from app.scheduler import execution as scheduler_execution +from app.scheduler.facade import Scheduler +from app.scheduler.registry import ExecutionRegistry def test_job_spec_exports_complete_legacy_state() -> None: @@ -99,3 +104,116 @@ def test_manual_job_declares_restart_policy() -> None: assert state["manual"] is True assert state["recovery"] == JobRecoveryPolicy.MANUAL_ONLY.value + + +async def _coro_job() -> None: + """ + 协程任务样本，只提供可内省的异步签名。 + """ + return None + + +def _resolved_timeout(spec: JobSpec, default_seconds: int) -> Any: + """ + 用最小 Scheduler owner 走一遍预算补齐路径，返回运行状态里的最终超时值。 + + :param spec: 待解析的任务声明 + :param default_seconds: 模拟的全局协程默认预算 + :return: 补齐后的 `timeout_seconds` + """ + scheduler = object.__new__(Scheduler) + scheduler._lock = threading.RLock() + scheduler._registry = ExecutionRegistry(scheduler._lock) + state = spec.to_runtime_state() + with patch.object( + scheduler_execution, + "get_runtime_setting", + return_value=default_seconds, + ): + scheduler._assign_job_generation(spec.job_id, state) + return state["timeout_seconds"] + + +def test_declared_zero_timeout_keeps_job_unlimited() -> None: + """ + 显式声明 0 秒的协程任务不得继承全局预算：托管消费者取消不掉，套上总时长上限只会 + 把正常长轮次记成超时失败。 + """ + spec = JobSpec( + "queue", + "队列恢复", + _coro_job, + "subscription", + timeout_seconds=0, + recovery=JobRecoveryPolicy.DURABLE_QUEUE, + ) + + assert _resolved_timeout(spec, default_seconds=3600) is None + + +def test_undeclared_coroutine_timeout_inherits_global_default() -> None: + """ + 未声明预算的协程任务继承全局默认，同步任务保持不限时。 + """ + assert _resolved_timeout( + JobSpec("coro", "协程任务", _coro_job, "runtime"), + default_seconds=120, + ) == 120 + assert _resolved_timeout( + JobSpec("sync", "同步任务", MagicMock(), "runtime"), + default_seconds=120, + ) is None + + +def test_positive_declared_timeout_wins_over_global_default() -> None: + """ + 正数声明是本次执行的总时长上限，不被全局默认覆盖。 + """ + assert _resolved_timeout( + JobSpec("bounded", "有界任务", _coro_job, "runtime", timeout_seconds=30), + default_seconds=3600, + ) == 30 + + +def test_timeout_error_keeps_descriptive_message() -> None: + """ + `asyncio` 超时异常的文本为空，日志、系统通知和事件都必须带上可读的超时说明。 + """ + job = {"name": "恢复订阅搜索队列", "timeout_seconds": 3600} + helper = MagicMock() + with ( + patch.object(scheduler_execution.logger, "error") as error_log, + patch.object(scheduler_execution, "_message_helper_factory", new=lambda: helper), + patch.object(scheduler_execution.eventmanager, "send_event") as send_event, + ): + Scheduler._handle_job_error( + job_id="subscribe_search_queue", + job=job, + error=asyncio.TimeoutError(), + detail="任务执行超时（3600 秒）", + ) + + expected = "任务执行超时（3600 秒）" + assert expected in error_log.call_args.args[0] + assert helper.put.call_args.kwargs["message"] == expected + assert send_event.call_args.args[1]["error"] == expected + + +def test_job_error_falls_back_to_exception_text() -> None: + """ + 未提供说明时沿用异常文本，普通异常的错误链路不变。 + """ + helper = MagicMock() + with ( + patch.object(scheduler_execution.logger, "error") as error_log, + patch.object(scheduler_execution, "_message_helper_factory", new=lambda: helper), + patch.object(scheduler_execution.eventmanager, "send_event"), + ): + Scheduler._handle_job_error( + job_id="job", + job={"name": "普通任务"}, + error=RuntimeError("站点无响应"), + ) + + assert "站点无响应" in error_log.call_args.args[0] + assert helper.put.call_args.kwargs["message"] == "站点无响应"