feat(velocity): pace drain transfers instead of sending them all at once

velocity/src/main/kotlin/gg/grounds/drain/DrainConfig.kt

@@ -17,12 +17,20 @@ package gg.grounds.drain
  * is looked up by `CONTINENT` first and by `REGION` (`de-nbg1`) only as a fallback for a catalogue
  * keyed by datacentre.
  * - `GROUNDS_DRAIN_HTTP_PORT` — loopback port the preStop hook calls. `0` disables the endpoint.
+ * - `GROUNDS_DRAIN_TRANSFERS_PER_SECOND` — how many players a drain transfers per second (default
+ *   20). A full proxy then reconnects over seconds instead of in one burst of logins.
  */
-data class DrainConfig(val transferHost: String?, val transferPort: Int, val httpPort: Int) {
+data class DrainConfig(
+    val transferHost: String?,
+    val transferPort: Int,
+    val httpPort: Int,
+    val transfersPerSecond: Int = DEFAULT_TRANSFERS_PER_SECOND,
+) {
 
     companion object {
         const val DEFAULT_TRANSFER_PORT = 25565
         const val DEFAULT_HTTP_PORT = 8085
+        const val DEFAULT_TRANSFERS_PER_SECOND = 20
 
         fun fromEnv(env: Map<String, String> = System.getenv()): DrainConfig {
             val rawTarget =
@@ -53,7 +61,18 @@ data class DrainConfig(val transferHost: String?, val transferPort: Int, val htt
                             )
                     } ?: DEFAULT_HTTP_PORT
 
-            return DrainConfig(host, port, httpPort)
+            val transfersPerSecond =
+                env["GROUNDS_DRAIN_TRANSFERS_PER_SECOND"]
+                    ?.trim()
+                    ?.takeIf { it.isNotEmpty() }
+                    ?.let {
+                        it.toIntOrNull()?.takeIf { parsed -> parsed >= 1 }
+                            ?: throw IllegalArgumentException(
+                                "GROUNDS_DRAIN_TRANSFERS_PER_SECOND '$it' must be a whole number ≥ 1"
+                            )
+                    } ?: DEFAULT_TRANSFERS_PER_SECOND
+
+            return DrainConfig(host, port, httpPort, transfersPerSecond)
         }
 
         /**

velocity/src/main/kotlin/gg/grounds/drain/DrainManager.kt

@@ -3,37 +3,26 @@ package gg.grounds.drain
 import com.velocitypowered.api.network.ProtocolVersion
 import com.velocitypowered.api.proxy.Player
 import com.velocitypowered.api.proxy.ProxyServer
+import com.velocitypowered.api.scheduler.ScheduledTask
 import java.net.InetSocketAddress
 import java.util.concurrent.TimeUnit
 import net.kyori.adventure.text.Component
 import org.slf4j.Logger
 
-internal fun <T> transferAllSafely(
-    players: Iterable<T>,
-    transfer: (T) -> Unit,
-    onFailure: (T, Exception) -> Unit,
-) {
-    players.forEach { player ->
-        try {
-            transfer(player)
-        } catch (error: Exception) {
-            onFailure(player, error)
-        }
-    }
-}
-
 /**
  * Moves players off this proxy before it shuts down, instead of letting Velocity kick them.
  *
  * Started by the pod's preStop hook via [DrainHttpServer]. From that moment on:
  * - new logins are denied (they bounce back through the public name to a live proxy),
- * - players sitting in a lobby are sent a Minecraft transfer packet right away,
+ * - players sitting in a lobby are sent a Minecraft transfer packet, paced at
+ *   [DrainConfig.transfersPerSecond] (see [PacedTransfers]) so a full proxy does not reconnect all
+ *   at once,
  * - players inside a round stay untouched — Agones keeps their GameServer alive, and the moment the
  *   round sends them back towards a lobby the transfer happens *instead of* that connect,
- * - at the deadline, whoever is *not* inside a round is transferred; players inside a round are
- *   never transferred — the round either ends within the grace period (and the lobby connect
- *   becomes the transfer), or the pod's termination ends the session. A transfer would end the
- *   round just as surely, only earlier.
+ * - at the deadline, whoever is *not* inside a round is transferred (paced the same way); players
+ *   inside a round are never transferred — the round either ends within the grace period (and the
+ *   lobby connect becomes the transfer), or the pod's termination ends the session. A transfer
+ *   would end the round just as surely, only earlier.
  *
  * "Inside a round" is decided by the server's `grounds/server-type` role: only the `game` and
  * `match` roles defer the transfer. A server that discovery has no role for cannot be a protected
@@ -50,6 +39,15 @@ class DrainManager(
     private val transferStager: DrainTransferStager = DrainTransferStager { action ->
         proxy.scheduler.buildTask(plugin, Runnable(action)).delay(1, TimeUnit.SECONDS).schedule()
     },
+    private val ticker: DrainTicker = DrainTicker { tick ->
+        var task: ScheduledTask? = null
+        task =
+            proxy.scheduler
+                .buildTask(plugin, Runnable { if (tick()) task?.cancel() })
+                .delay(1, TimeUnit.SECONDS)
+                .repeat(1, TimeUnit.SECONDS)
+                .schedule()
+    },
 ) {
     @Volatile
     var isDraining: Boolean = false
@@ -73,17 +71,7 @@ class DrainManager(
             .buildTask(plugin, Runnable { onDeadline() })
             .delay(deadlineSeconds, TimeUnit.SECONDS)
             .schedule()
-        transferAllSafely(
-            proxy.allPlayers.filter { !shouldDefer(roleOf(it), lobbyValue) },
-            { player -> transferOut(player, force = false) },
-            { player, error ->
-                logger.warn(
-                    "Failed to transfer draining player {}; leaving for deadline",
-                    player.username,
-                    error,
-                )
-            },
-        )
+        pace(proxy.allPlayers.filter { !shouldDefer(roleOf(it), lobbyValue) }, force = false)
         return true
     }
 
@@ -106,13 +94,7 @@ class DrainManager(
                 "Drain deadline reached; transferring {} players not inside a round",
                 drainable.size,
             )
-            transferAllSafely(
-                drainable,
-                { player -> transferOut(player, force = true) },
-                { player, error ->
-                    logger.warn("Failed to transfer draining player {}", player.username, error)
-                },
-            )
+            pace(drainable, force = true)
         }
         if (inRound.isNotEmpty()) {
             logger.warn(
@@ -123,6 +105,30 @@ class DrainManager(
         }
     }
 
+    /**
+     * Transfers [players] at [DrainConfig.transfersPerSecond]: the first batch now, the rest one
+     * batch a second. Whoever left or walked into a round while queued is skipped.
+     */
+    private fun pace(players: List<Player>, force: Boolean) {
+        val paced =
+            PacedTransfers(
+                players,
+                config.transfersPerSecond,
+                stillDue = { it.isActive && !shouldDefer(roleOf(it), lobbyValue) },
+                transfer = { transferOut(it, force) },
+                onFailure = { player, error ->
+                    logger.warn("Failed to transfer draining player {}", player.username, error)
+                },
+            )
+        if (paced.tick()) return
+        logger.info(
+            "Transferring the remaining {} players at {}/s",
+            paced.remaining,
+            config.transfersPerSecond,
+        )
+        ticker.everySecondUntilDone(paced::tick)
+    }
+
     /**
      * True when the transfer (or, under force, the disconnect) was issued. Without force, a player
      * we cannot transfer — no target configured, client older than 1.20.5 — is left alone: they

velocity/src/main/kotlin/gg/grounds/drain/PacedTransfers.kt

@@ -0,0 +1,53 @@
+package gg.grounds.drain
+
+/**
+ * Hands players to [transfer] at most [perSecond] at a time: one batch when [tick] is first called,
+ * the next on every later call (the caller ticks once a second).
+ *
+ * A drain used to send every transfer packet in the same instant, so a full proxy meant hundreds of
+ * clients reconnecting in one second: hundreds of logins, session checks against Mojang and service
+ * lookups landing on the replacement proxy at once. Paced, 400 players at 20/s take 20 s, well
+ * inside the drain deadline.
+ *
+ * A player is checked again when their turn comes ([stillDue]): one who left, or walked into a
+ * round while queued, is skipped and does not use up a slot of the batch. A transfer that throws is
+ * reported to [onFailure] and does not stop the rest.
+ */
+internal class PacedTransfers<T>(
+    players: Iterable<T>,
+    private val perSecond: Int,
+    private val stillDue: (T) -> Boolean,
+    private val transfer: (T) -> Unit,
+    private val onFailure: (T, Exception) -> Unit,
+) {
+    init {
+        require(perSecond >= 1) { "perSecond must be at least 1, was $perSecond" }
+    }
+
+    private val queue = ArrayDeque(players.toList())
+
+    val remaining: Int
+        @Synchronized get() = queue.size
+
+    /** Sends the next batch. True once nobody is left to send. */
+    @Synchronized
+    fun tick(): Boolean {
+        var sent = 0
+        while (sent < perSecond && queue.isNotEmpty()) {
+            val player = queue.removeFirst()
+            if (!stillDue(player)) continue
+            try {
+                transfer(player)
+            } catch (error: Exception) {
+                onFailure(player, error)
+            }
+            sent++
+        }
+        return queue.isEmpty()
+    }
+}
+
+/** Runs a tick once a second until it reports it is done. Injected so tests can tick by hand. */
+fun interface DrainTicker {
+    fun everySecondUntilDone(tick: () -> Boolean)
+}

velocity/src/test/kotlin/gg/grounds/drain/PacedTransfersTest.kt

@@ -0,0 +1,78 @@
+package gg.grounds.drain
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+
+class PacedTransfersTest {
+    @Test
+    fun `a full proxy is transferred in batches, not in one burst`() {
+        val sent = mutableListOf<Int>()
+        val paced = PacedTransfers((1..45).toList(), 20, { true }, { sent += it }, { _, _ -> })
+
+        assertFalse(paced.tick())
+        assertEquals(20, sent.size)
+        assertFalse(paced.tick())
+        assertEquals(40, sent.size)
+        assertTrue(paced.tick())
+        assertEquals((1..45).toList(), sent)
+    }
+
+    @Test
+    fun `a few players go out at once`() {
+        val sent = mutableListOf<String>()
+        val paced = PacedTransfers(listOf("a", "b"), 20, { true }, { sent += it }, { _, _ -> })
+
+        assertTrue(paced.tick())
+        assertEquals(listOf("a", "b"), sent)
+    }
+
+    @Test
+    fun `players who left or entered a round while queued are skipped without using a slot`() {
+        val gone = setOf(2, 3)
+        val sent = mutableListOf<Int>()
+        val paced = PacedTransfers((1..6).toList(), 2, { it !in gone }, { sent += it }, { _, _ -> })
+
+        paced.tick()
+        assertEquals(listOf(1, 4), sent)
+        assertTrue(paced.tick())
+        assertEquals(listOf(1, 4, 5, 6), sent)
+    }
+
+    @Test
+    fun `a failed transfer does not stop later drain transfers`() {
+        val attempted = mutableListOf<String>()
+        val failures = mutableListOf<String>()
+        val paced =
+            PacedTransfers(
+                listOf("first", "second"),
+                20,
+                { true },
+                { player ->
+                    attempted += player
+                    if (player == "first") error("connection closed")
+                },
+            ) { player, _ ->
+                failures += player
+            }
+
+        assertTrue(paced.tick())
+        assertEquals(listOf("first", "second"), attempted)
+        assertEquals(listOf("first"), failures)
+    }
+
+    @Test
+    fun `the pace defaults to 20 per second and can be set`() {
+        assertEquals(20, DrainConfig.fromEnv(emptyMap()).transfersPerSecond)
+        assertEquals(
+            50,
+            DrainConfig.fromEnv(mapOf("GROUNDS_DRAIN_TRANSFERS_PER_SECOND" to "50"))
+                .transfersPerSecond,
+        )
+        assertThrows<IllegalArgumentException> {
+            DrainConfig.fromEnv(mapOf("GROUNDS_DRAIN_TRANSFERS_PER_SECOND" to "0"))
+        }
+    }
+}