Fix a keepalive ping timeout killing the bot for good

Bot overrode disconnect() to tear down the workers and raise SystemExit,
but slixmpp calls disconnect() itself as the first half of a reconnect:
xep_0199's keepalive does self.xmpp.reconnect(0.0, "Ping timeout after
%ds") on a missed ping, and reconnect() calls disconnect(0.0, reason).
Against the override's signature (reconnect=False, wait=None, ...) that
0.0 landed on the reconnect parameter, so a single 30s ping timeout - a
routine hiccup - exited the process instead of reconnecting. The same
override also passed its bool into super().disconnect()'s wait: float.

Split the two concerns: shutdown() is the deliberate teardown (hangup
command, KeyboardInterrupt, action queue channel closed), disconnect()
is left to slixmpp.

That alone would leave the bot idle after a dropped connection, because
slixmpp does not reconnect an established stream on its own -
connection_made() clears _current_connection_attempt, so the later
connection_lost() only fires 'disconnected' and reschedule_connection_
attempt() bails on its is-None guard. So hook 'disconnected' and
reconnect ourselves, with a 1s..300s back-off reset on session_start, so
a server that accepts TCP but kills the stream (auth failure) doesn't
become a hot loop. When slixmpp is already retrying - its own reconnect()
path, or a _connect_loop still in flight - we stay out of the way rather
than cancelling that attempt with a competing connect().

session_start fires again on every reconnect, so guard
initialize_actionworker(): the action worker holds its own broker
connection that survived the c2s drop, and starting it twice would put
two consumers on the action queue.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QZhYrGpmCvZchZRJ6zJHUM

diff --git a/src/distbot/bot/action_worker.py b/src/distbot/bot/action_worker.py
index 02bffa4624caa82bf14e10501a399c234ff09a8c..7244fb89cb6d78d0c62b86a39457558b38545f74 100644 (file)
--- a/src/distbot/bot/action_worker.py
+++ b/src/distbot/bot/action_worker.py
@@ -36,7 +36,7 @@ class ActionWorker(object):
     async def start(self):
         logger.debug("connecting actionworker to durable queue %s", self._queue_name)
         self._channel = await self._broker.connect_durable_queue(
-            self._queue_name, self.on_message, on_channel_closed=self._bot.disconnect,
+            self._queue_name, self.on_message, on_channel_closed=self._bot.shutdown,
         )
         logger.info("actionworker consuming from %s", self._queue_name)
 

diff --git a/src/distbot/bot/bot.py b/src/distbot/bot/bot.py
index 34bbda063b5c21d914d25b899825d6bf86c99146..8f881edd6c8cf89788094ad58569ce3cf848051d 100644 (file)
--- a/src/distbot/bot/bot.py
+++ b/src/distbot/bot/bot.py
@@ -28,6 +28,15 @@ MUC_SELFPING_TIMEOUT = 10
 # NOT trigger a rejoin, so a flaky s2s link doesn't cause a rejoin storm.
 MUC_NOT_JOINED_CONDITIONS = frozenset({'not-acceptable', 'item-not-found'})
 
+# slixmpp never reconnects on its own: when the TCP connection drops it only
+# fires 'disconnected' and then sits idle forever. We reconnect ourselves, with
+# a back-off so a server that accepts the TCP connection but immediately kills
+# the stream (auth failure, ...) doesn't turn into a hot loop. Once connect()
+# is called slixmpp retries the connection attempt itself, so this delay only
+# separates whole connect cycles.
+RECONNECT_INITIAL_DELAY = 1
+RECONNECT_MAX_DELAY = 300
+
 
 class Bot(slixmpp.ClientXMPP):
     def __init__(self, jid, password, rooms, nick):
@@ -40,6 +49,7 @@ class Bot(slixmpp.ClientXMPP):
         self.rooms = rooms
         self.nick = nick
         self.add_event_handler('session_start', self.session_start)
