From 2730ee1f7b32d5e48dbc06922afeea496016d8b4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E4=BA=91=E9=BE=99?= <76432572+nankingjing@users.noreply.github.com> Date: Sun, 12 Jul 2026 20:41:41 +0800 Subject: [PATCH] fix(memory): replace busy timer spin with deferred single re-run flag (#4073) When _process_queue found another worker already processing, it called _schedule_timer(0), spawning a fresh Timer thread immediately and looping tightly (spawn -> busy -> reschedule -> spawn) until the active worker finished. Replace this with a _reprocess_pending flag: a concurrent caller sets the flag and returns, and the active worker reschedules exactly once in its finally block when work remains. Reset the flag in clear(). --- .../harness/deerflow/agents/memory/queue.py | 14 ++++- backend/tests/test_memory_queue.py | 59 ++++++++++++++++++- 2 files changed, 68 insertions(+), 5 deletions(-) diff --git a/backend/packages/harness/deerflow/agents/memory/queue.py b/backend/packages/harness/deerflow/agents/memory/queue.py index d7a04e518..6cfc2494e 100644 --- a/backend/packages/harness/deerflow/agents/memory/queue.py +++ b/backend/packages/harness/deerflow/agents/memory/queue.py @@ -42,6 +42,7 @@ class MemoryUpdateQueue: self._lock = threading.Lock() self._timer: threading.Timer | None = None self._processing = False + self._reprocess_pending = False @staticmethod def _queue_key( @@ -181,8 +182,12 @@ class MemoryUpdateQueue: with self._lock: if self._processing: - # Preserve immediate flush semantics even if another worker is active. - self._schedule_timer(0) + # Another worker is already draining the queue. Instead of + # spawning a tight timer spin (repeatedly re-scheduling a + # 0-delay Timer thread while busy), defer a single re-run: the + # active worker checks this flag in its finally block and + # reschedules once if work remains. + self._reprocess_pending = True return if not self._queue: @@ -232,6 +237,10 @@ class MemoryUpdateQueue: finally: with self._lock: self._processing = False + if self._reprocess_pending: + self._reprocess_pending = False + if self._queue: + self._schedule_timer(0) def flush(self) -> None: """Force immediate processing of the queue. @@ -263,6 +272,7 @@ class MemoryUpdateQueue: self._timer = None self._queue.clear() self._processing = False + self._reprocess_pending = False @property def pending_count(self) -> int: diff --git a/backend/tests/test_memory_queue.py b/backend/tests/test_memory_queue.py index 421d28fc7..c9bea04ea 100644 --- a/backend/tests/test_memory_queue.py +++ b/backend/tests/test_memory_queue.py @@ -134,17 +134,70 @@ def test_add_nowait_cancels_existing_timer_and_starts_immediate_timer() -> None: created_timer.start.assert_called_once_with() -def test_process_queue_reschedules_immediately_when_already_processing() -> None: +def test_process_queue_defers_reprocess_when_already_processing() -> None: + """When a timer fires while a worker is active, ``_process_queue`` must set the + deferred-rerun flag instead of spinning up a tight 0-delay Timer chain. + + The old behavior re-scheduled a 0-delay Timer on every re-entry while busy, + burning a fresh thread each time. The fix defers a single re-run via + ``_reprocess_pending`` that the finishing worker honors once. + """ queue = MemoryUpdateQueue() queue._processing = True - created_timer = MagicMock() - with patch("deerflow.agents.memory.queue.threading.Timer", return_value=created_timer) as timer_cls: + with patch("deerflow.agents.memory.queue.threading.Timer") as timer_cls: + queue._process_queue() + + timer_cls.assert_not_called() + assert queue._reprocess_pending is True + + +def test_finishing_worker_reschedules_once_when_reprocess_pending() -> None: + """A worker that finishes with ``_reprocess_pending`` set and work still queued + schedules exactly one follow-up run (not a per-arrival timer spin).""" + queue = MemoryUpdateQueue() + queue._queue = [ConversationContext(thread_id="thread-1", messages=["first"], agent_name="lead_agent")] + queue._reprocess_pending = True + created_timer = MagicMock() + mock_updater = MagicMock() + + def _enqueue_more_while_processing(**_kwargs) -> bool: + # Simulate a new update arriving mid-processing so the finally block sees + # remaining work and reschedules exactly once. + queue._queue.append(ConversationContext(thread_id="thread-2", messages=["second"], agent_name="lead_agent")) + return True + + mock_updater.update_memory.side_effect = _enqueue_more_while_processing + + with ( + patch("deerflow.agents.memory.updater.MemoryUpdater", return_value=mock_updater), + patch("deerflow.agents.memory.queue.threading.Timer", return_value=created_timer) as timer_cls, + ): queue._process_queue() timer_cls.assert_called_once_with(0, queue._process_queue) assert created_timer.daemon is True created_timer.start.assert_called_once_with() + assert queue._reprocess_pending is False + + +def test_finishing_worker_does_not_reschedule_when_no_work_remains() -> None: + """The deferred re-run is cleared even when nothing is left to process, so a + stray flag never leaves a dangling ``_reprocess_pending``.""" + queue = MemoryUpdateQueue() + queue._queue = [ConversationContext(thread_id="thread-1", messages=["only"], agent_name="lead_agent")] + queue._reprocess_pending = True + mock_updater = MagicMock() + mock_updater.update_memory.return_value = True + + with ( + patch("deerflow.agents.memory.updater.MemoryUpdater", return_value=mock_updater), + patch("deerflow.agents.memory.queue.threading.Timer") as timer_cls, + ): + queue._process_queue() + + timer_cls.assert_not_called() + assert queue._reprocess_pending is False def test_flush_nowait_is_non_blocking() -> None: