From 1c10b7e7dbbeba0524bfe1f589d9c60821b2308b Mon Sep 17 00:00:00 2001 From: Thorsten Date: Sat, 18 Jul 2026 09:45:07 +0200 Subject: [PATCH] Fix NATS reconnect handling: retry forever instead of giving up after ~2min nats-py's default max_reconnect_attempts (60 * 2s) meant a long-enough NATS outage would leave worker.py's per-plugin consumer threads permanently stuck with no recovery path - unlike the AsyncBroker path used by action_worker.py, they had no closed_cb to notice and react to a permanent disconnect. Setting max_reconnect_attempts=-1 makes nats-py retry indefinitely instead; confirmed via a local test that a plugin thread survives a NATS restart and resumes receiving messages afterward without any restart of its own. Also bumps connect_timeout from nats-py's 2s default to 10s, since munin (a Pi Zero-class host) observed transient connect timeouts when ~40 plugin threads dialed in simultaneously at startup under the tighter default. Co-Authored-By: Claude Sonnet 5 --- src/distbot/common/broker_nats.py | 21 +++++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/src/distbot/common/broker_nats.py b/src/distbot/common/broker_nats.py index dfadc95..824e71d 100644 --- a/src/distbot/common/broker_nats.py +++ b/src/distbot/common/broker_nats.py @@ -132,7 +132,15 @@ class NatsSyncBroker(SyncBroker): return asyncio.run_coroutine_threadsafe(coro, self._loop).result() async def _connect(self): - self._nc = await nats.connect(conf_get("nats_uri")) + # max_reconnect_attempts=-1: retry forever rather than give up after + # nats-py's default ~2 minutes (60 attempts * 2s) - there is no + # closed_cb here to notice a permanent give-up, so without this a + # long-enough NATS outage would leave the consumer thread silently + # stuck forever (run_forever()'s wait() never unblocks on its own). + # connect_timeout=10: nats-py's 2s default was too tight when ~40 + # plugin threads all dial in at once on weak hardware (observed + # transient TimeoutErrors on a Raspberry Pi Zero-class host). + self._nc = await nats.connect(conf_get("nats_uri"), max_reconnect_attempts=-1, connect_timeout=10) self._js = self._nc.jetstream() def close(self) -> None: @@ -219,13 +227,18 @@ class NatsAsyncBroker(AsyncBroker): self, queue: str, on_message: OnMessage, on_channel_closed: Callable[[], None] = None, ) -> DurableChannel: async def _on_closed(): - # fires once nats-py gives up reconnecting entirely - the closest - # analog to pika's channel-close callback, which today fires on - # any lost connection (this app doesn't implement reconnection). + # With max_reconnect_attempts=-1 this should only fire on a + # genuinely unexpected permanent closure (nats-py never gives up + # from a normal outage) - kept as a defense-in-depth backstop, + # the closest analog to pika's channel-close callback which + # today fires on any lost connection (this app doesn't + # implement AMQP reconnection at all). on_channel_closed() self._nc = await nats.connect( conf_get("nats_uri"), + max_reconnect_attempts=-1, + connect_timeout=10, closed_cb=_on_closed if on_channel_closed else None, ) self._js = self._nc.jetstream() -- 2.47.3