feat: Limit the rate at which a conference adds endpoints to a bridge

jicofo-selector/src/main/kotlin/org/jitsi/jicofo/bridge/BridgeConfig.kt

@@ -37,6 +37,19 @@ class BridgeConfig private constructor() {
         "$BASE.max-bridge-participants".from(JitsiConfig.newConfig)
     }
 
+    /**
+     * The maximum number of participants that a single conference may add to a single bridge within
+     * [maxBridgeParticipantsInterval]. Use -1 to disable.
+     */
+    val maxBridgeParticipantsPerInterval: Int by config {
+        "$BASE.max-bridge-participants-per-interval".from(JitsiConfig.newConfig)
+    }
+
+    /** The interval over which [maxBridgeParticipantsPerInterval] is enforced. */
+    val maxBridgeParticipantsInterval: Duration by config {
+        "$BASE.max-bridge-participants-interval".from(JitsiConfig.newConfig)
+    }
+
     val averageParticipantStress: Double by config {
         "$BASE.average-participant-stress".from(JitsiConfig.newConfig)
     }

jicofo-selector/src/main/kotlin/org/jitsi/jicofo/bridge/BridgeMetrics.kt

@@ -24,5 +24,10 @@ class BridgeMetrics {
             "Total number of endpoints moved away from a bridge for automatic load redistribution.",
             labelNames = listOf("jvb")
         )
+        val rateLimited = metricsContainer.registerCounter(
+            "bridge_selection_rate_limited",
+            "Total number of times a bridge was considered overloaded for a conference because the conference had " +
+                "reached max-bridge-participants-per-interval on it."
+        )
     }
 }

jicofo-selector/src/main/kotlin/org/jitsi/jicofo/bridge/BridgeSelectionStrategy.kt

