fix: Handle a visitor node restarting.

jicofo-common/src/test/kotlin/org/jitsi/jicofo/mock/MockChatRoom.kt

@@ -21,6 +21,7 @@ import io.mockk.mockk
 import org.jitsi.jicofo.xmpp.Features
 import org.jitsi.jicofo.xmpp.XmppProvider
 import org.jitsi.jicofo.xmpp.muc.ChatRoom
+import org.jitsi.jicofo.xmpp.muc.ChatRoomInfo
 import org.jitsi.jicofo.xmpp.muc.ChatRoomListener
 import org.jitsi.jicofo.xmpp.muc.ChatRoomMember
 import org.jitsi.jicofo.xmpp.muc.MemberRole
@@ -41,13 +42,20 @@ class MockChatRoom(
     var audioSenders = 0
     var videoSenders = 0
 
+    /** Settable visitor count (the real ChatRoom derives it from member presence). */
+    var visitors = 0
+
     val chatRoom = mockk<ChatRoom>(relaxed = true) {
         every { addListener(capture(chatRoomListeners)) } returns Unit
         every { roomJid } returns this@MockChatRoom.roomJid
         every { members } returns memberList
         every { memberCount } answers { memberList.size }
         every { audioSendersCount } answers { audioSenders }
         every { videoSendersCount } answers { videoSenders }
+        every { visitorCount } answers { visitors }
+        // Without this a relaxed mock returns a ChatRoomInfo with a non-null mainRoomJid, i.e. the room looks like a
+        // breakout room. Let jicofo generate the meeting ID, as it does when the MUC does not advertise one.
+        every { join() } returns ChatRoomInfo(meetingId = null, mainRoomJid = null)
         every { xmppProvider } returns this@MockChatRoom.xmppProvider
         every { debugState } returns JsonNodeFactory.instance.objectNode()
         every { getChatMember(any()) } answers { memberList.find { it.occupantJid == arg(0) } }

jicofo-common/src/test/kotlin/org/jitsi/jicofo/mock/MockXmppProvider.kt

@@ -19,14 +19,50 @@ import io.mockk.every
 import io.mockk.mockk
 import org.jitsi.jicofo.xmpp.XmppProvider
 import org.jivesoftware.smack.AbstractXMPPConnection
+import org.jivesoftware.smack.ConnectionListener
 import org.jxmpp.jid.EntityBareJid
+import org.jxmpp.jid.impl.JidCreate
 
-class MockXmppProvider(val xmppConnection: AbstractXMPPConnection = MockXmppConnection().xmppConnection) {
+class MockXmppProvider(
+    val xmppConnection: AbstractXMPPConnection = MockXmppConnection().xmppConnection,
+    /** The name of the connection, as it appears in [XmppProvider.getConfig]. */
+    val name: String = "mock",
+    /** The XMPP domain of the connection, needed to map a main room JID to a visitor room JID. */
+    xmppDomain: String? = null
+) {
     val chatRooms = mutableMapOf<EntityBareJid, MockChatRoom>()
+
+    /** Settable registration state, to simulate the XMPP connection going down and coming back up. */
+    var registered = true
+
     val xmppProvider = mockk<XmppProvider>(relaxed = true) {
-        every { registered } returns true
+        every { registered } answers { this@MockXmppProvider.registered }
         every { findOrCreateRoom(any(), any()) } answers { getRoom(arg(0)).chatRoom }
         every { xmppConnection } returns this@MockXmppProvider.xmppConnection
+        every { config } returns mockk(relaxed = true) {
+            every { this@mockk.name } returns this@MockXmppProvider.name
+            every { this@mockk.xmppDomain } returns xmppDomain?.let { JidCreate.domainBareFrom(it) }
+        }
+    }
+
+    /**
+     * The Smack connection listeners that were registered on [xmppConnection]. Note that when instances share an
+     * [xmppConnection] only the instance that was created last captures the listeners.
+     */
+    val connectionListeners = mutableListOf<ConnectionListener>()
+
+    init {
+        every { xmppConnection.addConnectionListener(capture(connectionListeners)) } returns Unit
+        every { xmppConnection.removeConnectionListener(any()) } answers {
+            connectionListeners.remove(arg(0))
+            Unit
+        }
+    }
+
+    /** Simulate the connection authenticating, i.e. coming up or coming back up after a disconnect. */
+    fun authenticated(resumed: Boolean) {
+        registered = true
+        connectionListeners.toList().forEach { it.authenticated(xmppConnection, resumed) }
     }
 
     fun getRoom(jid: EntityBareJid): MockChatRoom =

jicofo/src/main/java/org/jitsi/jicofo/conference/JitsiMeetConference.java

@@ -179,5 +179,11 @@ MuteResult handleMuteRequest(
     String redirectVisitor(boolean visitorRequested, @Nullable String userId, @Nullable String groupId)
             throws Exception;
 
+    /**
+     * Notify this conference that the XMPP stream to the visitor node {@code node} was re-established without being
+     * resumed, so the MUC that this conference joined on that node (if any) is not joined anymore.
+     */
+    void visitorConnectionReset(@NotNull String node);
+
     void setPresenceExtension(@NotNull ExtensionElement extension);
 }

jicofo/src/main/java/org/jitsi/jicofo/conference/JitsiMeetConferenceImpl.java

@@ -2118,7 +2118,37 @@ private String selectVisitorNode()
             chatRoomToJoin.visitorInvited();
         }
 
-        chatRoomToJoin.join();
+        try
+        {
+            chatRoomToJoin.join();
+        }
+        catch (Exception e)
+        {
+            // Do not keep a room that we failed to join. We would never join it, and we would keep sending visitors
+            // to it because of the check above.
+            logger.error("Failed to join the visitor room on node " + node, e);
+            synchronized (visitorChatRooms)
+            {
+                visitorChatRooms.remove(node, chatRoomToJoin);
+            }
+            chatRoomToJoin.removeAllListeners();
+            chatRoomToJoin.leave();
+            throw e;
+        }
+
+        // The room may have been discarded while we were joining it, for example because the node was restarted.
+        // Leave it, otherwise we stay in a MUC that we do not track anymore.
+        synchronized (visitorChatRooms)
+        {
+            if (visitorChatRooms.get(node) != chatRoomToJoin)
+            {
+                logger.warn("The visitor room on node " + node + " was discarded while we were joining it.");
+                chatRoomToJoin.removeAllListeners();
+                chatRoomToJoin.leave();
+                return null;
+            }
+        }
+
         Collection<ExtensionElement> presenceExtensions = new ArrayList<>();
 
         ComponentVersionsExtension versionsExtension = new ComponentVersionsExtension();
@@ -2149,6 +2179,63 @@ private String selectVisitorNode()
         return node;
     }
 
+    @Override
+    public void visitorConnectionReset(@NotNull String node)
+    {
+        final ChatRoom staleChatRoom;
+        synchronized (visitorChatRooms)
+        {
+            staleChatRoom = visitorChatRooms.remove(node);
+        }
+
+        if (staleChatRoom == null)
+        {
+            // This conference does not use the node.
+            return;
+        }
+
+        logger.info("The connection to visitor node " + node + " was reset, discarding the visitor room.");
+        discardVisitorChatRoom(node, staleChatRoom);
+    }
+
+    /**
+     * Clean up after a visitor {@link ChatRoom} that we are not joined in anymore. Terminate the visitors that were
+     * in the room, leave the room, and tell the visitors component to stop sending visitors to the node.
+     *
+     * The caller must remove the room from {@link #visitorChatRooms} first.
+     *
+     * @param node the ID of the visitor node that the room is on.
+     * @param staleChatRoom the room to discard.
+     */
+    private void discardVisitorChatRoom(@NotNull String node, @NotNull ChatRoom staleChatRoom)
+    {
+        TaskPools.getIoPool().submit(() ->
+        {
+            try
+            {
+                // We are not in the room anymore, so we do not receive presence for the visitors leaving it. Without
+                // this they leak, because each one keeps a Participant and an endpoint on a bridge.
+                for (ChatRoomMember member : staleChatRoom.getMembers())
+                {
+                    if (member.getRole() == MemberRole.VISITOR)
+                    {
+                        onMemberLeft(member);
+                    }
+                }
+
+                staleChatRoom.removeAllListeners();
+                staleChatRoom.leave();
+            }
+            catch (Exception e)
+            {
+                logger.error("Failed to discard the visitor room on node " + node, e);
+            }
+        });
+
+        xmppServices.getVisitorsManager().sendIqToComponent(
+                roomName, Collections.singletonList(new DisconnectVnodePacketExtension(node)));
+    }
+
     private void onBridgeUp(Jid bridgeJid)
     {
         // Check if we're not shutting down
@@ -2666,7 +2753,6 @@ private VisitorChatRoomListenerImpl(ChatRoom chatRoom)
         public void roomDestroyed(String reason)
         {
             logger.info("Visitor room destroyed with reason=" + reason);
-            ChatRoom chatRoomToLeave = null;
             String vnode = null;
             synchronized (visitorChatRooms)
             {
@@ -2675,33 +2761,14 @@ public void roomDestroyed(String reason)
                             .filter(e -> e.getValue() == chatRoom).findFirst().orElse(null);
                 if (entry != null)
                 {
-                    chatRoomToLeave = entry.getValue();
                     vnode = entry.getKey();
                     visitorChatRooms.remove(vnode);
                 }
             }
 
-            if (chatRoomToLeave != null)
+            if (vnode != null)
             {
-                ChatRoom finalChatRoom = chatRoomToLeave;
-                TaskPools.getIoPool().submit(() ->
-                {
-                    try
-                    {
-                        logger.info("Removing visitor chat room");
-                        finalChatRoom.leave();
-                    }
-                    catch (Exception e)
-                    {
-                        logger.warn("Error while leaving chat room.", e);
-                    }
-                });
-
-                if (vnode != null)
-                {
-                    xmppServices.getVisitorsManager().sendIqToComponent(
-                            roomName, Collections.singletonList(new DisconnectVnodePacketExtension(vnode)));
-                }
+                discardVisitorChatRoom(vnode, chatRoom);
             }
         }
 

jicofo/src/main/kotlin/org/jitsi/jicofo/FocusManager.kt

@@ -384,4 +384,16 @@ class FocusManager(
     override fun registrationChanged(registered: Boolean) {
         conferencesCache.forEach { it.registrationChanged(registered) }
     }
+
+    /**
+     * Notify the conferences that the XMPP stream to the visitor node [node] was re-established without being
+     * resumed. The MUCs that jicofo joined on that node are not joined anymore.
+     */
+    fun visitorConnectionReset(node: String) = conferencesCache.forEach {
+        try {
+            it.visitorConnectionReset(node)
+        } catch (e: Exception) {
+            logger.error("Failed to reset visitor node $node for conference ${it.roomName}", e)
+        }
+    }
 }

jicofo/src/main/kotlin/org/jitsi/jicofo/conference/ConferenceUtil.kt

@@ -60,7 +60,12 @@ internal fun List<ContentPacketExtension>.getTransport(): IceUdpTransportPacketE
 }
 
 internal fun selectVisitorNode(existingNodes: Map<String, ChatRoom>, allNodes: List<XmppProvider>): String? {
-    val min = existingNodes.minByOrNull { it.value.visitorCount }
+    val registeredNodeNames = allNodes.filter { it.registered }.map { it.config.name }.toSet()
+
+    // Re-use a node that we already have a room on, if it has capacity. Skip a node whose XMPP connection is down,
+    // because we can not signal to the visitors that we send there.
+    val min = existingNodes.filterKeys { registeredNodeNames.contains(it) }
+        .minByOrNull { it.value.visitorCount }
     if (min != null && min.value.visitorCount < VisitorsConfig.config.maxVisitorsPerNode) {
         return min.key
     }

jicofo/src/main/kotlin/org/jitsi/jicofo/xmpp/VisitorConnectionMonitor.kt

@@ -0,0 +1,68 @@
+/*
+ * Jicofo, the Jitsi Conference Focus.
+ *
+ * Copyright @ 2026 - present 8x8, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.jitsi.jicofo.xmpp
+
+import org.jitsi.jicofo.TaskPools
+import org.jitsi.utils.logging2.createLogger
+import org.jivesoftware.smack.ConnectionListener
+import org.jivesoftware.smack.XMPPConnection
+
+/**
+ * Monitors the XMPP connections to the visitor nodes.
+ *
+ * A visitor node keeps the state of the MUCs that jicofo joined on it. This state is lost when the XMPP stream is
+ * not resumed, for example because the visitor node restarted. Jicofo is not an occupant of these MUCs anymore, so
+ * it does not receive presence from them. The state that jicofo keeps for them is stale and it must be discarded.
+ *
+ * Smack does not re-join the MUCs and it does not change its own state, so [onConnectionReset] is the only
+ * notification that this happened.
+ */
+class VisitorConnectionMonitor(
+    visitorConnections: List<XmppProvider>,
+    /** Called with the name of a visitor node whose XMPP stream was re-established without being resumed. */
+    private val onConnectionReset: (String) -> Unit
+) {
+    private val logger = createLogger()
+
+    private val connectionListeners: List<Pair<XmppProvider, ConnectionListener>> = visitorConnections.map { provider ->
+        val name = provider.config.name
+        val listener = object : ConnectionListener {
+            override fun authenticated(connection: XMPPConnection?, resumed: Boolean) {
+                if (resumed) {
+                    // The visitor node kept our session, so the MUCs that we joined on it are still joined.
+                    return
+                }
+                logger.info("The XMPP stream to visitor node $name was not resumed.")
+                // Do not do the work in Smack's thread.
+                TaskPools.ioPool.submit {
+                    try {
+                        onConnectionReset(name)
+                    } catch (e: Throwable) {
+                        logger.error("Failed to handle a connection reset for visitor node $name", e)
+                    }
+                }
+            }
+        }
+        provider.xmppConnection.addConnectionListener(listener)
+        provider to listener
+    }
+
+    fun shutdown() = connectionListeners.forEach { (provider, listener) ->
+        provider.xmppConnection.removeConnectionListener(listener)
+    }
+}

jicofo/src/main/kotlin/org/jitsi/jicofo/xmpp/XmppServices.kt

@@ -59,6 +59,10 @@ class XmppServices(
         }
     }
 
+    private val visitorConnectionMonitor = VisitorConnectionMonitor(visitorConnections) { node ->
+        focusManager.visitorConnectionReset(node)
+    }
+
     fun getXmppConnectionByName(name: XmppConnectionEnum) = when (name) {
         XmppConnectionEnum.Client -> clientConnection
         XmppConnectionEnum.Service -> serviceConnection
@@ -143,6 +147,7 @@ class XmppServices(
         avModerationHandler.shutdown()
         roomMetadataHandler.shutdown()
         jingleHandler.shutdown()
+        visitorConnectionMonitor.shutdown()
 
         clientConnection.xmppConnection.unregisterIQRequestHandler(conferenceIqHandler)
         authenticationIqHandler?.let {