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 · Jul 11, 2022 · b4cd41d · b4cd41d
1 parent c8c6e8e
commit b4cd41d
Show file tree

Hide file tree

Showing 4 changed files with 28 additions and 11 deletions.
diff --git a/jvb/src/main/java/org/jitsi/videobridge/Conference.java b/jvb/src/main/java/org/jitsi/videobridge/Conference.java
@@ -840,7 +840,7 @@ else if (existingEndpoint != null)
         }
 
         final Endpoint endpoint = new Endpoint(id, this, logger, iceControlling, sourceNames);
-        videobridge.endpointCreated();
+        videobridge.localEndpointCreated();
 
         subscribeToEndpointEvents(endpoint);
 
@@ -1084,11 +1084,13 @@ 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;
         }
 
+<<<<<<< HEAD
         endpointsById.forEach((i, senderEndpoint) -> senderEndpoint.removeReceiver(id));
 
         endpoint.getSsrcs().forEach(ssrc -> endpointsBySsrc.remove(ssrc, endpoint));
@@ -1101,10 +1103,19 @@ public void endpointExpired(AbstractEndpoint endpoint)
         relaysById.forEach((i, relay) -> relay.localEndpointExpired(id));
 
         if (removedEndpoint != null)
+=======
+        if (removedEndpoint instanceof Endpoint)
+>>>>>>> 340062a5b (Fix local endpoint count (prevent premature shutdown). (#1906))
         {
-            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
@@ -258,14 +258,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
@@ -794,7 +794,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
@@ -58,7 +58,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") {
@@ -75,7 +75,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
                             mockExecutor.clock.elapse(ShutdownConfig.config.shuttingDownDelay)
                             mockExecutor.run()