From 143aac9d5f5595b5f7f4c874a2a313888eb03cd2 Mon Sep 17 00:00:00 2001 From: Boris Grozev Date: Wed, 19 Aug 2026 11:13:44 -0500 Subject: [PATCH] fix: Handle a visitor node restarting. When the XMPP stream to a visitor node is re-established without being resumed, the node has lost the state of the MUCs that jicofo joined on it. Smack does not re-join them and it keeps reporting them as joined, so jicofo kept sending visitors to a room that it was not an occupant of anymore. Detect this, then discard the visitor room, terminate the visitors that were in it, and disconnect the node. Do not re-use a node whose XMPP connection is down. Do not keep a room that we failed to join. --- .../org/jitsi/jicofo/mock/MockChatRoom.kt | 8 + .../org/jitsi/jicofo/mock/MockXmppProvider.kt | 40 ++++- .../conference/JitsiMeetConference.java | 6 + .../conference/JitsiMeetConferenceImpl.java | 113 ++++++++++--- .../kotlin/org/jitsi/jicofo/FocusManager.kt | 12 ++ .../jitsi/jicofo/conference/ConferenceUtil.kt | 7 +- .../jicofo/xmpp/VisitorConnectionMonitor.kt | 68 ++++++++ .../org/jitsi/jicofo/xmpp/XmppServices.kt | 5 + .../jicofo/conference/ConferenceUtilTest.kt | 95 +++++++++++ .../conference/ConferenceVisitorsTest.kt | 154 ++++++++++++++++++ .../jitsi/jicofo/mock/ConferenceHarness.kt | 48 +++++- .../xmpp/VisitorConnectionMonitorTest.kt | 82 ++++++++++ 12 files changed, 610 insertions(+), 28 deletions(-) create mode 100644 jicofo/src/main/kotlin/org/jitsi/jicofo/xmpp/VisitorConnectionMonitor.kt create mode 100644 jicofo/src/test/kotlin/org/jitsi/jicofo/conference/ConferenceUtilTest.kt create mode 100644 jicofo/src/test/kotlin/org/jitsi/jicofo/conference/ConferenceVisitorsTest.kt create mode 100644 jicofo/src/test/kotlin/org/jitsi/jicofo/xmpp/VisitorConnectionMonitorTest.kt diff --git a/jicofo-common/src/test/kotlin/org/jitsi/jicofo/mock/MockChatRoom.kt b/jicofo-common/src/test/kotlin/org/jitsi/jicofo/mock/MockChatRoom.kt index aa4c037507..33edcc8bc7 100644 --- a/jicofo-common/src/test/kotlin/org/jitsi/jicofo/mock/MockChatRoom.kt +++ b/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,6 +42,9 @@ 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(relaxed = true) { every { addListener(capture(chatRoomListeners)) } returns Unit every { roomJid } returns this@MockChatRoom.roomJid @@ -48,6 +52,10 @@ class MockChatRoom( 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) } } diff --git a/jicofo-common/src/test/kotlin/org/jitsi/jicofo/mock/MockXmppProvider.kt b/jicofo-common/src/test/kotlin/org/jitsi/jicofo/mock/MockXmppProvider.kt index 87daf251e1..57e66fec15 100644 --- a/jicofo-common/src/test/kotlin/org/jitsi/jicofo/mock/MockXmppProvider.kt +++ b/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() + + /** Settable registration state, to simulate the XMPP connection going down and coming back up. */ + var registered = true + val xmppProvider = mockk(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() + + 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 = diff --git a/jicofo/src/main/java/org/jitsi/jicofo/conference/JitsiMeetConference.java b/jicofo/src/main/java/org/jitsi/jicofo/conference/JitsiMeetConference.java index 6cbec96a70..86844ae684 100644 --- a/jicofo/src/main/java/org/jitsi/jicofo/conference/JitsiMeetConference.java +++ b/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); } diff --git a/jicofo/src/main/java/org/jitsi/jicofo/conference/JitsiMeetConferenceImpl.java b/jicofo/src/main/java/org/jitsi/jicofo/conference/JitsiMeetConferenceImpl.java index dc465cb5c3..2dfb4310a3 100644 --- a/jicofo/src/main/java/org/jitsi/jicofo/conference/JitsiMeetConferenceImpl.java +++ b/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 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); } } diff --git a/jicofo/src/main/kotlin/org/jitsi/jicofo/FocusManager.kt b/jicofo/src/main/kotlin/org/jitsi/jicofo/FocusManager.kt index 0835127b36..ba3fdf03f3 100644 --- a/jicofo/src/main/kotlin/org/jitsi/jicofo/FocusManager.kt +++ b/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) + } + } } diff --git a/jicofo/src/main/kotlin/org/jitsi/jicofo/conference/ConferenceUtil.kt b/jicofo/src/main/kotlin/org/jitsi/jicofo/conference/ConferenceUtil.kt index da87be7d62..f9586f9219 100644 --- a/jicofo/src/main/kotlin/org/jitsi/jicofo/conference/ConferenceUtil.kt +++ b/jicofo/src/main/kotlin/org/jitsi/jicofo/conference/ConferenceUtil.kt @@ -60,7 +60,12 @@ internal fun List.getTransport(): IceUdpTransportPacketE } internal fun selectVisitorNode(existingNodes: Map, allNodes: List): 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 } diff --git a/jicofo/src/main/kotlin/org/jitsi/jicofo/xmpp/VisitorConnectionMonitor.kt b/jicofo/src/main/kotlin/org/jitsi/jicofo/xmpp/VisitorConnectionMonitor.kt new file mode 100644 index 0000000000..a8edfd3655 --- /dev/null +++ b/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, + /** 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> = 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) + } +} diff --git a/jicofo/src/main/kotlin/org/jitsi/jicofo/xmpp/XmppServices.kt b/jicofo/src/main/kotlin/org/jitsi/jicofo/xmpp/XmppServices.kt index b59c1575da..cf61ebff36 100644 --- a/jicofo/src/main/kotlin/org/jitsi/jicofo/xmpp/XmppServices.kt +++ b/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 { diff --git a/jicofo/src/test/kotlin/org/jitsi/jicofo/conference/ConferenceUtilTest.kt b/jicofo/src/test/kotlin/org/jitsi/jicofo/conference/ConferenceUtilTest.kt new file mode 100644 index 0000000000..123788b9fb --- /dev/null +++ b/jicofo/src/test/kotlin/org/jitsi/jicofo/conference/ConferenceUtilTest.kt @@ -0,0 +1,95 @@ +/* + * 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.conference + +import io.kotest.core.spec.IsolationMode +import io.kotest.core.spec.style.ShouldSpec +import io.kotest.matchers.collections.shouldContain +import io.kotest.matchers.shouldBe +import org.jitsi.config.withNewConfig +import org.jitsi.jicofo.mock.MockXmppProvider +import org.jitsi.jicofo.xmpp.muc.ChatRoom +import org.jxmpp.jid.impl.JidCreate + +class ConferenceUtilTest : ShouldSpec() { + override fun isolationMode(): IsolationMode = IsolationMode.InstancePerLeaf + + private val v1 = MockXmppProvider(name = "v1", xmppDomain = "v1.example.com") + private val v2 = MockXmppProvider(name = "v2", xmppDomain = "v2.example.com") + private val allNodes = listOf(v1.xmppProvider, v2.xmppProvider) + + /** Create a room on [provider] with [visitors] visitors in it. */ + private fun room(provider: MockXmppProvider, visitors: Int): ChatRoom = + provider.getRoom(JidCreate.entityBareFrom("room@conference.${provider.name}")) + .also { it.visitors = visitors }.chatRoom + + init { + context("selectVisitorNode") { + withNewConfig("jicofo.visitors.max-visitors-per-node = 10") { + context("With no nodes in use") { + should("select one of the registered nodes") { + listOf("v1", "v2") shouldContain selectVisitorNode(emptyMap(), allNodes) + } + should("not select a node whose connection is down") { + v1.registered = false + selectVisitorNode(emptyMap(), allNodes) shouldBe "v2" + } + } + context("With a node already in use") { + val existingNodes = mapOf("v1" to room(v1, 1)) + + should("re-use it while it has capacity") { + selectVisitorNode(existingNodes, allNodes) shouldBe "v1" + } + should("select another node once it is full") { + selectVisitorNode(mapOf("v1" to room(v1, 10)), allNodes) shouldBe "v2" + } + // Without this, jicofo keeps sending visitors to a node that it can not signal to, which is what + // happens while the node is restarting. + should("not re-use it while its connection is down") { + v1.registered = false + selectVisitorNode(existingNodes, allNodes) shouldBe "v2" + } + should("re-use it once its connection is back up") { + v1.registered = false + selectVisitorNode(existingNodes, allNodes) shouldBe "v2" + v1.registered = true + selectVisitorNode(existingNodes, allNodes) shouldBe "v1" + } + should("fall back to it when it is the only node, even if its connection is down") { + v1.registered = false + selectVisitorNode(existingNodes, listOf(v1.xmppProvider)) shouldBe "v1" + } + } + context("With all nodes in use and down") { + should("still select a node") { + v1.registered = false + v2.registered = false + val existingNodes = mapOf("v1" to room(v1, 1), "v2" to room(v2, 1)) + listOf("v1", "v2") shouldContain selectVisitorNode(existingNodes, allNodes) + } + } + context("With no nodes configured") { + should("return null") { + selectVisitorNode(emptyMap(), emptyList()) shouldBe null + } + } + } + } + } +} diff --git a/jicofo/src/test/kotlin/org/jitsi/jicofo/conference/ConferenceVisitorsTest.kt b/jicofo/src/test/kotlin/org/jitsi/jicofo/conference/ConferenceVisitorsTest.kt new file mode 100644 index 0000000000..0b8ae5e355 --- /dev/null +++ b/jicofo/src/test/kotlin/org/jitsi/jicofo/conference/ConferenceVisitorsTest.kt @@ -0,0 +1,154 @@ +/* + * 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.conference + +import io.kotest.assertions.throwables.shouldThrow +import io.kotest.core.spec.IsolationMode +import io.kotest.core.spec.style.ShouldSpec +import io.kotest.core.test.TestCase +import io.kotest.core.test.TestResult +import io.kotest.matchers.collections.shouldBeEmpty +import io.kotest.matchers.collections.shouldContain +import io.kotest.matchers.collections.shouldContainExactly +import io.kotest.matchers.shouldBe +import io.kotest.matchers.shouldNotBe +import io.mockk.every +import io.mockk.verify +import org.jitsi.config.withNewConfig +import org.jitsi.jicofo.TaskPools +import org.jitsi.jicofo.mock.ConferenceHarness +import org.jitsi.jicofo.mock.inPlaceExecutor +import org.jitsi.jicofo.mock.inPlaceScheduledExecutor +import org.jitsi.jicofo.xmpp.muc.ChatRoomInfo +import org.jitsi.jicofo.xmpp.muc.MemberRole + +/** + * Tests the way a conference handles the visitor node that it uses, in particular the node restarting. + */ +class ConferenceVisitorsTest : ShouldSpec() { + override fun isolationMode(): IsolationMode = IsolationMode.InstancePerLeaf + + override suspend fun beforeAny(testCase: TestCase) = super.beforeAny(testCase).also { + TaskPools.ioPool = inPlaceExecutor + TaskPools.scheduledPool = inPlaceScheduledExecutor + } + + override suspend fun afterAny(testCase: TestCase, result: TestResult) = super.afterAny(testCase, result).also { + TaskPools.resetIoPool() + TaskPools.resetScheduledPool() + } + + init { + context("A conference with a visitor node") { + withNewConfig( + """ + jicofo.visitors { + enabled = true + max-participants = 1 + max-visitors-per-node = 100 + require-muc-config-flag = false + auto-enable-broadcast = true + } + jicofo.xmpp.visitors.v1 { + hostname = "v1.example.com" + xmpp-domain = "v1.example.com" + conference-service = "conference.v1.example.com" + } + """ + ) { + val harness = ConferenceHarness(visitorNodeNames = listOf("v1")) + val conference = harness.conference + val visitorRoom = harness.visitorRoom("v1") + every { harness.chatRoom.chatRoom.visitorsEnabled } returns true + + // Two participants in the main room, so that the conference starts inviting. + harness.addParticipants(2) + + context("Redirecting a visitor") { + conference.redirectVisitor(true, null, null) shouldBe "v1" + + should("join the visitor room and connect the node") { + verify(exactly = 1) { visitorRoom.chatRoom.join() } + harness.connectedVnodes() shouldContain "v1" + conference.visitorRoomsJids shouldContainExactly listOf(visitorRoom.roomJid) + } + should("invite the visitors that join the room") { + val visitor = visitorRoom.addMember("visitor-1", MemberRole.VISITOR) + conference.getParticipant(visitor.occupantJid) shouldNotBe null + } + } + + context("When the visitor node is restarted") { + conference.redirectVisitor(true, null, null) shouldBe "v1" + val visitor = visitorRoom.addMember("visitor-1", MemberRole.VISITOR) + conference.getParticipant(visitor.occupantJid) shouldNotBe null + + // The node lost the state of the MUC, so jicofo is not an occupant of it anymore. + conference.visitorConnectionReset("v1") + + should("leave the stale visitor room") { + verify { visitorRoom.chatRoom.leave() } + conference.visitorRoomsJids.shouldBeEmpty() + } + should("disconnect the node") { + harness.disconnectedVnodes() shouldContain "v1" + } + // We do not receive presence for these visitors leaving, so without this each one keeps a + // Participant and an endpoint on a bridge for the rest of the conference. + should("terminate the visitors that were in the stale room") { + conference.getParticipant(visitor.occupantJid) shouldBe null + } + should("join the room again when the next visitor is redirected") { + conference.redirectVisitor(true, null, null) shouldBe "v1" + verify(exactly = 2) { visitorRoom.chatRoom.join() } + conference.visitorRoomsJids shouldContainExactly listOf(visitorRoom.roomJid) + harness.connectedVnodes() shouldContainExactly listOf("v1", "v1") + } + } + + context("When a node that the conference does not use is restarted") { + conference.redirectVisitor(true, null, null) shouldBe "v1" + + should("keep the visitor room") { + conference.visitorConnectionReset("v2") + + harness.disconnectedVnodes().shouldBeEmpty() + conference.visitorRoomsJids shouldContainExactly listOf(visitorRoom.roomJid) + } + } + + context("When joining the visitor room fails") { + every { visitorRoom.chatRoom.join() } throws Exception("Failed to join") + shouldThrow { conference.redirectVisitor(true, null, null) } + + // Otherwise the conference keeps sending visitors to a room that it never joined. + should("not keep the room that it failed to join") { + conference.visitorRoomsJids.shouldBeEmpty() + } + should("try to join again for the next visitor") { + every { visitorRoom.chatRoom.join() } returns ChatRoomInfo(null, null) + + conference.redirectVisitor(true, null, null) shouldBe "v1" + verify(exactly = 2) { visitorRoom.chatRoom.join() } + conference.visitorRoomsJids shouldContainExactly listOf(visitorRoom.roomJid) + } + } + } + } + } +} diff --git a/jicofo/src/test/kotlin/org/jitsi/jicofo/mock/ConferenceHarness.kt b/jicofo/src/test/kotlin/org/jitsi/jicofo/mock/ConferenceHarness.kt index 18d18a04ae..67abc9b346 100644 --- a/jicofo/src/test/kotlin/org/jitsi/jicofo/mock/ConferenceHarness.kt +++ b/jicofo/src/test/kotlin/org/jitsi/jicofo/mock/ConferenceHarness.kt @@ -20,8 +20,12 @@ import io.mockk.every import io.mockk.mockk import org.jitsi.jicofo.conference.JitsiMeetConferenceImpl import org.jitsi.jicofo.conference.Participant +import org.jitsi.jicofo.xmpp.VisitorsManager import org.jitsi.jicofo.xmpp.jingle.JingleSession import org.jitsi.jicofo.xmpp.muc.ChatRoomMember +import org.jitsi.xmpp.extensions.visitors.ConnectVnodePacketExtension +import org.jitsi.xmpp.extensions.visitors.DisconnectVnodePacketExtension +import org.jivesoftware.smack.packet.ExtensionElement import org.jxmpp.jid.impl.JidCreate import java.util.logging.Level @@ -29,13 +33,48 @@ import java.util.logging.Level * Wires up a real [JitsiMeetConferenceImpl] against mock XMPP (colibri2 and Jingle responders, a mock chat room) so * conference-level behavior can be tested without a real XMPP connection or bridge. */ -class ConferenceHarness(roomNameString: String = "test@example.com") { +class ConferenceHarness( + roomNameString: String = "test@example.com", + /** The XMPP domain of the main connection. It must match the domain of [roomNameString]. */ + val xmppDomain: String = "example.com", + /** The names of the visitor nodes to make available to the conference. */ + visitorNodeNames: List = emptyList() +) { val roomName = JidCreate.entityBareFrom(roomNameString) val xmppConnection = ColibriAndJingleXmppConnection() val jingleSessions = mutableListOf() - val xmppProvider = MockXmppProvider(xmppConnection.xmppConnection) + val xmppProvider = MockXmppProvider(xmppConnection.xmppConnection, "client", xmppDomain) val chatRoom = xmppProvider.getRoom(roomName) + /** The mock XMPP connections to the visitor nodes, mapped by node name. */ + val visitorProviders: Map = visitorNodeNames.associateWith { + MockXmppProvider(xmppConnection.xmppConnection, it, "$it.$xmppDomain") + } + + /** The extensions of the VisitorsIqs that the conference sent to the visitors component. */ + val visitorsIqExtensions = mutableListOf>() + + val visitorsManager: VisitorsManager = mockk(relaxed = true) { + every { sendIqToComponent(any(), capture(visitorsIqExtensions)) } returns Unit + every { sendIqToComponentAndGetResponse(any(), capture(visitorsIqExtensions)) } returns null + } + + /** + * The visitor room on the node [node] for this conference. The conference joins this same room, because it looks + * it up by JID with [MockXmppProvider.getRoom]. + */ + fun visitorRoom(node: String): MockChatRoom = visitorProviders[node]!!.getRoom( + JidCreate.entityBareFrom(roomName.toString().replace(xmppDomain, "$node.$xmppDomain")) + ) + + /** The names of the visitor nodes that the conference asked the visitors component to disconnect. */ + fun disconnectedVnodes(): List = visitorsIqExtensions.flatten() + .filterIsInstance().map { it.vnode } + + /** The names of the visitor nodes that the conference asked the visitors component to connect. */ + fun connectedVnodes(): List = visitorsIqExtensions.flatten() + .filterIsInstance().map { it.vnode } + /** Whether the conference has ended (fired conferenceEnded on its listener). */ var ended = false private set @@ -59,6 +98,11 @@ class ConferenceHarness(roomNameString: String = "test@example.com") { every { jingleHandler } returns mockk(relaxed = true) { every { registerSession(capture(jingleSessions)) } returns Unit } + every { visitorConnections } returns visitorProviders.values.map { it.xmppProvider } + every { getXmppVisitorConnectionByName(any()) } answers { + visitorProviders[arg(0)]?.xmppProvider + } + every { visitorsManager } returns this@ConferenceHarness.visitorsManager }, mockk(relaxed = true) { every { selectBridge(any(), any(), any()) } returns mockk(relaxed = true) { diff --git a/jicofo/src/test/kotlin/org/jitsi/jicofo/xmpp/VisitorConnectionMonitorTest.kt b/jicofo/src/test/kotlin/org/jitsi/jicofo/xmpp/VisitorConnectionMonitorTest.kt new file mode 100644 index 0000000000..5ac866bb10 --- /dev/null +++ b/jicofo/src/test/kotlin/org/jitsi/jicofo/xmpp/VisitorConnectionMonitorTest.kt @@ -0,0 +1,82 @@ +/* + * 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 io.kotest.core.spec.IsolationMode +import io.kotest.core.spec.style.ShouldSpec +import io.kotest.core.test.TestCase +import io.kotest.core.test.TestResult +import io.kotest.matchers.collections.shouldBeEmpty +import io.kotest.matchers.collections.shouldContainExactly +import org.jitsi.jicofo.TaskPools +import org.jitsi.jicofo.mock.MockXmppConnection +import org.jitsi.jicofo.mock.MockXmppProvider +import org.jitsi.jicofo.mock.inPlaceExecutor + +class VisitorConnectionMonitorTest : ShouldSpec() { + override fun isolationMode(): IsolationMode = IsolationMode.InstancePerLeaf + + override suspend fun beforeAny(testCase: TestCase) = super.beforeAny(testCase).also { + TaskPools.ioPool = inPlaceExecutor + } + + override suspend fun afterAny(testCase: TestCase, result: TestResult) = super.afterAny(testCase, result).also { + TaskPools.resetIoPool() + } + + init { + val v1 = MockXmppProvider(MockXmppConnection().xmppConnection, "v1") + val v2 = MockXmppProvider(MockXmppConnection().xmppConnection, "v2") + val resetNodes = mutableListOf() + val monitor = VisitorConnectionMonitor(listOf(v1.xmppProvider, v2.xmppProvider)) { resetNodes.add(it) } + + context("When a visitor connection authenticates") { + should("not report a reset if the stream was resumed") { + v1.authenticated(resumed = true) + + resetNodes.shouldBeEmpty() + } + should("report a reset if the stream was not resumed") { + v1.authenticated(resumed = false) + + resetNodes shouldContainExactly listOf("v1") + } + should("report a reset for the node that reconnected only") { + v2.authenticated(resumed = false) + v1.authenticated(resumed = true) + + resetNodes shouldContainExactly listOf("v2") + } + should("report every reset") { + v1.authenticated(resumed = false) + v2.authenticated(resumed = false) + v1.authenticated(resumed = false) + + resetNodes shouldContainExactly listOf("v1", "v2", "v1") + } + } + context("After shutdown") { + should("not report a reset") { + monitor.shutdown() + v1.authenticated(resumed = false) + + resetNodes.shouldBeEmpty() + } + } + } +}