From: Thorsten Date: Tue, 1 Sep 2026 19:27:46 +0000 (+0200) Subject: Fix a keepalive ping timeout killing the bot for good X-Git-Url: https://git.aero2k.de/?a=commitdiff_plain;h=5e4eb96f030ec358ab08a5e5ba239cea98f9b820;p=urlbot-v3.git 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 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 02bffa4..7244fb8 100644 --- 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 34bbda0..8f881ed 100644 --- 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()