From 8941fbe872e574b5018a29ef03d4d2e0753bc7e3 Mon Sep 17 00:00:00 2001 From: RongJie G <111257566+CorgiBoyG@users.noreply.github.com> Date: Sun, 18 Oct 2026 09:39:08 +0800 Subject: [PATCH] fix(runtime): bound RunJournal progress teardown (#6990) Cancel event and progress writes before force-close waits, bound progress settlement, and supervise cancellation-resistant reporters until their late result is observed. Fixes #6986 Signed-off-by: CorgiBoyG <111257566+CorgiBoyG@users.noreply.github.com> --- .../harness/deerflow/runtime/AGENTS.md | 2 +- .../harness/deerflow/runtime/journal.py | 73 +++++++++++++-- backend/tests/test_run_journal.py | 88 ++++++++++++++++--- 3 files changed, 144 insertions(+), 19 deletions(-) diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index c1460f4b52..e532474066 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -138,7 +138,7 @@ prompts, and derived hashes do not enter the event. Ordinary task-tool subagents recorder proxy, never the journal itself; durable batch runs have no parent journal and emit no such event. Recorder failures are fail-open. -**RunJournal write ownership** (`runtime/journal.py`): a batch handed to `store.put_batch()` is owned by a journal task until its outcome settles, so a cancelled `flush()`/`close()` caller detaches the waiter and re-raises `CancelledError` only after the owned write is drained — never abandon or replay the in-flight batch. A generic `put_batch` exception is an ambiguous outcome: it is quarantined, blocks every successor write, and makes `flush()`/`close()` report non-success instead of replaying the batch. Only `RunEventStoreNotCommittedError` means the store proved none of the batch committed and may be deliberately retried; stores must raise it instead of a bare exception for that case. Explicit `flush()` and `close(flush=True)` share one per-journal operation lock, so a successor batch can never overtake an unsettled predecessor, every waiter observes the same settlement, and a normal close detaches only after the whole buffer has settled. `close(flush=False)` is a separate abort path that never queues on that lock: it revokes write ownership before its first await and makes any in-flight flush report non-success rather than a late success. A cancelled write waits a bounded time for the store call to end; `Task.cancelled()`/a clean return does not prove an uncooperative backend's I/O actually stopped, so this is a bound on waiting, not a termination guarantee. An inner store task that outlives that bound stays supervised until it settles, so its late result is retrieved and logged rather than dropped. Two adjacent limits are known and tracked separately, not fixed here: the bounded wait covers the store write, not a `progress_reporter` that ignores cancellation (a pre-existing hazard on the force-close path); and background-write warnings log the exception type and message, whose text is backend-controlled and may echo driver-level parameters, so log consumers must not treat it as payload-free. +**RunJournal write ownership** (`runtime/journal.py`): a batch handed to `store.put_batch()` is owned by a journal task until its outcome settles, so a cancelled `flush()`/`close()` caller detaches the waiter and re-raises `CancelledError` only after the owned write is drained — never abandon or replay the in-flight batch. A generic `put_batch` exception is an ambiguous outcome: it is quarantined, blocks every successor write, and makes `flush()`/`close()` report non-success instead of replaying the batch. Only `RunEventStoreNotCommittedError` means the store proved none of the batch committed and may be deliberately retried; stores must raise it instead of a bare exception for that case. Explicit `flush()` and `close(flush=True)` share one per-journal operation lock, so a successor batch can never overtake an unsettled predecessor, every waiter observes the same settlement, and a normal close detaches only after the whole buffer has settled. `close(flush=False)` is a separate abort path that never queues on that lock: it revokes write ownership before its first await and makes any in-flight flush report non-success rather than a late success. A cancelled event write or pending progress snapshot waits a bounded time to settle; `Task.cancelled()`/a clean return does not prove an uncooperative backend's I/O actually stopped, so this is a bound on waiting, not a termination guarantee. A task that outlives that bound stays supervised until it settles, so its late result is retrieved and logged rather than dropped while lease-loss finalization continues. Background-write and progress-reporter warnings log the exception type and message, whose text is backend-controlled and may echo driver-level parameters, so log consumers must not treat it as payload-free. **JSONL record boundaries** (`runtime/events/store/jsonl.py`): thread reads, run reads, and sequence recovery split on physical newlines. Do not use diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index 38a356d374..5ea8e1ffaa 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/backend/packages/harness/deerflow/runtime/journal.py @@ -354,6 +354,10 @@ class RunJournal(BaseCallbackHandler): # late result is retrieved and logged rather than dropped. This is not a # termination guarantee: it only keeps settlement responsibility. self._abandoned_store_tasks: dict[asyncio.Task[None], int] = {} + # Progress persistence is best-effort, but a cancellation-resistant + # reporter must not block lease-loss close. Keep any task that outlives + # bounded teardown strongly referenced until its result is retrieved. + self._abandoned_progress_tasks: set[asyncio.Task[None]] = set() # Batches whose outcome is UNKNOWN (a generic store failure) must never # be replayed: the store may have committed before losing its ACK. # Keeping them here also fences every successor write for this run. @@ -1184,6 +1188,57 @@ class RunJournal(BaseCallbackHandler): batch_size, ) + def _supervise_abandoned_progress_task(self, task: asyncio.Task[None]) -> None: + """Retain a progress task that outlived bounded journal teardown.""" + if task in self._abandoned_progress_tasks: + return + self._abandoned_progress_tasks.add(task) + task.add_done_callback(self._on_abandoned_progress_task_settled) + + def _observe_progress_task_failure(self, task: asyncio.Task[None], *, after_abandonment: bool) -> None: + if task.cancelled(): + return + exc = task.exception() + if exc is None: + return + logger.warning( + "RunJournal progress reporter for run %s failed %s (%s: %s)", + self.run_id, + "after journal abandonment" if after_abandonment else "during journal teardown", + type(exc).__name__, + exc, + ) + + def _on_abandoned_progress_task_settled(self, task: asyncio.Task[None]) -> None: + self._abandoned_progress_tasks.discard(task) + if task.cancelled(): + return + if task.exception() is not None: + self._observe_progress_task_failure(task, after_abandonment=True) + return + logger.warning( + "RunJournal progress reporter for run %s settled after journal abandonment", + self.run_id, + ) + + async def _settle_cancelled_progress_task(self, task: asyncio.Task[None]) -> None: + """Bound teardown of a progress task already cancelled by force-close.""" + try: + done, _pending = await asyncio.wait( + {task}, + timeout=_WRITE_SETTLE_TIMEOUT_SECONDS, + ) + except asyncio.CancelledError: + if task.done(): + self._observe_progress_task_failure(task, after_abandonment=False) + else: + self._supervise_abandoned_progress_task(task) + raise + if task not in done: + self._supervise_abandoned_progress_task(task) + return + self._observe_progress_task_failure(task, after_abandonment=False) + async def _drain_pending_writes(self) -> None: """Settle every journal-owned write before any successor may start. @@ -1653,8 +1708,8 @@ class RunJournal(BaseCallbackHandler): self._pending_llm_response = None self._owned_writes = {} self._draining_tasks = set() - # Intentionally NOT cleared: abandoned inner store tasks stay strongly - # referenced here until they settle and their result is logged. + # Intentionally NOT cleared: abandoned store/progress tasks stay strongly + # referenced by their supervisor collections until they settle. self._pending_progress_task = None self._pending_progress_delayed = False self._progress_dirty = False @@ -1713,12 +1768,20 @@ class RunJournal(BaseCallbackHandler): pending_owned = tuple(self._owned_writes) for task in pending_owned: task.cancel() - if pending_owned: - await asyncio.gather(*pending_owned, return_exceptions=True) pending_progress_task = self._pending_progress_task + cleanup_tasks = list(pending_owned) if pending_progress_task is not None: + # Revoke both write classes before the first await. Progress + # teardown must not queue behind an unrelated event write. pending_progress_task.cancel() - await asyncio.gather(pending_progress_task, return_exceptions=True) + cleanup_tasks.append( + asyncio.create_task( + self._settle_cancelled_progress_task(pending_progress_task), + name=f"run-journal-progress-teardown-{self.run_id}", + ) + ) + if cleanup_tasks: + await asyncio.gather(*cleanup_tasks, return_exceptions=True) finally: self._detach_runtime_dependencies() diff --git a/backend/tests/test_run_journal.py b/backend/tests/test_run_journal.py index 659029e3d8..715661e0ca 100644 --- a/backend/tests/test_run_journal.py +++ b/backend/tests/test_run_journal.py @@ -1152,15 +1152,12 @@ class TestOperationOwnership: await asyncio.gather(first, second, return_exceptions=True) @pytest.mark.anyio - async def test_force_close_progress_reporter_unbounded_is_characterised(self): - """Characterisation of a pre-existing limitation, tracked separately. + @pytest.mark.parametrize("late_failure", [False, True], ids=["late-success", "late-failure"]) + async def test_force_close_bounds_and_supervises_progress_reporter(self, monkeypatch, caplog, late_failure): + """Lease-loss close is bounded while a late progress result stays observed.""" + import deerflow.runtime.journal as journal_module - close(flush=False) bounds the store write (see the abandoned-store tests) - but still awaits the pending progress task without a deadline, so a - progress reporter that swallows cancellation can block a force close. - This reproduces identically on the pre-PR base, so it is documented here - rather than fixed inside the write-ownership PR. - """ + monkeypatch.setattr(journal_module, "_WRITE_SETTLE_TIMEOUT_SECONDS", 0.02, raising=False) started = asyncio.Event() release = asyncio.Event() @@ -1170,22 +1167,87 @@ class TestOperationOwnership: await asyncio.Event().wait() except asyncio.CancelledError: await release.wait() + if late_failure: + raise RuntimeError("late progress failure") journal = RunJournal("r-prog", "t-prog", MemoryRunEventStore(), flush_threshold=100, progress_reporter=reporter) journal._schedule_progress_flush() await started.wait() assert journal._pending_progress_task is not None - bounded = True + close_task = asyncio.create_task(journal.close(flush=False)) try: - await asyncio.wait_for(journal.close(flush=False), timeout=0.25) - except TimeoutError: - bounded = False + done, _pending = await asyncio.wait({close_task}, timeout=0.25) + assert close_task in done, "force close waited indefinitely for a cancellation-resistant progress reporter" + assert close_task.exception() is None + assert journal._closed is True + assert len(journal._abandoned_progress_tasks) == 1, "late progress task was not supervised" + + with caplog.at_level(logging.WARNING, logger="deerflow.runtime.journal"): + release.set() + await _pump_loop() + + assert journal._abandoned_progress_tasks == set() + if late_failure: + assert "Failed to persist progress snapshot for run r-prog" in caplog.text + assert "progress reporter for run r-prog settled after journal abandonment" in caplog.text finally: release.set() + await asyncio.gather(close_task, return_exceptions=True) + + @pytest.mark.anyio + async def test_force_close_progress_cleanup_survives_repeated_caller_cancellation(self, monkeypatch): + """Caller cancellation detaches cleanup without dropping the late task.""" + import deerflow.runtime.journal as journal_module + + monkeypatch.setattr(journal_module, "_WRITE_SETTLE_TIMEOUT_SECONDS", 1.0, raising=False) + write_started = asyncio.Event() + progress_started = asyncio.Event() + progress_cancelled = asyncio.Event() + release_write = asyncio.Event() + release_progress = asyncio.Event() + + class UncooperativeStore(MemoryRunEventStore): + async def put_batch(self, events): + write_started.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + await release_write.wait() + + async def reporter(snapshot): + progress_started.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + progress_cancelled.set() + await release_progress.wait() + + journal = RunJournal("r-prog-cancel", "t-prog-cancel", UncooperativeStore(), flush_threshold=1, progress_reporter=reporter) + journal._put(event_type="A", category="trace", content="one") + journal._schedule_progress_flush() + await asyncio.wait_for(asyncio.gather(write_started.wait(), progress_started.wait()), timeout=1) + + close_task = asyncio.create_task(journal.close(flush=False)) + try: + # Progress cancellation must not wait behind an unrelated event + # writer's bounded teardown. + await asyncio.wait_for(progress_cancelled.wait(), timeout=0.25) + + close_task.cancel("first close cancellation") + close_task.cancel("repeated close cancellation") + with pytest.raises(asyncio.CancelledError): + await close_task + + assert journal._closed is True + assert len(journal._abandoned_progress_tasks) == 1, "caller cancellation dropped progress-task supervision" + finally: + release_write.set() + release_progress.set() + await asyncio.gather(close_task, return_exceptions=True) await _pump_loop() - assert bounded is False, "force close became bounded; please update this characterisation" + assert journal._abandoned_progress_tasks == set() @pytest.mark.anyio async def test_concurrent_flush_not_committed_retry_is_characterised(self):