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 <noreply@anthropic.com>

diff --git a/src/distbot/common/broker_nats.py b/src/distbot/common/broker_nats.py
index dfadc95b6da3dd980d6fdd296bfaa409481440bb..824e71de923b7b8b734e6cc25c9e6726c50594e9 100644 (file)
--- 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()