Fix local endpoint count (prevent premature shutdown). (#1906)

* fix: Do not call endpointExpired() for RelayedEndpoints. * ref: Rename endpointExpired to localEndpointExpired. * fix: Log, do not shutdown if endpoint count is invalid. * ref: Rename localEndpointExpired to endpointExpired. * fix: Call Relay.endpointExpired for Relayed endpoints.
jitsi · Jun 6, 2022 · 340062a · 340062a
1 parent a9b07d4
commit 340062a
Show file tree

Hide file tree

Showing 4 changed files with 25 additions and 18 deletions.
diff --git a/jvb/src/main/java/org/jitsi/videobridge/Conference.java b/jvb/src/main/java/org/jitsi/videobridge/Conference.java
@@ -753,7 +753,7 @@ public Endpoint createLocalEndpoint(String id, boolean iceControlling, boolean s
         }
 
         final Endpoint endpoint = new Endpoint(id, this, logger, iceControlling, sourceNames);
-        videobridge.endpointCreated();
+        videobridge.localEndpointCreated();
 
         subscribeToEndpointEvents(endpoint);
 
@@ -996,22 +996,23 @@ public void endpointExpired(AbstractEndpoint endpoint)
         final AbstractEndpoint removedEndpoint;
         String id = endpoint.getId();
         removedEndpoint = endpointsById.remove(id);
-        if (removedEndpoint != null)
+        if (removedEndpoint == null)
         {
-            updateEndpointsCache();
+            logger.warn("No endpoint found, id=" + id);
+            return;
         }
 
-        endpointsById.forEach((i, senderEndpoint) -> senderEndpoint.removeReceiver(id));
-
-        endpoint.getSsrcs().forEach(ssrc -> endpointsBySsrc.remove(ssrc, endpoint));
-
-        relaysById.forEach((i, relay) -> relay.localEndpointExpired(id));
-
-        if (removedEndpoint != null)
+        if (removedEndpoint instanceof Endpoint)
         {
-            endpointsChanged();
-            videobridge.endpointExpired();
+            // The removed endpoint was a local Endpoint as opposed to a RelayedEndpoint.
+            updateEndpointsCache();
+            endpointsById.forEach((i, senderEndpoint) -> senderEndpoint.removeReceiver(id));
+            videobridge.localEndpointExpired();
         }
+
+        relaysById.forEach((i, relay) -> relay.endpointExpired(id));
+        endpoint.getSsrcs().forEach(ssrc -> endpointsBySsrc.remove(ssrc, endpoint));
+        endpointsChanged();
     }
 
     /**

diff --git a/jvb/src/main/java/org/jitsi/videobridge/Videobridge.java b/jvb/src/main/java/org/jitsi/videobridge/Videobridge.java
@@ -257,14 +257,20 @@ public JvbHealthChecker getJvbHealthChecker()
         return conference;
     }
 
-    void endpointCreated()
+    void localEndpointCreated()
     {
         statistics.currentLocalEndpoints.incrementAndGet();
     }
 
-    void endpointExpired()
+    void localEndpointExpired()
     {
-        shutdownManager.maybeShutdown(statistics.currentLocalEndpoints.decrementAndGet());
+        long remainingEndpoints = statistics.currentLocalEndpoints.decrementAndGet();
+        if (remainingEndpoints < 0)
+        {
+            logger.warn("Invalid endpoint count " + remainingEndpoints + ". Disabling endpoint-count based shutdown!");
+            return;
+        }
+        shutdownManager.maybeShutdown(remainingEndpoints);
     }
 
     /**

diff --git a/jvb/src/main/kotlin/org/jitsi/videobridge/relay/Relay.kt b/jvb/src/main/kotlin/org/jitsi/videobridge/relay/Relay.kt
@@ -793,7 +793,7 @@ class Relay @JvmOverloads constructor(
         }
     }
 
-    fun localEndpointExpired(id: String) {
+    fun endpointExpired(id: String) {
         val s = senders.remove(id)
         s?.expire()
     }

diff --git a/jvb/src/test/kotlin/org/jitsi/videobridge/VideobridgeTest.kt b/jvb/src/test/kotlin/org/jitsi/videobridge/VideobridgeTest.kt
@@ -57,7 +57,7 @@ class VideobridgeTest : ShouldSpec() {
         context("Shutdown") {
             context("when a conference is active") {
                 withNewConfig("videobridge.shutdown.graceful-shutdown-min-participants=10") {
-                    repeat(15) { videobridge.endpointCreated() }
+                    repeat(15) { videobridge.localEndpointCreated() }
                     context("starting a graceful shutdown") {
                         videobridge.shutdown(true)
                         should("report that shutdown is in progress") {
@@ -74,7 +74,7 @@ class VideobridgeTest : ShouldSpec() {
                             resp.error.condition shouldBe StanzaError.Condition.service_unavailable
                         }
                         context("When the number of participants drops below the threshold") {
-                            repeat(10) { videobridge.endpointExpired() }
+                            repeat(10) { videobridge.localEndpointExpired() }
                             videobridge.shutdownState shouldBe ShutdownState.SHUTTING_DOWN
                             fakeExecutor.clock.elapse(ShutdownConfig.config.shuttingDownDelay)
                             fakeExecutor.run()