Add XEP-0410 MUC self-ping to auto-rejoin after silent eviction

A c2s reconnect re-fires session_start and rejoins the rooms, but a
silent MUC eviction (e.g. an s2s flap when the account server restarts)
never touches the c2s stream: the bot stays online and JID-pingable yet
quietly drops out of the room with no way to notice.

Periodically ping our own occupant JID and rejoin on not-acceptable /
item-not-found; treat every other error (and timeouts) as inconclusive
so a flaky s2s link can't cause a rejoin storm.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

diff --git a/src/distbot/bot/bot.py b/src/distbot/bot/bot.py
index 39ce429633121883250d718888c21fdfb4fe832c..a4dcb04248dd5cdd217ade963d7ca93f447b1b88 100644 (file)
--- a/src/distbot/bot/bot.py
+++ b/src/distbot/bot/bot.py
@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
 # -*- coding: utf-8 -*-

+import asyncio

 import json
 import logging
 import re
 import json
 import logging
 import re


@@ -8,6 +9,7 @@ import pika
 from pika.adapters.asyncio_connection import AsyncioConnection
 
 import slixmpp
 from pika.adapters.asyncio_connection import AsyncioConnection
 
 import slixmpp

+from slixmpp.exceptions import IqError, IqTimeout

 from slixmpp.stanza import Message
 
 from distbot.bot import action_worker
 from slixmpp.stanza import Message
 
 from distbot.bot import action_worker


@@ -20,6 +22,15 @@ logging.getLogger("pika").setLevel(logging.WARN)
 logging.getLogger('slixmpp').setLevel(logging.DEBUG)
 logging.getLogger('slixmpp.xmlstream.xmlstream').setLevel(logging.DEBUG)
 
 logging.getLogger('slixmpp').setLevel(logging.DEBUG)
 logging.getLogger('slixmpp.xmlstream.xmlstream').setLevel(logging.DEBUG)
 

+# XEP-0410: how often to verify MUC membership, and the ping timeout (seconds).
+MUC_SELFPING_INTERVAL = 60
+MUC_SELFPING_TIMEOUT = 10
+# Error conditions from a MUC self-ping that mean "you are no longer joined".
+# Everything else (service-unavailable, feature-not-implemented,
+# remote-server-not-found/-timeout, an IqTimeout, ...) is inconclusive and must
+# 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'})
+

 
 class Bot(slixmpp.ClientXMPP):
     def __init__(self, jid, password, rooms, nick):
 
 class Bot(slixmpp.ClientXMPP):
     def __init__(self, jid, password, rooms, nick):


@@ -40,6 +51,7 @@ class Bot(slixmpp.ClientXMPP):
 
         self._initialize_plugins()
         self.muted = False
 
         self._initialize_plugins()
         self.muted = False

+        self._selfping_task = None

 
     def _initialize_plugins(self):
         self.register_plugin('xep_0045')
 
     def _initialize_plugins(self):
         self.register_plugin('xep_0045')


@@ -61,6 +73,8 @@ class Bot(slixmpp.ClientXMPP):
 
     def disconnect(self, reconnect=False, wait=None, send_close=True):
         logger.info("Stopping all workers...")
 
     def disconnect(self, reconnect=False, wait=None, send_close=True):
         logger.info("Stopping all workers...")

+        if self._selfping_task is not None:
+            self._selfping_task.cancel()

         self.kill_workers()
         logger.info("Stopping self...")
         super(Bot, self).disconnect(reconnect)
         self.kill_workers()
         logger.info("Stopping self...")
         super(Bot, self).disconnect(reconnect)


@@ -77,6 +91,41 @@ class Bot(slixmpp.ClientXMPP):
             ret = self.plugin['xep_0045'].join_muc(room, self.nick)
             logger.info('%s: joined with code %s' % (room, ret))
 
             ret = self.plugin['xep_0045'].join_muc(room, self.nick)
             logger.info('%s: joined with code %s' % (room, ret))
 

+        # Start the XEP-0410 self-ping loop once. It survives across c2s
+        # reconnects, and — unlike session_start — also recovers from a silent
+        # MUC eviction (e.g. an s2s flap) that never touches the c2s stream.
+        if self._selfping_task is None or self._selfping_task.done():
+            self._selfping_task = asyncio.ensure_future(self._muc_selfping_loop())
+
+    async def _muc_selfping_loop(self):
+        """XEP-0410: periodically confirm we are still joined, rejoin if not."""
+        while True:
+            try:
+                await asyncio.sleep(MUC_SELFPING_INTERVAL)
+                for room in self.rooms:
+                    await self._muc_selfping(room)
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:  # never let the loop die
+                logger.exception("MUC self-ping loop error: %s", e)
+
+    async def _muc_selfping(self, room):
+        """Ping our own occupant JID; rejoin if the MUC says we're not in it."""
+        occupant = '%s/%s' % (room, self.nick)
+        try:
+            await self.plugin['xep_0199'].ping(jid=occupant, timeout=MUC_SELFPING_TIMEOUT)
+            # An IQ result means the MUC routed the ping to us: still joined.
+        except IqError as e:
+            if e.condition in MUC_NOT_JOINED_CONDITIONS:
+                logger.warning("%s: self-ping says not joined (%s); rejoining",
+                               room, e.condition)
+                self.plugin['xep_0045'].join_muc(room, self.nick)
+            else:
+                logger.debug("%s: self-ping inconclusive (%s); not rejoining",
+                             room, e.condition)
+        except IqTimeout:
+            logger.debug("%s: self-ping timed out; assuming still joined", room)
+

     def muc_message(self, msg: Message):
         if msg['mucnick'] == self.nick or 'groupchat' != msg['type']:
             return False
     def muc_message(self, msg: Message):
         if msg['mucnick'] == self.nick or 'groupchat' != msg['type']:
             return False