From 56aa71c5831ec9540a11612477f24b3cba653907 Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Mon, 5 Oct 2026 08:07:04 +0900 Subject: [PATCH] fix(runtime): keep renewing the lease while a staged terminal run finalizes (#6263) * fix(runtime): keep renewing the lease while a staged terminal run finalizes With an event store, run_agent stages the terminal status in memory and commits it to the run store only after the journal flush, delivery receipt, workspace scan and duration checkpoint. _renew_leases selected runs by local status, so it stopped renewing as soon as the status was staged. A finalization longer than lease_seconds + grace_seconds let a peer, or this worker's own reconciler, claim the still-active row and record the successful run as an orphan error. The worker now marks RunRecord.terminal_commit_pending until the deferred commit is attempted, and lease renewal treats it like the existing scheduled-goal barrier: keep renewing, release on a confirmed own commit, and fence when a peer claims the row. A fenced run stops renewing. * docs(changelog): reference #6263 for the staged terminal lease renewal fix --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 11 ++ CHANGELOG_zh.md | 8 + backend/docs/runtime-guidance-details.md | 3 +- .../harness/deerflow/runtime/AGENTS.md | 7 + .../harness/deerflow/runtime/runs/manager.py | 52 ++++--- .../harness/deerflow/runtime/runs/worker.py | 6 + .../tests/test_multi_worker_run_ownership.py | 137 ++++++++++++++++++ 7 files changed, 204 insertions(+), 20 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cc795935a7..a672add8ed 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -466,6 +466,16 @@ This release closes that milestone with **301 merged pull requests**. ### Fixed +- **runtime:** A multi-worker run that finished successfully is no longer + reclaimed as an orphan `error` while its worker is still finalizing. With an + event store, the worker records the terminal status in memory first and + writes it to the run store only after the journal flush, delivery receipt, + workspace scan and duration checkpoint. Lease renewal skipped runs whose + local status was already terminal, so a finalization longer than the lease + plus grace (about 30–40 seconds by default) let a peer, or the worker's own + reconciler, claim the still-active row. The heartbeat now keeps renewing + until that deferred write is attempted, and fences the run if a peer claims + it. Affects only `run_ownership.heartbeat_enabled` deployments. ([#6263]) - **channels:** Buzz now follows a thread without a fresh mention for authors bound with `/connect`. With `channel_connections.enabled`, the manager maps a bound author's threads only in the connection repository, but Buzz's @@ -7711,3 +7721,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6232]: https://github.com/bytedance/deer-flow/pull/6232 [#6238]: https://github.com/bytedance/deer-flow/pull/6238 [#6255]: https://github.com/bytedance/deer-flow/pull/6255 +[#6263]: https://github.com/bytedance/deer-flow/pull/6263 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 633a6f4350..8b5317bc2b 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -423,6 +423,13 @@ ### 修复 +- **运行时：** 多 worker 部署中已成功结束的运行，不会再在其 worker 仍在收尾时被 + 当作孤儿运行回收为 `error`。配置了事件存储时，worker 先在内存中记录终态，直到 + journal 刷新、交付回执、工作区扫描与时长 checkpoint 写入完成后才写入运行存储。 + 此前租约续期会跳过本地状态已是终态的运行，收尾一旦超过租约加宽限期（默认约 + 30–40 秒），其他 worker 或本 worker 自己的回收器就会接管这条仍处于活动状态的 + 记录。心跳现在会持续续期直到这次延迟写入被执行，若期间被其他 worker 接管则隔离 + 本地运行。仅影响启用 `run_ownership.heartbeat_enabled` 的部署。([#6263]) - **渠道：** 通过 `/connect` 绑定的 Buzz 作者在已参与的话题中回复时，无需再次提及 机器人。开启 `channel_connections.enabled` 后，管理器只在连接仓库中记录已绑定作者 的话题映射，而 Buzz 的话题跟随判断只读取 JSON 渠道存储，导致机器人正在回复的话题 @@ -6437,3 +6444,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#6232]: https://github.com/bytedance/deer-flow/pull/6232 [#6238]: https://github.com/bytedance/deer-flow/pull/6238 [#6255]: https://github.com/bytedance/deer-flow/pull/6255 +[#6263]: https://github.com/bytedance/deer-flow/pull/6263 diff --git a/backend/docs/runtime-guidance-details.md b/backend/docs/runtime-guidance-details.md index be563e6b6a..b311b69f21 100644 --- a/backend/docs/runtime-guidance-details.md +++ b/backend/docs/runtime-guidance-details.md @@ -55,7 +55,8 @@ cancellation while waiting for prior finalization still emit a zero-delivery receipt. The worker flushes ordinary journal events, idempotently persists the run-scoped receipt, and only then persists the staged terminal run status. A receipt failure is retried on a short bounded schedule while the owning worker -still knows the real outcome and holds the lease. Delivery candidates are every +still knows the real outcome and holds the lease; `terminal_commit_pending` keeps +heartbeat renewal running until the staged status is committed. Delivery candidates are every regular file created or modified under `/mnt/user-data/outputs`; internal process-feedback files are excluded (the scanner's `EXCLUDED_DIR_NAMES` plus the configured `tool_output.storage_subdir`), so a run that only externalized diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index e166e96022..10304622de 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -72,6 +72,13 @@ answers for paginated history. See `docs/skill-usage-ui.md`. **Run delivery receipts:** Journal artifact evidence and terminal status must finalize before an ordinary satisfied goal is cleared. That cleanup uses a durable checkpoint-write reservation; delivery failure retains the ordinary goal without another continuation. The scheduled-only exception is described below. Details: `backend/docs/runtime-guidance-details.md`. +**Deferred terminal commit:** With an event store, the worker stages its terminal +status locally and commits it only after finalization's receipt and duration +writes. `RunRecord.terminal_commit_pending` keeps `_renew_leases()` renewing that +still-active row until the commit is attempted; a renewal rejected by the worker's +own commit is confirmed by re-reading the row, while a peer claim fences the run. +Never select runs for renewal by local status alone. + **Deferred-tool promotion event deduplication** (`runtime/journal.py`): one `RunJournal` owns the lead graph's run-scoped atomic promotion claim. Parallel `tool_search` Sends read the same pre-step state, so state diffing alone can diff --git a/backend/packages/harness/deerflow/runtime/runs/manager.py b/backend/packages/harness/deerflow/runtime/runs/manager.py index 66221868b1..208b2bf9b4 100644 --- a/backend/packages/harness/deerflow/runtime/runs/manager.py +++ b/backend/packages/harness/deerflow/runtime/runs/manager.py @@ -230,6 +230,10 @@ class RunRecord: goal_verdict: dict[str, Any] | None = None # Process-local finalization barrier for a worker-installed scheduled goal. scheduled_goal_cleanup_pending: bool = False + # Process-local renewal barrier: the worker stages its terminal status in + # memory and commits it only at the end of finalization, so the durable row + # stays active and its lease must keep renewing until that commit. + terminal_commit_pending: bool = False idempotency_key: str | None = None # True only on the caller that recovered an existing idempotent admission; # that caller must not attach a second worker to the durable run. @@ -2111,6 +2115,8 @@ class RunManager: if record.ownership_lost: return True record.ownership_lost = True + # A fenced worker must not keep the peer-recoverable row alive. + record.terminal_commit_pending = False record.abort_event.set() record.status = RunStatus.error record.error = reason @@ -2198,13 +2204,18 @@ class RunManager: if cycle % 3 == 0: self._schedule_orphan_reconciliation() - async def _scheduled_terminal_write_confirmed(self, record: RunRecord) -> bool: + @staticmethod + def _awaits_terminal_commit(record: RunRecord) -> bool: + """Whether a run keeps its active durable row until a deferred terminal commit.""" + return record.scheduled_goal_cleanup_pending or record.terminal_commit_pending + + async def _terminal_write_confirmed(self, record: RunRecord) -> bool: """Recognize this worker's acknowledged outcome at the commit boundary. A peer takeover may preserve owner_worker_id, so compare every terminal outcome field rather than accepting any terminal row with that owner. """ - if not record.scheduled_goal_cleanup_pending or record.status in (RunStatus.pending, RunStatus.running) or self._store is None: + if not self._awaits_terminal_commit(record) or record.status in (RunStatus.pending, RunStatus.running) or self._store is None: return False row = await self._store.get(record.run_id, user_id=record.user_id) return ( @@ -2218,16 +2229,17 @@ class RunManager: and row.get("goal_verdict") == record.goal_verdict ) - async def _release_confirmed_scheduled_barrier(self, record: RunRecord, *, timeout: float) -> bool: - if not record.scheduled_goal_cleanup_pending or record.status in (RunStatus.pending, RunStatus.running): + async def _release_confirmed_terminal_barrier(self, record: RunRecord, *, timeout: float) -> bool: + if not self._awaits_terminal_commit(record) or record.status in (RunStatus.pending, RunStatus.running): return False try: async with asyncio.timeout(timeout): - if await self._scheduled_terminal_write_confirmed(record): + if await self._terminal_write_confirmed(record): record.scheduled_goal_cleanup_pending = False + record.terminal_commit_pending = False return True except Exception: - logger.warning("Unable to verify scheduled terminal commit for run %s", record.run_id, exc_info=True) + logger.warning("Unable to verify terminal commit for run %s", record.run_id, exc_info=True) return False async def _renew_leases(self) -> None: @@ -2254,30 +2266,32 @@ class RunManager: # saturation, slow checkpoint hydrate on a fresh worker), peer # reconciliation will reclaim the run as an orphan and mark it # ``error`` even though this worker still intends to execute it. + # The same holds for a locally terminal run whose worker is still + # finalizing before its deferred terminal commit. active_runs = [ (rid, record) for rid, record in self._runs.items() - if (record.status in (RunStatus.pending, RunStatus.running) or record.scheduled_goal_cleanup_pending) and record.owner_worker_id == self._worker_id and (record.task is None or not record.task.done()) + if (record.status in (RunStatus.pending, RunStatus.running) or self._awaits_terminal_commit(record)) and record.owner_worker_id == self._worker_id and (record.task is None or not record.task.done()) ] for run_id, record in active_runs: - if record.status not in (RunStatus.pending, RunStatus.running) and not record.scheduled_goal_cleanup_pending: + if record.status not in (RunStatus.pending, RunStatus.running) and not self._awaits_terminal_commit(record): continue confirmed_deadline = self._parse_lease_deadline(record.lease_expires_at) - if record.scheduled_goal_cleanup_pending and record.status not in (RunStatus.pending, RunStatus.running): + if self._awaits_terminal_commit(record) and record.status not in (RunStatus.pending, RunStatus.running): # The DB may have committed the terminal CAS while the worker # is still receiving its acknowledgement. It no longer needs # renewal and must not be mistaken for a peer takeover. remaining = (confirmed_deadline - datetime.now(UTC)).total_seconds() if confirmed_deadline is not None else 0 - if await self._release_confirmed_scheduled_barrier(record, timeout=remaining if remaining > 0 else lease_seconds): + if await self._release_confirmed_terminal_barrier(record, timeout=remaining if remaining > 0 else lease_seconds): continue if confirmed_deadline is None or confirmed_deadline <= datetime.now(UTC): - if await self._release_confirmed_scheduled_barrier(record, timeout=lease_seconds): + if await self._release_confirmed_terminal_barrier(record, timeout=lease_seconds): continue await self._mark_ownership_lost( record, reason="Lease ownership could not be confirmed before the last confirmed lease expired.", - require_active=not record.scheduled_goal_cleanup_pending, + require_active=not self._awaits_terminal_commit(record), ) continue @@ -2296,12 +2310,12 @@ class RunManager: ) if renewal.renewed: if confirmed_deadline <= datetime.now(UTC): - if await self._release_confirmed_scheduled_barrier(record, timeout=lease_seconds): + if await self._release_confirmed_terminal_barrier(record, timeout=lease_seconds): continue await self._mark_ownership_lost( record, reason="Lease renewal completed after the last confirmed lease had already expired.", - require_active=not record.scheduled_goal_cleanup_pending, + require_active=not self._awaits_terminal_commit(record), ) continue # Unsynced write is benign: ``lease_expires_at`` is the @@ -2330,7 +2344,7 @@ class RunManager: async with self._lock: still_active = ( self._runs.get(run_id) is record - and (record.status in (RunStatus.pending, RunStatus.running) or record.scheduled_goal_cleanup_pending) + and (record.status in (RunStatus.pending, RunStatus.running) or self._awaits_terminal_commit(record)) and record.owner_worker_id == self._worker_id and (record.task is None or not record.task.done()) ) @@ -2338,7 +2352,7 @@ class RunManager: # A detached precheck may have observed running just # before our terminal CAS committed. Re-read before # treating the rejected renewal as lost ownership. - if await self._release_confirmed_scheduled_barrier(record, timeout=lease_seconds): + if await self._release_confirmed_terminal_barrier(record, timeout=lease_seconds): continue logger.warning( "Run %s lease renewal failed (status=%s,owner=%s) – worker likely taken over; aborting local task", @@ -2349,16 +2363,16 @@ class RunManager: await self._mark_ownership_lost( record, reason="The durable store rejected lease renewal for this worker.", - require_active=not record.scheduled_goal_cleanup_pending, + require_active=not self._awaits_terminal_commit(record), ) except Exception: if confirmed_deadline <= datetime.now(UTC): - if await self._release_confirmed_scheduled_barrier(record, timeout=lease_seconds): + if await self._release_confirmed_terminal_barrier(record, timeout=lease_seconds): continue await self._mark_ownership_lost( record, reason="Lease ownership could not be confirmed before the last confirmed lease expired.", - require_active=not record.scheduled_goal_cleanup_pending, + require_active=not self._awaits_terminal_commit(record), ) else: logger.warning( diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index e79f241bdf..a8034e52c8 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -913,6 +913,9 @@ async def run_agent( run_events_config = ctx.run_events_config thread_store = ctx.thread_store terminal_status_kwargs = {"persist": False} if event_store is not None else {} + # Staged terminal statuses commit only after receipt and duration writes; + # keep the still-active durable row's lease renewed until then. + record.terminal_commit_pending = event_store is not None run_id = record.run_id thread_id = record.thread_id @@ -1773,6 +1776,9 @@ async def run_agent( await run_manager.persist_current_status(run_id) except Exception: logger.warning("Failed to persist terminal status for run %s after delivery receipt attempts", run_id, exc_info=True) + # The deferred commit has been attempted. A failed write is left to + # completion fallback or lease recovery, as before. + record.terminal_commit_pending = False if not record.ownership_lost and journal is not None and persist_completion: try: diff --git a/backend/tests/test_multi_worker_run_ownership.py b/backend/tests/test_multi_worker_run_ownership.py index 7b4af73a57..5bda66218c 100644 --- a/backend/tests/test_multi_worker_run_ownership.py +++ b/backend/tests/test_multi_worker_run_ownership.py @@ -8,6 +8,7 @@ Coverage: - periodic reconciliation notifies Gateway recovery orchestration - Worker reconciliation skips runs with unexpired leases - Lease heartbeat renews active run leases +- Lease heartbeat keeps renewing a staged terminal run until its commit - GATEWAY_WORKERS=1 + heartbeat_enabled=false behaviour unchanged """ @@ -15,14 +16,17 @@ from __future__ import annotations import asyncio from datetime import UTC, datetime, timedelta +from types import SimpleNamespace from unittest.mock import AsyncMock import pytest from deerflow.config.run_ownership_config import RunOwnershipConfig from deerflow.runtime import ORPHAN_RECOVERY_STOP_REASON, RunManager, RunStatus, ThreadOperationKind +from deerflow.runtime.events.store.memory import MemoryRunEventStore from deerflow.runtime.runs.manager import CancelOutcome, ConflictError, _generate_worker_id from deerflow.runtime.runs.store.memory import MemoryRunStore +from deerflow.runtime.runs.worker import RunContext, run_agent # --------------------------------------------------------------------------- # Helpers @@ -1114,6 +1118,139 @@ async def test_heartbeat_skips_runs_not_owned_by_this_worker(): assert stored["lease_expires_at"] == old_lease +@pytest.mark.anyio +async def test_staged_terminal_run_keeps_lease_through_worker_finalization(): + """A worker finalizing a staged success must keep its lease renewed. + + With an event store the worker stages the terminal status locally and + commits it only after its receipt and duration writes. The durable row + stays active meanwhile, so an unrenewed lease would let a peer reclaim + the successful run as an orphan ``error``. + """ + config = _lease_config(lease_seconds=30, grace_seconds=0, heartbeat_enabled=True) + store = MemoryRunStore() + owner = _make_manager(store=store, worker_id="worker-a", run_ownership_config=config) + peer = _make_manager(store=store, worker_id="worker-b", run_ownership_config=config) + record = await owner.create_or_reject("thread-finalize") + bridge = SimpleNamespace(publish=AsyncMock(), publish_end=AsyncMock(), cleanup=AsyncMock()) + finalizing = asyncio.Event() + release = asyncio.Event() + + class SlowReceiptEventStore(MemoryRunEventStore): + async def put_if_absent(self, **event): + if event["event_type"] == "run.delivery": + finalizing.set() + await release.wait() + return await super().put_if_absent(**event) + + class FinishingAgent: + async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False): + del graph_input, config, stream_mode, subgraphs + yield {"messages": []} + + task = asyncio.create_task( + run_agent( + bridge, + owner, + record, + ctx=RunContext(checkpointer=None, event_store=SlowReceiptEventStore()), + agent_factory=lambda **_kwargs: FinishingAgent(), + graph_input={}, + config={}, + ) + ) + record.task = task + + try: + await asyncio.wait_for(finalizing.wait(), timeout=1) + assert record.status == RunStatus.success + assert (await store.get(record.run_id))["status"] == "running" + + # Shrink the confirmed lease so it lapses during the test; only a + # heartbeat renewal can keep the peer from claiming the row. + near_expiry = (datetime.now(UTC) + timedelta(milliseconds=50)).isoformat() + record.lease_expires_at = near_expiry + store._runs[record.run_id]["lease_expires_at"] = near_expiry + await owner._renew_leases() + await asyncio.sleep(0.1) + + assert await peer.reconcile_orphaned_inflight_runs(error="orphaned") == [] + finally: + release.set() + await asyncio.wait_for(task, timeout=1) + + assert record.ownership_lost is False + assert record.terminal_commit_pending is False + assert (await store.get(record.run_id))["status"] == "success" + + +@pytest.mark.anyio +async def test_renewal_rejected_by_own_terminal_commit_releases_barrier(): + """A renewal racing this worker's terminal commit must not fence the run.""" + config = _lease_config(lease_seconds=30, heartbeat_enabled=True) + store = MemoryRunStore() + manager = _make_manager(store=store, worker_id="worker-a", run_ownership_config=config) + record = await manager.create_or_reject("thread-1") + record.task = asyncio.create_task(asyncio.sleep(3600)) + record.terminal_commit_pending = True + await manager.set_status(record.run_id, RunStatus.success, persist=False) + # The terminal write lands before the worker clears its barrier. + assert (await store.finalize_if_not_cancelled(record.run_id, status="success")).finalized + + try: + await manager._renew_leases() + + assert record.ownership_lost is False + assert record.terminal_commit_pending is False + assert record.status == RunStatus.success + finally: + record.task.cancel() + with pytest.raises(asyncio.CancelledError): + await record.task + + +@pytest.mark.anyio +async def test_renewal_rejected_by_peer_fences_staged_terminal_run(): + """A peer claim during finalization must fence the staged outcome.""" + config = _lease_config(lease_seconds=30, heartbeat_enabled=True) + store = MemoryRunStore() + manager = _make_manager(store=store, worker_id="worker-a", run_ownership_config=config) + record = await manager.create_or_reject("thread-1") + record.task = asyncio.create_task(asyncio.sleep(3600)) + record.terminal_commit_pending = True + await manager.set_status(record.run_id, RunStatus.success, persist=False) + store._runs[record.run_id]["owner_worker_id"] = "worker-b" + + await manager._renew_leases() + await asyncio.sleep(0) + + assert record.ownership_lost is True + assert record.task.cancelled() + + +@pytest.mark.anyio +async def test_fenced_staged_terminal_run_stops_renewing(): + """Fencing ends the terminal-commit barrier, so its lease can lapse.""" + config = _lease_config(lease_seconds=30, heartbeat_enabled=True) + store = MemoryRunStore() + manager = _make_manager(store=store, worker_id="worker-a", run_ownership_config=config) + record = await manager.create_or_reject("thread-1") + record.task = asyncio.create_task(asyncio.sleep(3600)) + record.terminal_commit_pending = True + await manager.set_status(record.run_id, RunStatus.success, persist=False) + await manager._mark_ownership_lost(record, reason="Test fence", require_active=False) + store.update_lease = AsyncMock(wraps=store.update_lease) + + try: + await manager._renew_leases() + + store.update_lease.assert_not_awaited() + assert record.terminal_commit_pending is False + finally: + with pytest.raises(asyncio.CancelledError): + await record.task + + @pytest.mark.anyio async def test_heartbeat_not_started_when_disabled(): """When heartbeat_enabled is False, start_heartbeat must be a no-op."""