feat(velocity)!: fill the fullest lobby below a 100-player cap

README.md

@@ -37,6 +37,7 @@ deployments do not need to set anything.
 | `GROUNDS_AGONES_ADDRESS_TYPE`    | `PodIP`                                         | Which entry of `status.addresses` to dial (`PodIP`, `ExternalIP`, …)  |
 | `GROUNDS_AGONES_PORT`            | `25565`                                         | TCP port on the GameServer                                            |
 | `GROUNDS_STATIC_SERVERS`         | _(none)_                                        | Comma-separated `name=host:port` static Velocity backends             |
+| `GROUNDS_LOBBY_PLAYER_CAP`       | `100`                                           | Players a lobby is filled to before joins go to the next one          |
 
 `GROUNDS_STATIC_SERVERS` is validated strictly: names and hosts must be non-empty, ports must be
 between `1` and `65535`, and names must be unique. For example, Stage proxies can use:
@@ -48,6 +49,14 @@ GROUNDS_STATIC_SERVERS=buildserver=buildserver:25565
 This only registers the backend with Velocity; configuring the backend itself for Velocity
 forwarding remains a separate deployment concern.
 
+Joins, lobby kicks and plugin-match's returns from a match all go to the fullest lobby that is
+below `GROUNDS_LOBBY_PLAYER_CAP`, and to an emptier one only when every lobby is at the cap. A
+return goes back to the lobby the player queued from first, while that lobby is current and below
+the cap.
+Counts are the larger of the network-wide count (plugin-proxy) and this proxy's own, plus the
+players this proxy has sent but who have not arrived yet, so a burst of joins on one proxy cannot
+overfill a lobby. Across proxies the cap is soft by the couple of seconds the network count lags.
+
 Typical Helm chart wiring uses a `ConfigMap` consumed via `envFrom`, plus
 `POD_NAMESPACE` from the Downward API for clusters where the proxy should
 watch its own namespace:

velocity/src/main/kotlin/gg/grounds/GroundsPluginAgones.kt

@@ -22,6 +22,7 @@ import gg.grounds.i18n.Translations
 import gg.grounds.presentation.VelocitySystemMessages
 import gg.grounds.proxy.api.ProxyServiceRegistry
 import gg.grounds.proxy.api.ServerDisplayQuery
+import java.util.UUID
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
@@ -55,7 +56,7 @@ constructor(private val proxyServer: ProxyServer, private val logger: Logger) {
             )
         val discoveryConfig = DiscoveryConfig.fromEnv()
         logger.info(
-            "Loaded Agones discovery config (namespace={}, labelSelector={}, lobbyLabel={}, lobbyValue={}, runningStates={}, pollInterval={}s, addressType={}, port={})",
+            "Loaded Agones discovery config (namespace={}, labelSelector={}, lobbyLabel={}, lobbyValue={}, runningStates={}, pollInterval={}s, addressType={}, port={}, lobbyPlayerCap={})",
             discoveryConfig.namespace,
             discoveryConfig.labelSelector.ifEmpty { "<none>" },
             discoveryConfig.lobbyLabel.ifEmpty { "<none>" },
@@ -64,6 +65,7 @@ constructor(private val proxyServer: ProxyServer, private val logger: Logger) {
             discoveryConfig.pollInterval.toSeconds(),
             discoveryConfig.addressType,
             discoveryConfig.port,
+            discoveryConfig.lobbyPlayerCap,
         )
 
         stateManager =
@@ -117,7 +119,7 @@ constructor(private val proxyServer: ProxyServer, private val logger: Logger) {
                 { serverName ->
                     discoveryService.getServerRole(serverName) == discoveryConfig.lobbyValue
                 },
-                { exclude -> discoveryService.pickLobby(exclude) },
+                { exclude, playerId -> discoveryService.pickLobby(exclude, playerId) },
             ),
         )
         outdatedLobbyMover =
@@ -127,6 +129,7 @@ constructor(private val proxyServer: ProxyServer, private val logger: Logger) {
                     logger,
                     discoveryService::lobbyCandidates,
                     { drainManager.isDraining },
+                    lobbyPlayerCap = discoveryConfig.lobbyPlayerCap,
                     messages = messages,
                 )
                 .also { it.start() }
@@ -156,6 +159,23 @@ constructor(private val proxyServer: ProxyServer, private val logger: Logger) {
         coroutineScope.cancel()
     }
 
