diff --git a/CHANGELOG.md b/CHANGELOG.md index e659d4973d..bbaa5bc49d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -941,6 +941,14 @@ This release closes that milestone with **765 merged pull requests**. ### Fixed +- **scheduler:** Pausing a scheduled task no longer loses the pause when a + dispatch is in flight on SQLite. `release_dispatch_lease` guards on the lease + owner — which pausing clears — but read the row without taking SQLite's + writer, so a stale read passed the guard and wrote the task back to + `enabled` with `next_run_at` untouched, leaving the scheduler firing a task + the API had reported as paused. The read now takes the writer first, as every + other mutating path in that repository does. PostgreSQL was unaffected. + ([#5777]) - **uploads:** Deleting an uploaded document no longer deletes the converted Markdown beside it. Conversion names a companion after the document's stem and falls back to a `_N` suffix when that name is taken, so the `.md` next to @@ -4379,3 +4387,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#5673]: https://github.com/bytedance/deer-flow/pull/5673 [#5734]: https://github.com/bytedance/deer-flow/pull/5734 [#5776]: https://github.com/bytedance/deer-flow/pull/5776 +[#5777]: https://github.com/bytedance/deer-flow/pull/5777 + diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 04a11f86d5..a040253c6d 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -727,6 +727,11 @@ ### 修复 +- **调度器:** 在 SQLite 上,调度分发进行中暂停计划任务时不再丢失暂停状态。 + `release_dispatch_lease` 依据租约持有者做校验(暂停会清除该字段),但读取任务行时没有先获取 + SQLite 的写锁,因此过期的读取会通过校验,并把任务状态写回 `enabled` 且不改动 `next_run_at`, + 导致接口已回复“已暂停”的任务仍被继续触发。现在该读取会像该仓储中其他写入路径一样先获取写锁。 + PostgreSQL 不受影响。([#5777]) - **上传:** 删除已上传的文档时,不再连带删除其旁边转换生成的 Markdown。转换以文档主干名 命名配套文件,名称被占用时回退为 `_N` 后缀,因此文档旁的 `.md` 可能属于主干名相同的另一个 文档,或属于用户自己:上传 `a.docx` 与 `a.pdf` 会生成 `a.md` 与 `a_1.md`,删除 `a.pdf` @@ -3555,3 +3560,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子 [#5673]: https://github.com/bytedance/deer-flow/pull/5673 [#5734]: https://github.com/bytedance/deer-flow/pull/5734 [#5776]: https://github.com/bytedance/deer-flow/pull/5776 +[#5777]: https://github.com/bytedance/deer-flow/pull/5777 diff --git a/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py b/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py index 5039195936..6a3ba8acbf 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py +++ b/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py @@ -348,9 +348,16 @@ class ScheduledTaskRepository: expected_lease_owner: str | None, status: str, ) -> bool: - """Release the short due-task claim after its occurrence is queued.""" + """Release the short due-task claim after its occurrence is queued. + + The lease-owner guard is only as fresh as the row it reads, so the + read takes the writer first (``_lock_task``): on SQLite a plain + ``SELECT`` sees a pre-pause snapshot, and the unconditional + ``row.status = status`` below would then write the caller's + ``"enabled"`` over a pause that had already committed. + """ async with self._sf() as session: - row = await session.get(ScheduledTaskRow, task_id, with_for_update=True) + row = await self._lock_task(session, task_id) if row is None: return False if expected_lease_owner is not None and row.lease_owner != expected_lease_owner: diff --git a/backend/tests/test_scheduled_task_dispatch_race.py b/backend/tests/test_scheduled_task_dispatch_race.py index 0fe25f41cc..80b18f61f5 100644 --- a/backend/tests/test_scheduled_task_dispatch_race.py +++ b/backend/tests/test_scheduled_task_dispatch_race.py @@ -69,7 +69,31 @@ def _make_service(task_repo, run_repo, launched: list) -> ScheduledTaskService: ) -async def _seed_task(task_repo: ScheduledTaskRepository, task_id: str) -> dict: +class _PausingTaskRepo(ScheduledTaskRepository): + """Real repository that parks the first task-row lock it takes. + + The pause transaction stops after acquiring the row (holding SQLite's + writer) and before committing, so a dispatch release genuinely runs while + a pause is in flight — the window in which the release used to read a + stale ``lease_owner`` and write the task back to ``enabled``. + """ + + def __init__(self, session_factory, *, holding: asyncio.Event, resume: asyncio.Event) -> None: + super().__init__(session_factory) + self._holding = holding + self._resume = resume + self._parked = False + + async def _lock_task(self, session, task_id): # type: ignore[override] + row = await ScheduledTaskRepository._lock_task(session, task_id) + if not self._parked: + self._parked = True + self._holding.set() + await self._resume.wait() + return row + + +async def _seed_task(task_repo: ScheduledTaskRepository, task_id: str, *, next_run_at: datetime | None = None) -> dict: # fresh_thread_per_run: every dispatch gets a NEW thread_id, so #4003's # per-thread uq_runs_thread_active can never fire for two dispatches of the # same task — this is precisely the gap the per-task index closes. @@ -84,7 +108,7 @@ async def _seed_task(task_repo: ScheduledTaskRepository, task_id: str) -> dict: schedule_type="cron", schedule_spec={"cron": "*/5 * * * *"}, timezone="UTC", - next_run_at=None, + next_run_at=next_run_at, ) task = await task_repo.get(task_id, user_id="user-1") assert task is not None @@ -217,3 +241,40 @@ async def test_partial_unique_index_enforces_one_active_run_per_task(tmp_path): assert await run_repo.has_active_runs("t1") is True finally: await close_engine() + + +async def test_release_dispatch_lease_does_not_revive_a_paused_task(tmp_path): + """A pause landing mid-dispatch must survive the scheduler's lease release. + + ``release_dispatch_lease`` writes the status its caller asked for + (``"enabled"`` for scheduled dispatch) and only guards on the lease owner. + Reading that owner outside SQLite's writer lock let the release pass its + guard against a pre-pause snapshot and then overwrite the user's pause. + """ + await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path))) + try: + sf = get_session_factory() + assert sf is not None + holding, resume = asyncio.Event(), asyncio.Event() + task_repo = _PausingTaskRepo(sf, holding=holding, resume=resume) + now = datetime.now(UTC) + await _seed_task(task_repo, "task-pause-race", next_run_at=now) + claimed = await task_repo.claim_due_tasks(now=now, lease_owner="scheduler-A", lease_seconds=120, limit=10) + assert [row["id"] for row in claimed] == ["task-pause-race"] + + pause = asyncio.create_task(task_repo.pause_with_queue_cancellation("task-pause-race", user_id="user-1", error="paused by user", now=now)) + await asyncio.wait_for(holding.wait(), timeout=5) + + release = asyncio.create_task(task_repo.release_dispatch_lease("task-pause-race", expected_lease_owner="scheduler-A", status="enabled")) + await asyncio.sleep(0.2) # the release is in flight while the pause holds the row + resume.set() + + assert await asyncio.wait_for(pause, timeout=10) == "paused" + assert await asyncio.wait_for(release, timeout=10) is False + + task = await task_repo.get("task-pause-race", user_id="user-1") + assert task is not None + assert task["status"] == "paused", "the user's pause must not be reverted by the lease release" + assert task["lease_owner"] is None + finally: + await close_engine()