feat(velocity): drain players via transfer before shutdown

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

@@ -9,6 +9,10 @@ import com.velocitypowered.api.proxy.ProxyServer
 import gg.grounds.command.AgonesCommand
 import gg.grounds.discovery.DiscoveryConfig
 import gg.grounds.discovery.DiscoveryService
+import gg.grounds.drain.DrainConfig
+import gg.grounds.drain.DrainHttpServer
+import gg.grounds.drain.DrainListener
+import gg.grounds.drain.DrainManager
 import gg.grounds.gameserver.GameServerStateManager
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@@ -30,6 +34,7 @@ constructor(private val proxyServer: ProxyServer, private val logger: Logger) {
     private val coroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
     private lateinit var stateManager: GameServerStateManager
     private lateinit var discoveryService: DiscoveryService
+    private lateinit var drainHttpServer: DrainHttpServer
 
     @Subscribe
     fun onProxyInitialize(event: ProxyInitializeEvent) {
@@ -56,11 +61,28 @@ constructor(private val proxyServer: ProxyServer, private val logger: Logger) {
             AgonesCommand(proxyServer, { serverName -> discoveryService.getServerRole(serverName) }),
         )
 
+        val drainConfig = DrainConfig.fromEnv()
+        val drainManager =
+            DrainManager(
+                this,
+                proxyServer,
+                logger,
+                drainConfig,
+                { serverName -> discoveryService.getServerRole(serverName) },
+                discoveryConfig.lobbyValue,
+            )
+        proxyServer.eventManager.register(this, DrainListener(drainManager))
+        drainHttpServer =
+            DrainHttpServer(drainManager, drainConfig.httpPort, logger).also { it.start() }
+
         logger.info("Initialized Agones plugin (platform=velocity)")
     }
 
     @Subscribe
     fun onProxyShutdown(event: ProxyShutdownEvent) {
+        if (this::drainHttpServer.isInitialized) {
+            drainHttpServer.stop()
+        }
         if (this::discoveryService.isInitialized) {
             discoveryService.stop()
         }

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

@@ -0,0 +1,73 @@
+package gg.grounds.drain
+
+/**
+ * Drain configuration sourced from environment variables. All keys are optional; with none set the
+ * HTTP endpoint still comes up (loopback only) and a drain degrades to "wait for players to leave",
+ * which is still strictly better than the kick it replaces.
+ *
+ * Environment keys:
+ * - `GROUNDS_DRAIN_TRANSFER_HOST` — `host[:port]` the Minecraft transfer packet sends players to.
+ *   This is the *public* name the client reconnects through (mc-router resolves it to whichever
+ *   proxy is alive), not a backend address. Optional: when unset, the target is this region's own
+ *   entry in `REGIONS` (the `code=host[:port]` catalogue `/region` already transfers players with,
+ *   selected by `REGION`) — the same values file then works in every region. With neither set,
+ *   players are never transferred, only waited for.
+ * - `GROUNDS_DRAIN_HTTP_PORT` — loopback port the preStop hook calls. `0` disables the endpoint.
+ */
+data class DrainConfig(val transferHost: String?, val transferPort: Int, val httpPort: Int) {
+
+    companion object {
+        const val DEFAULT_TRANSFER_PORT = 25565
+        const val DEFAULT_HTTP_PORT = 8085
+
+        fun fromEnv(env: Map<String, String> = System.getenv()): DrainConfig {
+            val rawTarget =
+                env["GROUNDS_DRAIN_TRANSFER_HOST"]?.trim()?.takeIf { it.isNotEmpty() }
+                    ?: ownRegionTarget(env["REGIONS"], env["REGION"])
+            // rsplit, because a host can contain no colon but a port always follows the last one.
+            val host =
+                rawTarget?.substringBeforeLast(':', rawTarget)?.trim()?.takeIf { it.isNotEmpty() }
+            val portText = rawTarget?.substringAfterLast(':', "")?.trim().orEmpty()
+            val port =
+                when {
+                    portText.isEmpty() -> DEFAULT_TRANSFER_PORT
+                    else ->
+                        portText.toIntOrNull()?.takeIf { it in 1..65535 }
+                            ?: throw IllegalArgumentException(
+                                "drain transfer target '$rawTarget' has a bad port"
+                            )
+                }
+            val httpPort =
+                env["GROUNDS_DRAIN_HTTP_PORT"]
+                    ?.trim()
+                    ?.takeIf { it.isNotEmpty() }
+                    ?.let {
+                        it.toIntOrNull()?.takeIf { parsed -> parsed in 0..65535 }
+                            ?: throw IllegalArgumentException(
+                                "GROUNDS_DRAIN_HTTP_PORT '$it' must be a port number or 0"
+                            )
+                    } ?: DEFAULT_HTTP_PORT
+
+            return DrainConfig(host, port, httpPort)
+        }
+
+        /**
+         * This region's `host[:port]` out of the `REGIONS` catalogue. Choosing the *own* region on
+         * purpose: a drained player should land where they already are, latency-wise — the
+         * replacement proxy in the same region — not wherever a geo name happens to steer them.
+         * Null (no catalogue, no region, region not listed) leaves the drain wait-only.
+         */
+        internal fun ownRegionTarget(regions: String?, region: String?): String? {
+            if (regions.isNullOrBlank() || region.isNullOrBlank()) return null
+            val code = region.trim()
+            return regions
+                .split(',')
+                .map { it.trim() }
+                .firstNotNullOfOrNull { entry ->
+                    val entryCode = entry.substringBefore('=', "").trim()
+                    val target = entry.substringAfter('=', "").trim()
+                    target.takeIf { entryCode.equals(code, ignoreCase = true) && it.isNotEmpty() }
+                }
+        }
+    }
+}

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

@@ -0,0 +1,83 @@
+package gg.grounds.drain
+
+import com.sun.net.httpserver.HttpExchange
+import com.sun.net.httpserver.HttpServer
+import java.net.InetAddress
+import java.net.InetSocketAddress
+import java.nio.charset.StandardCharsets
+import org.slf4j.Logger
+
+/**
+ * The loopback endpoint the pod's preStop hook talks to. Loopback on purpose: the only caller is
+ * `sh` inside the same container, and a drain trigger reachable from the cluster network would be a
+ * kick-everyone button.
+ * - `GET/POST /drain/start?deadlineSeconds=N` — begin draining; idempotent.
+ * - `GET /drain/players` — plain-text player count; the hook polls this until it reads `0`.
+ * - `GET /drain/status` — the same, for humans: `{"draining":bool,"players":N}`.
+ */
+class DrainHttpServer(
+    private val drainManager: DrainManager,
+    private val port: Int,
+    private val logger: Logger,
+) {
+    private var server: HttpServer? = null
+
+    fun start() {
+        if (port == 0) {
+            logger.info("Drain HTTP endpoint disabled (GROUNDS_DRAIN_HTTP_PORT=0)")
+            return
+        }
+        val httpServer =
+            HttpServer.create(InetSocketAddress(InetAddress.getLoopbackAddress(), port), 0)
+        httpServer.createContext("/drain/start") { exchange ->
+            val deadline = deadlineSeconds(exchange.requestURI.query)
+            val started = drainManager.start(deadline)
+            respond(exchange, 200, if (started) "started" else "already-draining")
+        }
+        httpServer.createContext("/drain/players") { exchange ->
+            respond(exchange, 200, drainManager.playersRemaining().toString())
+        }
+        httpServer.createContext("/drain/status") { exchange ->
+            respond(
+                exchange,
+                200,
+                """{"draining":${drainManager.isDraining},"players":${drainManager.playersRemaining()}}""",
+            )
+        }
+        httpServer.start()
+        server = httpServer
+        logger.info("Drain HTTP endpoint listening (port={})", port)
+    }
+
+    fun stop() {
+        server?.stop(0)
+        server = null
+    }
+
+    private fun respond(exchange: HttpExchange, status: Int, body: String) {
+        val bytes = body.toByteArray(StandardCharsets.UTF_8)
+        exchange.use {
+            it.responseHeaders.set("Content-Type", "text/plain; charset=utf-8")
+            it.sendResponseHeaders(status, bytes.size.toLong())
+            it.responseBody.write(bytes)
+        }
+    }
+
+    companion object {
+        const val DEFAULT_DEADLINE_SECONDS = 600L
+
+        /**
+         * Clamped rather than rejected: the caller is a shell one-liner, not a client we argue
+         * with.
+         */
+        internal fun deadlineSeconds(query: String?): Long {
+            val raw =
+                query
+                    ?.split('&')
+                    ?.firstOrNull { it.startsWith("deadlineSeconds=") }
+                    ?.substringAfter('=')
+                    ?.toLongOrNull() ?: return DEFAULT_DEADLINE_SECONDS
+            return raw.coerceIn(10L, 86_400L)
+        }
+    }
+}

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

@@ -0,0 +1,33 @@
+package gg.grounds.drain
+
+import com.velocitypowered.api.event.ResultedEvent
+import com.velocitypowered.api.event.Subscribe
+import com.velocitypowered.api.event.connection.LoginEvent
+import com.velocitypowered.api.event.player.ServerPreConnectEvent
+
+class DrainListener(private val drainManager: DrainManager) {
+
+    /**
+     * A draining proxy takes no new players. The deny message is what a client sees in the rare
+     * window where a connection still reaches this pod after it left the Service's endpoints —
+     * reconnecting through the public name lands them on a live proxy.
+     */
+    @Subscribe
+    fun onLogin(event: LoginEvent) {
+        if (!drainManager.isDraining) return
+        event.result = ResultedEvent.ComponentResult.denied(DrainManager.RESTART_MESSAGE)
+    }
+
+    /**
+     * The moment a round is over, its players head back towards a lobby — on a draining proxy that
+     * connect becomes the transfer to another proxy instead. Every path to a lobby runs through
+     * this event: an explicit connection request, a kick-redirect, a plugin's fireAndForget.
+     */
+    @Subscribe
+    fun onServerPreConnect(event: ServerPreConnectEvent) {
+        val target = event.result.server.orElse(null) ?: return
+        if (drainManager.interceptConnect(event.player, target.serverInfo.name)) {
+            event.result = ServerPreConnectEvent.ServerResult.denied()
+        }
+    }
+}

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

@@ -0,0 +1,124 @@
+package gg.grounds.drain
+
+import com.velocitypowered.api.network.ProtocolVersion
+import com.velocitypowered.api.proxy.Player
+import com.velocitypowered.api.proxy.ProxyServer
+import java.net.InetSocketAddress
+import java.util.concurrent.TimeUnit
+import net.kyori.adventure.text.Component
+import org.slf4j.Logger
+
+/**
+ * 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 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 everyone still here is transferred regardless, because a transfer that ends a
+ *   round early still beats the kick that is otherwise seconds away.
+ *
+ * "Inside a round" is decided by the server's `grounds/server-type` role: anything that is not the
+ * lobby role defers the transfer. A server that discovery has no role for cannot be a protected
+ * round.
+ */
+class DrainManager(
+    private val plugin: Any,
+    private val proxy: ProxyServer,
+    private val logger: Logger,
+    private val config: DrainConfig,
+    private val serverRole: (String) -> String?,
+    private val lobbyValue: String,
+) {
+    @Volatile
+    var isDraining: Boolean = false
+        private set
+
+    fun playersRemaining(): Int = proxy.allPlayers.size
+
+    /** Starts the drain. Idempotent — a second call reports `false` and changes nothing. */
+    @Synchronized
+    fun start(deadlineSeconds: Long): Boolean {
+        if (isDraining) return false
+        isDraining = true
+        logger.info(
+            "Drain started (players={}, deadline={}s, transferTarget={})",
+            proxy.allPlayers.size,
+            deadlineSeconds,
+            config.transferHost?.let { "$it:${config.transferPort}" } ?: "<none — wait only>",
+        )
+
+        proxy.allPlayers.forEach { player ->
+            if (!shouldDefer(roleOf(player), lobbyValue)) {
+                transferOut(player, force = false)
+            }
+        }
+
+        proxy.scheduler
+            .buildTask(plugin, Runnable { onDeadline() })
+            .delay(deadlineSeconds, TimeUnit.SECONDS)
+            .schedule()
+        return true
+    }
+
+    /**
+     * A player is heading to [targetServer] — when draining and the target is not a round, send
+     * them to another proxy instead. Returns true when the connect should be cancelled.
+     */
+    fun interceptConnect(player: Player, targetServer: String): Boolean {
+        if (!isDraining) return false
+        if (shouldDefer(serverRole(targetServer), lobbyValue)) return false
+        return transferOut(player, force = false)
+    }
+
+    private fun onDeadline() {
+        val remaining = proxy.allPlayers
+        if (remaining.isEmpty()) return
+        logger.warn("Drain deadline reached with {} players left; transferring all", remaining.size)
+        remaining.forEach { transferOut(it, force = true) }
+    }
+
+    /**
+     * 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
+     * keep playing until the deadline, and the deadline path disconnects them with an honest
+     * message rather than the raw proxy-shutdown kick.
+     */
+    private fun transferOut(player: Player, force: Boolean): Boolean {
+        val host = config.transferHost
+        val transferable =
+            host != null && player.protocolVersion >= ProtocolVersion.MINECRAFT_1_20_5
+        if (transferable) {
+            logger.info(
+                "Draining player via transfer (player={}, target={}:{})",
+                player.username,
+                host,
+                config.transferPort,
+            )
+            player.transferToHost(InetSocketAddress.createUnresolved(host!!, config.transferPort))
+            return true
+        }
+        if (force) {
+            player.disconnect(RESTART_MESSAGE)
+            return true
+        }
+        return false
+    }
+
+    private fun roleOf(player: Player): String? =
+        player.currentServer.map { it.serverInfo.name }.orElse(null)?.let(serverRole)
+
+    companion object {
+        val RESTART_MESSAGE: Component =
+            Component.text("This proxy is restarting — please reconnect.")
+
+        /**
+         * A transfer is deferred only for players on a server whose role is a real, non-lobby role:
+         * that is where a round can be running. No server or no role means nothing to protect.
+         */
+        @JvmStatic
+        fun shouldDefer(role: String?, lobbyValue: String): Boolean =
+            role != null && role != lobbyValue
+    }
+}