+        self.add_event_handler('disconnected', self.on_disconnected)
         self.add_event_handler('groupchat_message', self.muc_message)
         self.add_event_handler('message', self.message_handler)
         self.add_event_handler("groupchat_invite", self.group_invite)
@@ -49,6 +59,10 @@ class Bot(slixmpp.ClientXMPP):
         self._initialize_plugins()
         self.muted = False
         self._selfping_task = None
+        self._actionworker_started = False
+        self._shutting_down = False
+        self._reconnect_delay = RECONNECT_INITIAL_DELAY
+        self._reconnect_task = None
 
     def _initialize_plugins(self):
         self.register_plugin('xep_0045')
@@ -58,22 +72,68 @@ class Bot(slixmpp.ClientXMPP):
     def kill_workers(self):
         if self.action_worker:
             self.action_worker.die()
+        self._actionworker_started = False
 
     def initialize_actionworker(self):
+        # session_start fires again after every reconnect, but the action
+        # worker holds its own (still alive) broker connection: starting it
+        # twice would consume the action queue from two consumers.
+        if self._actionworker_started:
+            return
+        self._actionworker_started = True
         asyncio.ensure_future(self.action_worker.start())
 
-    def disconnect(self, reconnect=False, wait=None, send_close=True):
+    def shutdown(self, *_):
+        """Tear everything down and exit for good.
+
+        This must NOT be hooked to disconnect(): slixmpp itself calls
+        disconnect() as the first half of a reconnect (e.g. xep_0199 on a
+        keepalive ping timeout), and killing the process there is how a plain
+        ping timeout used to take the bot down for good.
+        """
         logger.info("Stopping all workers...")
+        self._shutting_down = True
+        if self._reconnect_task is not None:
+            self._reconnect_task.cancel()
         if self._selfping_task is not None:
             self._selfping_task.cancel()
         self.kill_workers()
         logger.info("Stopping self...")
-        super(Bot, self).disconnect(reconnect)
+        self.disconnect()
         logger.info("Gudbai...")
         # TODO: cleaner shutdown?
         raise SystemExit()
 
+    def on_disconnected(self, reason):
+        """Reconnect after losing the c2s stream (slixmpp does not do it)."""
+        if self._shutting_down:
+            return
+        if self._reconnect_task is not None and not self._reconnect_task.done():
+            return
+        logger.warning("disconnected (%s); reconnecting in %ss",
+                       reason, self._reconnect_delay)
+        self._reconnect_task = asyncio.ensure_future(self._reconnect())
+
+    async def _reconnect(self):
+        delay = self._reconnect_delay
+        self._reconnect_delay = min(RECONNECT_MAX_DELAY, self._reconnect_delay * 2)
+        await asyncio.sleep(delay)
+        if self._shutting_down or self.is_connected():
+            return
+        attempt = self._current_connection_attempt
+        if attempt is not None and not attempt.done():
+            # slixmpp is already trying (e.g. its own reconnect() after a
+            # keepalive ping timeout); don't cancel that attempt by starting
+            # a competing one.
+            logger.debug("connection attempt already in flight; not reconnecting")
+            return
+        logger.info("reconnecting...")
+        self.connect()
+
     def session_start(self, _):
+        # A stream that got far enough to start a session is a healthy one:
+        # drop back to the short delay for the next disconnect.
+        self._reconnect_delay = RECONNECT_INITIAL_DELAY
         self.get_roster()
         self.send_presence(ppriority=0, pstatus=None, pshow=None)
         self.initialize_actionworker()
@@ -193,7 +253,7 @@ class Bot(slixmpp.ClientXMPP):
                 self.muted = False
 
             if "hangup" in message:
-                self.disconnect()
+                self.shutdown()
 
             # TODO: probably quite useless
             if "reset-worker-one" in message:
@@ -260,7 +320,7 @@ def run():
     try:
         bot.loop.run_forever()
     except KeyboardInterrupt:
-        bot.disconnect()
+        bot.shutdown()
     finally:
         bot.abort()