+    /**
+     * The lobby [playerId] should go to, never [exclude], with their place reserved; null when
+     * there is none or discovery has not started. [preferred] — the lobby they queued from — is
+     * taken while it is current and below the cap, reservations included; otherwise the fill rule
+     * decides.
+     *
+     * For plugin-match, which sends players back to a lobby after a match. It calls this by
+     * reflection — JDK types only — so the two plugins share one placement rule and one set of
+     * reservations without a compile-time dependency on each other.
+     */
+    fun pickLobby(playerId: UUID, exclude: String?, preferred: String?): String? =
+        if (this::discoveryService.isInitialized) {
+            discoveryService.pickLobby(exclude, playerId, preferred)?.serverInfo?.name
+        } else {
+            null
+        }
+
     /**
      * Publishes the server display query, if there is a proxy API to publish it into.
      *

velocity/src/main/kotlin/gg/grounds/discovery/DiscoveryConfig.kt

@@ -24,6 +24,8 @@ internal fun canonicalServerName(name: String): String = name.lowercase(Locale.R
  * - `GROUNDS_AGONES_PORT` — TCP port to dial on the discovered GameServer.
  * - `GROUNDS_STATIC_SERVERS` — Comma-separated `name=host:port` Velocity backends that are
  *   registered without Agones discovery.
+ * - `GROUNDS_LOBBY_PLAYER_CAP` — Players a lobby is filled up to before joins go to the next one
+ *   (default 100). Not `GROUNDS_AGONES_`-prefixed: plugin-match reads the same variable.
  */
 data class StaticServer(val name: String, val host: String, val port: Int)
 
