diff --git a/CHANGELOG.md b/CHANGELOG.md index 382bb17d1..86c8b6f6d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -198,6 +198,9 @@ This section accumulates work toward the **2.1.0** milestone ### Fixed +- **runtime:** Re-check orphan candidates through an atomic, lease-aware takeover + claim so a successful heartbeat after the scan keeps the run active and only + one reconciler reports recovery. ([#4424]) - **skills:** Apply `allowed-tools` only to slash-activated or actually loaded lead-agent skills, preventing passive enabled skills and evaluation fixtures from removing MCP, web, file, and delegation tools from every run. ([#4095], @@ -1114,3 +1117,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#4287]: https://github.com/bytedance/deer-flow/pull/4287 [#4288]: https://github.com/bytedance/deer-flow/pull/4288 [#4324]: https://github.com/bytedance/deer-flow/issues/4324 +[#4424]: https://github.com/bytedance/deer-flow/issues/4424 diff --git a/README.md b/README.md index d77e58444..9fc1a6507 100644 --- a/README.md +++ b/README.md @@ -276,6 +276,8 @@ DeerFlow still uses `Forwarded` / `X-Forwarded-*` headers to recover the browser > [!IMPORTANT] > The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), and `run_ownership.heartbeat_enabled: true`. The bridge shares SSE delivery and `Last-Event-ID` replay across workers, while lease reconciliation marks runs from dead workers as errors, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE and `/wait` consumers also refresh durable status on heartbeats as a fallback if terminal publication fails. The rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. IM channel state and other process-local services still need their own multi-worker coordination. +> +> Reconciliation uses an atomic takeover claim that re-checks the lease after candidate selection, so a successful owner renewal wins over orphan recovery and only one reconciler can report a run as recovered. See [CONTRIBUTING.md](CONTRIBUTING.md) for detailed Docker development guide. diff --git a/backend/packages/harness/deerflow/runtime/runs/manager.py b/backend/packages/harness/deerflow/runtime/runs/manager.py index 1dfaad500..a52739067 100644 --- a/backend/packages/harness/deerflow/runtime/runs/manager.py +++ b/backend/packages/harness/deerflow/runtime/runs/manager.py @@ -1213,7 +1213,10 @@ class RunManager: Rows with a still-valid lease are skipped — they belong to another live worker. Rows with a NULL lease (pre-ownership data) are reclaimed as - well, matching the original single-worker recovery behaviour. + well, matching the original single-worker recovery behaviour. The + candidate scan is only an optimization: each row is claimed with a + lease-aware conditional update so a heartbeat renewal after the scan + always wins over reconciliation. """ if self._store is None: return [] diff --git a/backend/tests/test_multi_worker_run_ownership.py b/backend/tests/test_multi_worker_run_ownership.py index 67f0f6fc2..57f237748 100644 --- a/backend/tests/test_multi_worker_run_ownership.py +++ b/backend/tests/test_multi_worker_run_ownership.py @@ -403,6 +403,51 @@ async def test_reconciliation_skips_candidate_when_owner_renews_lease_after_scan assert datetime.fromisoformat(stored["lease_expires_at"]) > datetime.now(UTC) +@pytest.mark.anyio +async def test_concurrent_reconcilers_report_one_successful_claim(): + """Two reapers scanning the same candidate must report one recovery.""" + store = MemoryRunStore() + grace = 10 + run_id = "contended-orphan" + await store.put( + run_id, + thread_id="thread-1", + status="running", + owner_worker_id="worker-dead", + lease_expires_at=(datetime.now(UTC) - timedelta(seconds=grace + 5)).isoformat(), + created_at=(datetime.now(UTC) - timedelta(seconds=120)).isoformat(), + ) + + original_scan = store.list_inflight_with_expired_lease + both_scanned = asyncio.Event() + scan_lock = asyncio.Lock() + scan_count = 0 + + async def synchronized_scan(*, before=None, grace_seconds=10): + nonlocal scan_count + rows = [dict(row) for row in await original_scan(before=before, grace_seconds=grace_seconds)] + async with scan_lock: + scan_count += 1 + if scan_count == 2: + both_scanned.set() + await asyncio.wait_for(both_scanned.wait(), timeout=1) + return rows + + store.list_inflight_with_expired_lease = synchronized_scan + managers = [ + _make_manager(store=store, run_ownership_config=_lease_config(heartbeat_enabled=True, grace_seconds=grace)), + _make_manager(store=store, run_ownership_config=_lease_config(heartbeat_enabled=True, grace_seconds=grace)), + ] + + results = await asyncio.gather(*(manager.reconcile_orphaned_inflight_runs(error="orphaned") for manager in managers)) + + assert sorted(len(recovered) for recovered in results) == [0, 1] + assert [record.run_id for recovered in results for record in recovered] == [run_id] + row = await store.get(run_id) + assert row is not None + assert row["status"] == "error" + + @pytest.mark.anyio async def test_reconciliation_claims_null_lease_runs(): """Pre-ownership rows (NULL lease) must be reclaimed.""" diff --git a/backend/tests/test_run_repository.py b/backend/tests/test_run_repository.py index 82a2b3ef8..95952d463 100644 --- a/backend/tests/test_run_repository.py +++ b/backend/tests/test_run_repository.py @@ -958,6 +958,47 @@ class TestRunRepository: assert row["status"] == "running" await _cleanup() + @pytest.mark.anyio + async def test_reconciliation_skips_run_renewed_after_scan(self, tmp_path): + """The SQL takeover CAS must reject a candidate renewed after its scan.""" + repo = await _make_repo(tmp_path) + grace = 10 + run_id = "renewed-after-scan" + owner_worker_id = "worker-alive" + try: + await repo.put( + run_id, + thread_id="t1", + status="running", + owner_worker_id=owner_worker_id, + lease_expires_at=(datetime.now(UTC) - timedelta(seconds=grace + 5)).isoformat(), + created_at=(datetime.now(UTC) - timedelta(seconds=120)).isoformat(), + ) + original_scan = repo.list_inflight_with_expired_lease + + async def scan_then_renew(*, before=None, grace_seconds=10): + rows = await original_scan(before=before, grace_seconds=grace_seconds) + renewed = await repo.update_lease( + run_id, + owner_worker_id=owner_worker_id, + lease_expires_at=(datetime.now(UTC) + timedelta(seconds=60)).isoformat(), + ) + assert renewed is True + return rows + + repo.list_inflight_with_expired_lease = scan_then_renew + manager = RunManager(store=repo) + + recovered = await manager.reconcile_orphaned_inflight_runs(error="orphaned") + + row = await repo.get(run_id) + assert recovered == [] + assert row is not None + assert row["status"] == "running" + assert datetime.fromisoformat(row["lease_expires_at"]) > datetime.now(UTC) + finally: + await _cleanup() + @pytest.mark.anyio async def test_claim_for_takeover_succeeds_with_null_lease(self, tmp_path): repo = await _make_repo(tmp_path)