From 015d0f38356978bace2eac6a67a0e05a338e0c0a Mon Sep 17 00:00:00 2001 From: Towaki Takikawa Date: Sun, 25 Jan 2026 14:05:26 -0800 Subject: [PATCH 1/3] Fix idempotency_key conflict in orphan task retry --- .../sqlc/set_orphaned_task_execution_to_lost_and_retry.py | 1 + 1 file changed, 1 insertion(+) diff --git a/hyrex/dispatcher/sqlc/set_orphaned_task_execution_to_lost_and_retry.py b/hyrex/dispatcher/sqlc/set_orphaned_task_execution_to_lost_and_retry.py index 6c05533..c51af0b 100644 --- a/hyrex/dispatcher/sqlc/set_orphaned_task_execution_to_lost_and_retry.py +++ b/hyrex/dispatcher/sqlc/set_orphaned_task_execution_to_lost_and_retry.py @@ -64,6 +64,7 @@ NOW() FROM lost_tasks WHERE attempt_number < max_retries +ON CONFLICT (task_name, idempotency_key) WHERE idempotency_key IS NOT NULL DO NOTHING """ From 344a81f8d72036e3d3bcae44118c739edf0f5729 Mon Sep 17 00:00:00 2001 From: Towaki Takikawa Date: Wed, 15 Apr 2026 00:58:54 -0700 Subject: [PATCH 2/3] Bound TimeSeriesAverager with a ring buffer to stop executor leak The executor's three TimeSeriesAverager instances (num_distinct_queues, refresh_queue_duration, dequeue_duration) were appending a DataPoint on every executor poll iteration and never pruning. Because run_round_robin_loop has no sleep when the queue is empty, an idle worker busy-polls the database hundreds of times per second, and each of those iterations submits to 2-3 averagers. On a quiet dev Postgres this leaked roughly 40 MB/min per worker, which at -p 8 produced ~320 MB/min of RSS growth and eventually OOM-killed unrelated processes on the host (dbus, wireplumber, etc.), which in turn took down NetworkManager as collateral damage. Swap data_points for a bounded collections.deque with maxlen=10_000 so append is O(1), old points are auto-evicted, and worst-case memory is capped at ~2 MB per averager regardless of poll rate. clear() and prune_data_older_than() are updated to preserve the deque's maxlen. At ~100 submits/sec the buffer still holds a couple of minutes of stats, which is enough for the existing get_time_series() consumers. Co-Authored-By: Claude Opus 4.6 (1M context) --- hyrex/worker/executor/time_series_averager.py | 20 ++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/hyrex/worker/executor/time_series_averager.py b/hyrex/worker/executor/time_series_averager.py index b7b5411..1f5c681 100644 --- a/hyrex/worker/executor/time_series_averager.py +++ b/hyrex/worker/executor/time_series_averager.py @@ -1,4 +1,5 @@ import time +from collections import deque from pydantic import BaseModel @@ -13,9 +14,17 @@ class MinuteAverage(BaseModel): average: float +# Bound the per-averager ring buffer. The executor poll loop submits to these +# averagers on every iteration and previously never pruned, which leaked ~40 +# MB/min per worker under an idle busy-poll. 10k entries is enough to hold +# several minutes of stats at realistic submit rates while capping worst-case +# memory at ~2 MB per averager. +_MAX_DATA_POINTS = 10_000 + + class TimeSeriesAverager: def __init__(self): - self.data_points: list[DataPoint] = [] + self.data_points: deque[DataPoint] = deque(maxlen=_MAX_DATA_POINTS) def _get_minute_timestamp(self, timestamp: int) -> int: # Round down to nearest minute @@ -71,9 +80,10 @@ def get_current_minute_average(self) -> MinuteAverage: return result def clear(self) -> None: - self.data_points = [] + self.data_points.clear() def prune_data_older_than(self, timestamp: int) -> None: - self.data_points = [ - point for point in self.data_points if point.timestamp >= timestamp - ] + self.data_points = deque( + (point for point in self.data_points if point.timestamp >= timestamp), + maxlen=_MAX_DATA_POINTS, + ) From 555e2f46d446f622183a9ec79460f7f553a67ea5 Mon Sep 17 00:00:00 2001 From: Towaki Takikawa Date: Mon, 20 Apr 2026 15:08:18 -0700 Subject: [PATCH 3/3] Fix worker shutdown hang from non-daemon message listener thread MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The WorkerRootProcess message-listener thread was started without daemon=True, and its body blocks indefinitely on a multiprocessing Queue.get(). The shutdown path tries to wake it with a put(None) sentinel, but after executor processes are SIGKILL'd their queue feeder threads can die mid-send and leave the parent receiver wedged. When join(timeout=5.0) returns with the thread still alive, the code emits a warning and continues — but Python won't exit the interpreter while a non-daemon thread is alive, so the worker process hangs forever after a Ctrl+C / SIGTERM. Mark the listener thread as a daemon at creation time so the interpreter can reap it on exit regardless of queue state. Also update the misleading comment in stop() that referred to promoting the thread to a daemon during shutdown — that's impossible once the thread has started (Thread.daemon setter raises RuntimeError). Co-Authored-By: Claude Opus 4.7 (1M context) --- hyrex/worker/root_process.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/hyrex/worker/root_process.py b/hyrex/worker/root_process.py index e9bb81e..be6cb04 100644 --- a/hyrex/worker/root_process.py +++ b/hyrex/worker/root_process.py @@ -227,7 +227,9 @@ def send_heartbeats(self): ) def run(self): - self.message_listener_thread = threading.Thread(target=self._message_listener) + self.message_listener_thread = threading.Thread( + target=self._message_listener, daemon=True + ) self.message_listener_thread.start() self.logger.info("Incoming message queue now active...") @@ -332,8 +334,7 @@ def stop(self): self.message_listener_thread.join(timeout=5.0) if self.message_listener_thread.is_alive(): self.logger.warning("Message listener thread did not exit cleanly within timeout.") - # Force terminate the thread by setting it as daemon and exiting - # Python will clean it up on process exit + # Thread is a daemon, so the interpreter will terminate it on process exit. else: self.logger.info("Message listener thread closed successfully.")