From 3ebd4de993d1e50524df03a460565987d2c3cfef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=80=E7=89=87=E4=B8=80=E7=89=87?= <3037678834@qq.com> Date: Mon, 9 Nov 2026 20:25:32 +0800 Subject: [PATCH] fix(scheduler): isolate a malformed stored schedule from the poll batch (#7999) remain from its review: - `stored_next_run_at` caught the ValueErrors a dropped zone or a newer cron parser raises, but a stored `schedule_spec` that is JSON null, a list or a string fails on `.get` with AttributeError. The launch-failure handler recomputes the schedule, so that escaped `run_once`'s claimed-task loop and skipped the rest of the batch. Catch it too; the task is paused with the reason like the other unusable schedules. - `run_once` dispatched its claimed batch with a bare loop, so any per-record raise in `dispatch_task` re-aborted the batch. Isolate each task's dispatch and keep claim-level store failures loud above the loop. Tests: `test_scheduler_poll_batch_isolation.py` (non-object specs, end-to-end pause, per-task isolation). Red on `91f20a7`, green at head. Refs #7998 --- CHANGELOG.md | 12 ++ backend/app/scheduler/service.py | 12 +- .../harness/deerflow/persistence/AGENTS.md | 2 +- .../harness/deerflow/scheduler/schedules.py | 12 +- .../test_scheduler_poll_batch_isolation.py | 128 ++++++++++++++++++ 5 files changed, 159 insertions(+), 7 deletions(-) create mode 100644 backend/tests/test_scheduler_poll_batch_isolation.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 11bc8cca30..14a99a842a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -70,6 +70,16 @@ This section accumulates work toward the **2.3.0** milestone preventing `OSError` failures on Windows hosts while keeping existing read and delete operations permissive. +- **scheduler:** A stored `schedule_spec` that is JSON null, a list or a string + no longer aborts the scheduler poll. `stored_next_run_at` caught the value + errors from a dropped zone or a newer cron parser, but a non-object spec + fails on `.get`; the launch-failure handler recomputed it again, so the + `AttributeError` escaped `run_once`'s claimed-task loop. Such a task is now + paused with the reason, like the other unusable schedules. The poll loop also + isolates one task's dispatch failure from the rest of its claimed batch, so a + future per-record raise costs one tick instead of every other claimed task. + ([#7998]) + - **scheduler:** One scheduled task whose saved schedule can no longer be computed no longer stops every other task. When a stored timezone or cron expression stopped working (a zone missing from the time zone database @@ -16186,3 +16196,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7992]: https://github.com/bytedance/deer-flow/pull/7992 [#7993]: https://github.com/bytedance/deer-flow/pull/7993 [#8006]: https://github.com/bytedance/deer-flow/pull/8006 + +[#7998]: https://github.com/bytedance/deer-flow/issues/7998 diff --git a/backend/app/scheduler/service.py b/backend/app/scheduler/service.py index a2b4273283..53fab188cf 100644 --- a/backend/app/scheduler/service.py +++ b/backend/app/scheduler/service.py @@ -480,7 +480,17 @@ class ScheduledTaskService: limit=self._max_concurrent_runs, ) for task in claimed: - await self.dispatch_task(task, now=now, trigger="scheduled") + try: + await self.dispatch_task(task, now=now, trigger="scheduled") + except Exception: + # A per-record failure inside one task's dispatch must not abort + # the rest of this poll's claimed batch: every claimed task + # already holds a lease, and the next tick would skip them the + # same way. Claim-level store failures stay loud above the loop. + logger.exception( + "Scheduled task %s: dispatch failed; continuing with the claimed batch", + task.get("id"), + ) @staticmethod def _is_overlap_conflict(exc: Exception) -> bool: diff --git a/backend/packages/harness/deerflow/persistence/AGENTS.md b/backend/packages/harness/deerflow/persistence/AGENTS.md index d04ca87759..0478988e72 100644 --- a/backend/packages/harness/deerflow/persistence/AGENTS.md +++ b/backend/packages/harness/deerflow/persistence/AGENTS.md @@ -47,7 +47,7 @@ Scheduled-task lifecycle rules live in one place each. `ScheduledTaskRepository. The per-owner run cap `scheduler.max_concurrent_runs_per_user` is effective `min(per_user, max_concurrent_runs)` (0 = off) and is checked in `claim_queued_run` right after the global count, under the same budget lock. `list_queued_runs` ranks same-thread FIFO heads per owner (`ROW_NUMBER() OVER (PARTITION BY owner)`) and orders the batch by that rank, so the drain is owner-fair even with the cap off; owners at their cap drop out before the batch limit, and orphan rows (task deleted while queued) are never held back by the cap. Scheduled-task repository boundaries coerce serialized timestamps before SQL `DateTime` binding. -Recovery locks task/run pairs by task/run ID; restore `run_id`/`started_at`/errors before releasing claims. Bind takeover grants to candidate incarnation/revision; log unconfirmed publication without reauthorizing. Launch/failure/timeout writes lock parent first; queue timeout fails the occurrence and advances scheduled work. Runtime cursor writes use `stored_next_run_at`: a schedule that stops computing pauses its task, never raises. +Recovery locks task/run pairs by task/run ID; restore `run_id`/`started_at`/errors before releasing claims. Bind takeover grants to candidate incarnation/revision; log unconfirmed publication without reauthorizing. Launch/failure/timeout writes lock parent first; queue timeout fails the occurrence and advances scheduled work. `stored_next_run_at` never raises (non-object specs included) and pauses the task; `run_once` isolates per-task failures. Single-instance polling also performs a non-destructive terminal-run reconciliation: a `running` occurrence whose durable `runs` row is already terminal is finalized from that row's status, error and goal verdict. Live runs are never taken over or interrupted; multi-instance reconciliation already covers terminal runs through lease ownership. This repairs a completion hook that failed after the run committed its terminal status. diff --git a/backend/packages/harness/deerflow/scheduler/schedules.py b/backend/packages/harness/deerflow/scheduler/schedules.py index 9c2ed2de4b..0c624bbafb 100644 --- a/backend/packages/harness/deerflow/scheduler/schedules.py +++ b/backend/packages/harness/deerflow/scheduler/schedules.py @@ -123,14 +123,16 @@ def stored_next_run_at( """``next_run_at`` for an admitted schedule, or why it no longer computes. Admission validates every schedule, but a stored one can stop computing - later: a zone the installed tz database no longer has, or a cron expression - a newer croniter rejects. Runtime bookkeeping runs inside the shared - scheduler poll, so it must not raise for one task; callers take the task - off the schedule with :func:`pauses_unschedulable` instead. + later: a zone the installed tz database no longer has, a cron expression + a newer croniter rejects, or a ``schedule_spec`` an operator edit or an old + release left as JSON null, a list or a string. Runtime bookkeeping runs + inside the shared scheduler poll, so it must not raise for one task; + callers take the task off the schedule with :func:`pauses_unschedulable` + instead. """ try: return next_run_at(schedule_type, schedule_spec, timezone_name, now=now), None - except (ValueError, OverflowError) as exc: + except (ValueError, OverflowError, AttributeError) as exc: return None, f"{UNSCHEDULABLE_ERROR_PREFIX}{exc}" diff --git a/backend/tests/test_scheduler_poll_batch_isolation.py b/backend/tests/test_scheduler_poll_batch_isolation.py new file mode 100644 index 0000000000..3d5dae4131 --- /dev/null +++ b/backend/tests/test_scheduler_poll_batch_isolation.py @@ -0,0 +1,128 @@ +"""One unusable record must not abort the scheduler's claimed-task batch. + +``stored_next_run_at`` tolerates a zone the installed tz database dropped and a +cron a newer croniter rejects, but a stored ``schedule_spec`` that is JSON +null, a list or a string raises ``AttributeError`` from ``.get``. Because the +launch-failure handler recomputes the schedule again, that escape aborted the +rest of ``run_once``'s claimed batch every poll. The loop also relied on every +dispatch path staying raise-free; it now isolates a single task's failure. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest + +from app.scheduler.service import ScheduledTaskService +from deerflow.config.database_config import DatabaseConfig +from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config +from deerflow.persistence.scheduled_task_runs import ScheduledTaskRunRepository +from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository +from deerflow.scheduler.schedules import UNSCHEDULABLE_ERROR_PREFIX, next_run_at, stored_next_run_at + +pytestmark = pytest.mark.asyncio + +DUE = datetime(2026, 1, 1, tzinfo=UTC) + + +def _make_service(task_repo, run_repo, launched: list) -> ScheduledTaskService: + async def fake_launch(**kwargs): + launched.append(kwargs) + return {"run_id": f"run-{len(launched)}", "thread_id": kwargs["thread_id"]} + + return ScheduledTaskService( + task_repo=task_repo, + task_run_repo=run_repo, + launch_run=fake_launch, + poll_interval_seconds=5, + lease_seconds=120, + max_concurrent_runs=10, + ) + + +async def _seed_task(task_repo, task_id: str, *, schedule_spec) -> None: + await task_repo.create( + task_id=task_id, + user_id="user-1", + thread_id=None, + context_mode="fresh_thread_per_run", + assistant_id="lead_agent", + title=task_id, + prompt="do the thing", + schedule_type="cron", + schedule_spec=schedule_spec, + timezone="UTC", + next_run_at=DUE, + ) + + +@pytest.mark.parametrize("spec", [None, [], "*/5 * * * *", 5]) +async def test_stored_next_run_at_rejects_a_non_object_spec_without_raising(spec): + # The raw helper still raises: only runtime bookkeeping tolerates it. + with pytest.raises(AttributeError): + next_run_at("cron", spec, "UTC", now=datetime.now(UTC)) + + next_at, reason = stored_next_run_at("cron", spec, "UTC", now=datetime.now(UTC)) + + assert next_at is None + assert reason is not None + assert reason.startswith(UNSCHEDULABLE_ERROR_PREFIX) + + +async def test_non_object_stored_spec_is_paused_instead_of_aborting_the_batch(tmp_path): + await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path))) + try: + sf = get_session_factory() + task_repo = ScheduledTaskRepository(sf) + run_repo = ScheduledTaskRunRepository(sf) + launched: list = [] + service = _make_service(task_repo, run_repo, launched) + + await _seed_task(task_repo, "poisoned-spec", schedule_spec=[]) + await _seed_task(task_repo, "healthy-cron", schedule_spec={"cron": "*/5 * * * *"}) + + # Before the fix the AttributeError escaped the claimed-task loop. + await service.run_once(now=datetime.now(UTC)) + + # The run itself still launches; only the schedule bookkeeping fails. + assert len(launched) == 2 + poisoned = await task_repo.get("poisoned-spec", user_id="user-1") + assert poisoned is not None + assert poisoned["next_run_at"] is None + assert poisoned["status"] == "paused" + assert UNSCHEDULABLE_ERROR_PREFIX in (poisoned["last_error"] or "") + healthy = await task_repo.get("healthy-cron", user_id="user-1") + assert healthy is not None + assert healthy["next_run_at"] is not None + finally: + await close_engine() + + +async def test_one_raising_dispatch_does_not_stop_the_rest_of_the_batch(tmp_path, monkeypatch): + await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path))) + try: + sf = get_session_factory() + task_repo = ScheduledTaskRepository(sf) + run_repo = ScheduledTaskRunRepository(sf) + launched: list = [] + service = _make_service(task_repo, run_repo, launched) + + await _seed_task(task_repo, "raising-task", schedule_spec={"cron": "*/5 * * * *"}) + await _seed_task(task_repo, "healthy-cron", schedule_spec={"cron": "*/5 * * * *"}) + + original_dispatch = service.dispatch_task + + async def flaky_dispatch(task, *, now, trigger): + if task["id"] == "raising-task": + raise RuntimeError("per-record failure") + return await original_dispatch(task, now=now, trigger=trigger) + + monkeypatch.setattr(service, "dispatch_task", flaky_dispatch) + + # The loop-level guard keeps the batch going and logs the failing task. + await service.run_once(now=datetime.now(UTC)) + + assert len(launched) == 1 + finally: + await close_engine()