@@ -42,6 +42,7 @@ abstract class BridgeSelectionStrategy {
         participantProperties: ParticipantProperties,
         allowMultiBridge: Boolean
     ): Bridge? {
+        logRateLimitedBridges(conferenceBridges)
         return if (conferenceBridges.isEmpty()) {
             val bridge = doSelect(bridges, conferenceBridges, participantProperties)
             if (bridge != null) {
@@ -319,7 +320,9 @@ abstract class BridgeSelectionStrategy {
      * @return `true` if the bridge should be considered overloaded.
      */
     private fun Bridge.isOverloaded(conferenceBridges: Map<Bridge, ConferenceBridgeProperties>): Boolean {
-        return isOverloaded || hasMaxParticipantsInConference(conferenceBridges)
+        return isOverloaded ||
+            hasMaxParticipantsInConference(conferenceBridges) ||
+            hasMaxRecentParticipantsInConference(conferenceBridges)
     }
 
     private fun Bridge.hasMaxParticipantsInConference(
@@ -329,4 +332,38 @@ abstract class BridgeSelectionStrategy {
             conferenceBridges.containsKey(this) &&
             conferenceBridges[this]!!.participantCount >= config.maxBridgeParticipants
     }
+
+    /**
+     * Whether the conference has recently added too many endpoints to this bridge, i.e. it is growing on this bridge
+     * faster than the bridge's stress reports can keep up with. Note that this is specific to the conference, and does
+     * not affect the way the bridge is treated for other conferences.
+     */
+    private fun Bridge.hasMaxRecentParticipantsInConference(
+        conferenceBridges: Map<Bridge, ConferenceBridgeProperties>
+    ): Boolean {
+        return config.maxBridgeParticipantsPerInterval > 0 &&
+            (conferenceBridges[this]?.recentlyAddedParticipantCount ?: 0) >=
+            config.maxBridgeParticipantsPerInterval
+    }
+
+    /**
+     * Log (and count) the bridges on which the conference has hit the [config.maxBridgeParticipantsPerInterval] limit.
+     * This is only for visibility, the limit itself is enforced in [hasMaxRecentParticipantsInConference].
+     */
+    private fun logRateLimitedBridges(conferenceBridges: Map<Bridge, ConferenceBridgeProperties>) {
+        if (config.maxBridgeParticipantsPerInterval <= 0) return
+        val rateLimited = conferenceBridges.filterValues {
+            it.recentlyAddedParticipantCount >= config.maxBridgeParticipantsPerInterval
+        }
+        if (rateLimited.isNotEmpty()) {
+            BridgeMetrics.rateLimited.inc()
+            logger.info(
+                "The conference has recently added too many endpoints to these bridges, they will not be used for " +
+                    "this participant: " +
+                    rateLimited.entries.joinToString {
+                        "${it.key}=${it.value.recentlyAddedParticipantCount}"
+                    }
+            )
+        }
+    }
 }

jicofo-selector/src/main/kotlin/org/jitsi/jicofo/bridge/BridgeSelector.kt

@@ -288,7 +288,12 @@ class BridgeSelector @JvmOverloads constructor(
 
 data class ConferenceBridgeProperties(
     val participantCount: Int,
-    val visitor: Boolean = false
+    val visitor: Boolean = false,
+    /**
+     * The number of endpoints that this conference has added to this bridge within
+     * [BridgeConfig.maxBridgeParticipantsInterval].
+     */
+    val recentlyAddedParticipantCount: Long = 0
 )
 
 data class ParticipantProperties(

jicofo-selector/src/main/kotlin/org/jitsi/jicofo/bridge/colibri/Colibri2Session.kt

@@ -23,6 +23,7 @@ import io.opentelemetry.api.trace.Span
 import io.opentelemetry.context.Context
 import org.jitsi.jicofo.OctoConfig
 import org.jitsi.jicofo.bridge.Bridge
+import org.jitsi.jicofo.bridge.BridgeConfig
 import org.jitsi.jicofo.bridge.CascadeLink
 import org.jitsi.jicofo.bridge.CascadeNode
 import org.jitsi.jicofo.codec.CodecUtil
@@ -32,6 +33,7 @@ import org.jitsi.jicofo.conference.source.EndpointSourceSet
 import org.jitsi.utils.MediaType
 import org.jitsi.utils.logging2.Logger
 import org.jitsi.utils.logging2.createChildLogger
+import org.jitsi.utils.stats.RateTracker
 import org.jitsi.xmpp.extensions.TraceParent
 import org.jitsi.xmpp.extensions.colibri.WebSocketPacketExtension
 import org.jitsi.xmpp.extensions.colibri2.Colibri2Endpoint
@@ -50,8 +52,13 @@ import org.jitsi.xmpp.extensions.jingle.IceUdpTransportPacketExtension
 import org.jivesoftware.smack.StanzaCollector
 import org.jivesoftware.smack.packet.IQ
 import org.jivesoftware.smackx.muc.MUCRole
+import java.time.Duration
 import java.util.Collections.singletonList
 import java.util.UUID
+import kotlin.math.ceil
+
+/** The size of the buckets used to track the rate at which endpoints are added to a session. */
+private const val BUCKET_SIZE_MS = 100L
 
 /** Represents a colibri2 session with one specific bridge. */
 class Colibri2Session(
@@ -67,6 +74,34 @@ class Colibri2Session(
     private val xmppConnection = colibriSessionManager.xmppConnection
     val id = UUID.randomUUID().toString()
 
+    /**
+     * Keep track of the endpoints that this conference recently added to this bridge. Note that this is
+     * per-(conference, bridge), as opposed to [Bridge]'s own tracker which is per-bridge (across all conferences).
+     *
+     * We use the number-of-buckets constructor (as opposed to the window-size one) so that an interval which is not a
+     * multiple of the bucket size is rounded up instead of failing at runtime.
+     */
+    private val newEndpointsRate = RateTracker(
+        numBuckets = ceil(
+            BridgeConfig.config.maxBridgeParticipantsInterval.toMillis().toDouble() / BUCKET_SIZE_MS
+        ).toInt().coerceAtLeast(1),
+        bucketSize = Duration.ofMillis(BUCKET_SIZE_MS),
+        clock = colibriSessionManager.clock
+    )
+
+    /** The number of endpoints that this conference recently added to this bridge. */
+    internal val recentlyAddedEndpointCount: Long
+        get() = newEndpointsRate.getAccumulatedCount()
+
+    /**
+     * Notifies this session that it was used for a new endpoint. Note that endpoints leaving are intentionally not
+     * subtracted (this models the rate at which endpoints were *added*, matching [Bridge.endpointAdded]).
+     */
+    internal fun endpointAdded() {
+        newEndpointsRate.update(1)
+        bridge.endpointAdded()
+    }
+
     /**
      * The colibri2 `<connect>`s currently active on this session, by id (the last set signaled to the bridge).
      * Managed via [setInitialConnects] (for the create request) and [setConnects] (delta updates).
@@ -410,6 +445,7 @@ class Colibri2Session(
         put("id", id)
         set<ObjectNode>("feedback_sources", feedbackSources.toJson())
         put("created", created)
+        put("recently_added_endpoints", recentlyAddedEndpointCount)
         set<ObjectNode>(
             "relays",
             JsonNodeFactory.instance.objectNode().apply {

jicofo-selector/src/main/kotlin/org/jitsi/jicofo/bridge/colibri/ColibriV2SessionManager.kt

@@ -59,6 +59,7 @@ import org.jivesoftware.smack.packet.StanzaError.Condition.item_not_found
 import org.jivesoftware.smack.packet.StanzaError.Condition.service_unavailable
 import java.net.URI
 import java.net.URLEncoder
+import java.time.Clock
 import java.util.Collections.singletonList
 
 /** The fixed connect id used for the (single) transcriber connect. */
@@ -100,7 +101,7 @@ internal fun perSourceTranslatorSpecs(
  * Implements [ColibriSessionManager] using colibri2.
  */
 @SuppressFBWarnings("BC_IMPOSSIBLE_INSTANCEOF")
-class ColibriV2SessionManager(
+class ColibriV2SessionManager @JvmOverloads constructor(
     internal val xmppConnection: AbstractXMPPConnection,
     private val bridgeSelector: BridgeSelector,
     internal val conferenceName: String,
@@ -111,7 +112,8 @@ class ColibriV2SessionManager(
     internal val meetingId: String,
     internal val rtcStatsEnabled: Boolean,
     private val bridgeVersion: String?,
-    parentLogger: Logger
+    parentLogger: Logger,
+    internal val clock: Clock = Clock.systemUTC()
 ) : ColibriSessionManager, Cascade<Colibri2Session, Colibri2Session.Relay> {
     private val logger = createChildLogger(parentLogger)
     private val tracer = TracingGlobal.sdk.getTracer("org.jitsi.jicofo.colibri")
@@ -477,7 +479,8 @@ class ColibriV2SessionManager(
                     it.key.bridge,
                     ConferenceBridgeProperties(
                         it.value.size,
-                        it.value.firstOrNull()?.visitor == true
+                        it.value.firstOrNull()?.visitor == true,
+                        it.key.recentlyAddedEndpointCount
                     )
                 )
             }
@@ -574,7 +577,7 @@ class ColibriV2SessionManager(
                 )
             }
             participantInfo = ParticipantInfo(participant, session)
-            session.bridge.endpointAdded()
+            session.endpointAdded()
             stanzaCollector = session.sendAllocationRequest(participantInfo)
             add(participantInfo)
             if (created) {

jicofo-selector/src/main/resources/reference.conf

@@ -5,6 +5,23 @@ jicofo {
   bridge {
     // The maximum number of participants in a single conference to put on one bridge (use -1 for no maximum).
     max-bridge-participants = 80
+    // The maximum number of participants that a single conference may add to a single bridge within
+    // [max-bridge-participants-interval]. A bridge which has reached this limit is considered overloaded *for that
+    // conference only* when selecting a bridge for a new participant; other conferences, and all other bridge-level
+    // logic (stress, sorting, load redistribution) are unaffected.
+    //
+    // This limits the overshoot that happens when a conference bursts: a bridge reports its stress level only every
+    // 5-10 seconds, so during a burst of joins the reported stress is stale by construction and jicofo can keep
+    // selecting the same bridge long after it is full. Note this is a limit on the *rate* at which a conference grows
+    // on a bridge, so it does not affect a conference which grows slowly (where the stress feedback has caught up),
+    // and it does not affect many small conferences sharing a bridge (each has a low rate of its own).
+    //
+    // Use -1 to disable. A suggested value for a production deployment is 30 (with the default 1 minute interval).
+    // Note that [max-bridge-participants] remains in effect as a hard limit.
+    max-bridge-participants-per-interval = -1
+    // The interval over which [max-bridge-participants-per-interval] is enforced. It should comfortably cover the
+    // interval at which bridges report their stress level (5-10 seconds).
+    max-bridge-participants-interval = 1 minute
     // The default assumed average stress per participant. This value is only used when a bridge does not report its
     // own value.
     average-participant-stress = 0.01

jicofo-selector/src/test/kotlin/org/jitsi/jicofo/bridge/BridgeConfigTest.kt

@@ -24,6 +24,7 @@ import org.jitsi.config.withNewConfig
 import org.jitsi.jicofo.bridge.BridgeConfig.Companion.config
 import org.jitsi.metaconfig.MetaconfigSettings
 import org.jitsi.utils.mins
+import org.jitsi.utils.secs
 
 class BridgeConfigTest : ShouldSpec() {
     override fun isolationMode() = IsolationMode.InstancePerLeaf
@@ -32,6 +33,20 @@ class BridgeConfigTest : ShouldSpec() {
         MetaconfigSettings.cacheEnabled = false
         context("with no config the defaults from reference.conf should be used") {
             config.maxBridgeParticipants shouldBe 80
+            // The rate limit is disabled by default.
+            config.maxBridgeParticipantsPerInterval shouldBe -1
+            config.maxBridgeParticipantsInterval shouldBe 1.mins
+        }
+        context("max-bridge-participants-per-interval") {
+            withNewConfig(
+                """
+                jicofo.bridge.max-bridge-participants-per-interval=30
+                jicofo.bridge.max-bridge-participants-interval=30 seconds
+                """.trimIndent()
+            ) {
+                config.maxBridgeParticipantsPerInterval shouldBe 30
+                config.maxBridgeParticipantsInterval shouldBe 30.secs
+            }
         }
         context("with legacy config") {
             withLegacyConfig(legacyConfig) {