diff --git a/velocity/src/main/kotlin/gg/grounds/drain/DrainConfig.kt b/velocity/src/main/kotlin/gg/grounds/drain/DrainConfig.kt index ded3254..50bb226 100644 --- a/velocity/src/main/kotlin/gg/grounds/drain/DrainConfig.kt +++ b/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 = 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) } /** diff --git a/velocity/src/main/kotlin/gg/grounds/drain/DrainManager.kt b/velocity/src/main/kotlin/gg/grounds/drain/DrainManager.kt index 90dd74b..dbb4153 100644 --- a/velocity/src/main/kotlin/gg/grounds/drain/DrainManager.kt +++ b/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 transferAllSafely( - players: Iterable, - 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, 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 diff --git a/velocity/src/main/kotlin/gg/grounds/drain/PacedTransfers.kt b/velocity/src/main/kotlin/gg/grounds/drain/PacedTransfers.kt new file mode 100644 index 0000000..07f13a4 --- /dev/null +++ b/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( + players: Iterable, + 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) +} diff --git a/velocity/src/test/kotlin/gg/grounds/drain/DrainTransferSafetyTest.kt b/velocity/src/test/kotlin/gg/grounds/drain/DrainTransferSafetyTest.kt deleted file mode 100644 index 67a6b0c..0000000 --- a/velocity/src/test/kotlin/gg/grounds/drain/DrainTransferSafetyTest.kt +++ /dev/null @@ -1,25 +0,0 @@ -package gg.grounds.drain - -import org.junit.jupiter.api.Assertions.assertEquals -import org.junit.jupiter.api.Test - -class DrainTransferSafetyTest { - @Test - fun `a failed transfer does not stop later drain transfers`() { - val attempted = mutableListOf() - val failures = mutableListOf() - - transferAllSafely( - listOf("first", "second"), - { player -> - attempted += player - if (player == "first") error("connection closed") - }, - ) { player, _ -> - failures += player - } - - assertEquals(listOf("first", "second"), attempted) - assertEquals(listOf("first"), failures) - } -} diff --git a/velocity/src/test/kotlin/gg/grounds/drain/PacedTransfersTest.kt b/velocity/src/test/kotlin/gg/grounds/drain/PacedTransfersTest.kt new file mode 100644 index 0000000..15a28b8 --- /dev/null +++ b/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() + 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() + 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() + 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() + val failures = mutableListOf() + 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 { + DrainConfig.fromEnv(mapOf("GROUNDS_DRAIN_TRANSFERS_PER_SECOND" to "0")) + } + } +}