@@ -37,6 +39,7 @@ data class DiscoveryConfig(
     val addressType: String,
     val port: Int,
     val staticServers: List<StaticServer>,
+    val lobbyPlayerCap: Int = LobbySelection.DEFAULT_PLAYER_CAP,
 ) {
     companion object {
         const val DEFAULT_NAMESPACE = "games"
@@ -68,6 +71,9 @@ data class DiscoveryConfig(
                 addressType = env["GROUNDS_AGONES_ADDRESS_TYPE"] ?: DEFAULT_ADDRESS_TYPE,
                 port = env["GROUNDS_AGONES_PORT"]?.toIntOrNull() ?: DEFAULT_PORT,
                 staticServers = parseStaticServers(env["GROUNDS_STATIC_SERVERS"]),
+                lobbyPlayerCap =
+                    env["GROUNDS_LOBBY_PLAYER_CAP"]?.trim()?.toIntOrNull()?.takeIf { it > 0 }
+                        ?: LobbySelection.DEFAULT_PLAYER_CAP,
             )
 
         private val DURATION_PATTERN = Regex("""^(\d+)\s*(s|m|h)$""")

velocity/src/main/kotlin/gg/grounds/discovery/DiscoveryPlayerListener.kt

@@ -50,10 +50,10 @@ class DiscoveryPlayerListener(
     private val proxyServer: ProxyServer,
     /**
      * The lobby a joining player lands on, or null when there is none — see
-     * [DiscoveryService.pickLobby]. Spreads by network-wide counts when the network can be asked,
-     * by this proxy's own view otherwise, and avoids outdated lobbies.
+     * [DiscoveryService.pickLobby]. Fills the fullest lobby below the cap and avoids outdated
+     * lobbies. Given a player it reserves their place; given null it only asks.
      */
-    private val pickLobby: () -> RegisteredServer?,
+    private val pickLobby: (UUID?) -> RegisteredServer?,
     private val serverRole: (String) -> String?,
     private val drainTransferCookie: DrainTransferCookie = DrainTransferCookie(),
     private val sourceCookiePending: (String) -> Boolean = { false },
@@ -70,7 +70,7 @@ class DiscoveryPlayerListener(
     fun onLogin(event: LoginEvent) {
         if (
             !shouldDenyInitialLogin(
-                pickLobby() != null,
+                pickLobby(null) != null,
                 proxyServer.allServers.any { server ->
                     serverRole(server.serverInfo.name) == STATIC_SERVER_ROLE
                 },
@@ -144,7 +144,7 @@ class DiscoveryPlayerListener(
             event.setInitialServer(preferred)
             return
         }
-        pickLobby()?.let(event::setInitialServer)
+        pickLobby(event.player.uniqueId)?.let(event::setInitialServer)
     }
 
     private inner class PendingCookieRequest(

velocity/src/main/kotlin/gg/grounds/discovery/DiscoveryService.kt

@@ -15,6 +15,7 @@ import io.kubernetes.client.openapi.apis.CoreV1Api
 import io.kubernetes.client.openapi.apis.CustomObjectsApi
 import io.kubernetes.client.util.Config
 import java.net.InetSocketAddress
+import java.util.UUID
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.TimeUnit
 import org.slf4j.Logger
@@ -80,6 +81,7 @@ class DiscoveryService(
     private val staticServerNames =
         config.staticServers.mapTo(mutableSetOf()) { canonicalServerName(it.name) }
     @Volatile private var countsSnapshot: Pair<Long, Map<String, Int>?>? = null
+    private val placement = LobbyPlacement(config.lobbyPlayerCap)
 
     fun start() {
         unregisterBakedInPlaceholders()
@@ -109,31 +111,47 @@ class DiscoveryService(
         owningGameServerSets[canonicalServerName(serverName)]?.let { it in retiredGameServerSets }
             ?: false
 
-    /** Every registered lobby, with its player count and whether it is outdated. */
-    fun lobbyCandidates(): List<LobbySelection.Candidate> {
+    /**
+     * Every registered lobby, with its player count and whether it is outdated. The count includes
+     * the players this proxy is still sending there; see [LobbyPlacement].
+     */
+    fun lobbyCandidates(): List<LobbySelection.Candidate> =
+        placement.withReservations(observedLobbies())
+
+    /**
+     * The lobby [playerId] should be sent to, never [exclude]; see [LobbySelection.pick]. With a
+     * [playerId] the place is reserved until they arrive; without one this only asks. [preferred]
+     * is taken while it has room; see [LobbyPlacement.place].
+     */
+    fun pickLobby(
+        exclude: String? = null,
+        playerId: UUID? = null,
+        preferred: String? = null,
+    ): RegisteredServer? {
+        val chosen = placement.place(observedLobbies(), playerId, exclude, preferred) ?: return null
+        return proxyServer.getServer(chosen).orElse(null)
+    }
+
+    /**
+     * Every registered lobby with the players already on it. The network count says how many are on
+     * it across every proxy, but only once their sessions are written and the cache has refreshed;
+     * this proxy's own `playersConnected` is exact the moment a connection completes. The larger of
+     * the two is the better answer for both — a lag in either only ever hides players.
+     */
+    private fun observedLobbies(): List<LobbySelection.Candidate> {
         val lobbies = proxyServer.allServers.filter { it.serverInfo.name in lobbyServers }
         if (lobbies.isEmpty()) return emptyList()
         val counts = networkCountsCached()
         return lobbies.map { server ->
             val name = server.serverInfo.name
             LobbySelection.Candidate(
                 name,
-                if (counts != null) counts[name] ?: 0 else server.playersConnected.size,
+                maxOf(counts?.get(name) ?: 0, server.playersConnected.size),
                 isOutdated(name),
             )
         }
     }
 
-    /** The lobby a player should be sent to, never [exclude]; see [LobbySelection.pick]. */
-    fun pickLobby(exclude: String? = null): RegisteredServer? {
-        val candidates =
-            lobbyCandidates().filter {
-                exclude == null || canonicalServerName(it.name) != canonicalServerName(exclude)
-            }
-        val chosen = LobbySelection.pick(candidates) ?: return null
-        return proxyServer.getServer(chosen).orElse(null)
-    }
-
     private fun unregisterBakedInPlaceholders() {
         for (server in proxyServer.allServers.filter(::isBakedInPlaceholder)) {
             proxyServer.unregisterServer(server.serverInfo)
@@ -172,12 +190,13 @@ class DiscoveryService(
     }
 
     private fun registerListeners() {
+        proxyServer.eventManager.register(plugin, placement.Listener())
         proxyServer.eventManager.register(
             plugin,
             DiscoveryPlayerListener(
                 plugin,
                 proxyServer,
-                { pickLobby() },
+                { playerId -> pickLobby(playerId = playerId) },
                 this::getServerRole,
                 drainTransferCookie,
                 sourceCookiePending,

velocity/src/main/kotlin/gg/grounds/discovery/LobbyKickListener.kt

@@ -3,6 +3,7 @@ package gg.grounds.discovery
 import com.velocitypowered.api.event.Subscribe
 import com.velocitypowered.api.event.player.KickedFromServerEvent
 import com.velocitypowered.api.proxy.server.RegisteredServer
+import java.util.UUID
 
 /**
  * A lobby that goes away takes its players to another lobby instead of off the network.
@@ -14,8 +15,11 @@ import com.velocitypowered.api.proxy.server.RegisteredServer
  */
 class LobbyKickListener(
     private val isLobby: (String) -> Boolean,
-    /** Another lobby to send the player to, never the one named; null when there is none. */
-    private val pickLobby: (exclude: String) -> RegisteredServer?,
+    /**
+     * Another lobby to send the player to, never the one named; null when there is none. Picked
+     * like a join — the fullest lobby with room — and reserved for the player.
+     */
+    private val pickLobby: (exclude: String, playerId: UUID) -> RegisteredServer?,
 ) {
 
     @Subscribe
@@ -24,7 +28,7 @@ class LobbyKickListener(
         if (event.result !is KickedFromServerEvent.DisconnectPlayer) return
         val kickedFrom = event.server.serverInfo.name
         if (!isLobby(kickedFrom)) return
-        val target = pickLobby(kickedFrom) ?: return
+        val target = pickLobby(kickedFrom, event.player.uniqueId) ?: return
         event.result =
             KickedFromServerEvent.RedirectPlayer.create(target, event.serverKickReason.orElse(null))
     }

velocity/src/main/kotlin/gg/grounds/discovery/LobbyPlacement.kt

@@ -0,0 +1,110 @@
+package gg.grounds.discovery
+
+import com.velocitypowered.api.event.Subscribe
+import com.velocitypowered.api.event.connection.DisconnectEvent
+import com.velocitypowered.api.event.player.ServerPostConnectEvent
+import java.time.Duration
+import java.util.UUID
+
+/**
+ * Picks a lobby and remembers the pick until the player is on it.
+ *
+ * The counts a pick works from lag the joins it makes: a player this proxy just sent to a lobby is
+ * not in Velocity's `playersConnected` until the connection completes, and not in the network count
+ * until their session says so and the cache has refreshed. Without a memory of its own picks, a
+ * burst of joins would all see the same 99 and all go to the same lobby.
+ *
+ * So each pick reserves a place for the player, and a candidate's count is its observed count plus
+ * the reservations still open on it. The pick and the reservation happen under one lock, which
+ * makes joins on this proxy exact: the 101st sees 100 and goes elsewhere. A reservation ends when
+ * the player reaches a server (they are then in `playersConnected`), leaves the proxy, or after
+ * [ttl] — a connection that fails quietly must not hold a place for ever.
+ *
+ * Joins on *another* proxy are only seen through the network count, a couple of seconds late, so
+ * the cap is soft across proxies: two proxies filling the same lobby at once can overshoot it by
+ * the joins each makes in that window. That costs a few players over 100, not a lobby of 200.
+ */
+class LobbyPlacement(
+    private val cap: Int = LobbySelection.DEFAULT_PLAYER_CAP,
+    private val ttl: Duration = DEFAULT_TTL,
+    private val clock: () -> Long = System::nanoTime,
+) {
+    private data class Reservation(val server: String, val expiresAt: Long)
+
+    private val reservations = HashMap<UUID, Reservation>()
+
+    /**
+     * The lobby for [playerId], never [exclude]; null when there is none. [observed] are the
+     * lobbies with the players already counted on them. A null [playerId] only asks — nothing is
+     * reserved.
+     *
+     * [preferred] wins over the fill rule while it is a current lobby below the cap, reservations
+     * included — the lobby a player queued from, for a return after their match. Gone, outdated or
+     * full, it is ignored and the fill rule decides.
+     */
+    @Synchronized
+    fun place(
+        observed: List<LobbySelection.Candidate>,
+        playerId: UUID? = null,
+        exclude: String? = null,
+        preferred: String? = null,
+    ): String? {
+        // A player placed again (kicked before arriving, say) must not count against themselves.
+        if (playerId != null) reservations.remove(playerId)
+        val candidates =
+            withReservations(observed).filter {
+                exclude == null || canonicalServerName(it.name) != canonicalServerName(exclude)
+            }
+        val home =
+            preferred?.let { name ->
+                candidates.firstOrNull {
+                    canonicalServerName(it.name) == canonicalServerName(name) &&
+                        !it.outdated &&
+                        it.players < cap
+                }
+            }
+        val chosen = home?.name ?: LobbySelection.pick(candidates, cap) ?: return null
+        if (playerId != null) {
+            reservations[playerId] =
+                Reservation(canonicalServerName(chosen), clock() + ttl.toNanos())
+        }
+        return chosen
+    }
+
+    /** [observed] with this proxy's open reservations added to each lobby's count. */
+    @Synchronized
+    fun withReservations(observed: List<LobbySelection.Candidate>): List<LobbySelection.Candidate> {
+        val now = clock()
+        reservations.values.removeIf { it.expiresAt - now <= 0 }
+        val pending = reservations.values.groupingBy { it.server }.eachCount()
+        return observed.map {
+            it.copy(players = it.players + (pending[canonicalServerName(it.name)] ?: 0))
+        }
+    }
+
+    @Synchronized
+    fun release(playerId: UUID) {
+        reservations.remove(playerId)
+    }
+
+    /** Ends a reservation once its player is on a server, or gone. */
+    inner class Listener {
+        // Post-connect, not ServerConnectedEvent: Velocity adds the player to the server's
+        // `playersConnected` after that event, and releasing first would let the next pick see the
+        // lobby one player short.
+        @Subscribe
+        fun onServerPostConnect(event: ServerPostConnectEvent) {
+            release(event.player.uniqueId)
+        }
+
+        @Subscribe
+        fun onDisconnect(event: DisconnectEvent) {
+            release(event.player.uniqueId)
+        }
+    }
+
+    companion object {
+        /** Long enough for a slow join, short enough that a lost one frees its place quickly. */
+        val DEFAULT_TTL: Duration = Duration.ofSeconds(15)
+    }
+}