diff --git a/README.md b/README.md index 3a6c5ec..8068ec0 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,17 @@ deployments do not need to set anything. | `GROUNDS_AGONES_POLL_INTERVAL` | `2s` | Accepts `Ns`, `Nm`, `Nh` | | `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_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: + +```text +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. Typical Helm chart wiring uses a `ConfigMap` consumed via `envFrom`, plus `POD_NAMESPACE` from the Downward API for clusters where the proxy should diff --git a/velocity/src/main/kotlin/gg/grounds/GroundsPluginAgones.kt b/velocity/src/main/kotlin/gg/grounds/GroundsPluginAgones.kt index 1c8090e..ddf9aaa 100644 --- a/velocity/src/main/kotlin/gg/grounds/GroundsPluginAgones.kt +++ b/velocity/src/main/kotlin/gg/grounds/GroundsPluginAgones.kt @@ -13,6 +13,7 @@ import gg.grounds.drain.DrainConfig import gg.grounds.drain.DrainHttpServer import gg.grounds.drain.DrainListener import gg.grounds.drain.DrainManager +import gg.grounds.drain.DrainTransferCookie import gg.grounds.gameserver.GameServerStateManager import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers @@ -53,8 +54,18 @@ constructor(private val proxyServer: ProxyServer, private val logger: Logger) { stateManager = GameServerStateManager(this, proxyServer, logger, coroutineScope).also { it.start() } + val drainCookie = DrainTransferCookie(System.getenv(DrainTransferCookie.SECRET_ENV)) + lateinit var drainManager: DrainManager discoveryService = - DiscoveryService(this, proxyServer, logger, discoveryConfig).also { it.start() } + DiscoveryService( + this, + proxyServer, + logger, + discoveryConfig, + drainTransferCookie = drainCookie, + sourceCookiePending = { playerId -> drainManager.isCookiePending(playerId) }, + ) + .also { it.start() } proxyServer.commandManager.register( proxyServer.commandManager.metaBuilder("agones").build(), @@ -62,7 +73,7 @@ constructor(private val proxyServer: ProxyServer, private val logger: Logger) { ) val drainConfig = DrainConfig.fromEnv() - val drainManager = + drainManager = DrainManager( this, proxyServer, @@ -70,6 +81,7 @@ constructor(private val proxyServer: ProxyServer, private val logger: Logger) { drainConfig, { serverName -> discoveryService.getServerRole(serverName) }, discoveryConfig.lobbyValue, + drainCookie, ) proxyServer.eventManager.register(this, DrainListener(drainManager)) drainHttpServer = diff --git a/velocity/src/main/kotlin/gg/grounds/discovery/DiscoveryConfig.kt b/velocity/src/main/kotlin/gg/grounds/discovery/DiscoveryConfig.kt index 0134cbb..2238227 100644 --- a/velocity/src/main/kotlin/gg/grounds/discovery/DiscoveryConfig.kt +++ b/velocity/src/main/kotlin/gg/grounds/discovery/DiscoveryConfig.kt @@ -1,6 +1,9 @@ package gg.grounds.discovery import java.time.Duration +import java.util.Locale + +internal fun canonicalServerName(name: String): String = name.lowercase(Locale.ROOT) /** * Discovery configuration sourced from environment variables. All keys are optional; the defaults @@ -19,7 +22,11 @@ import java.time.Duration * - `GROUNDS_AGONES_ADDRESS_TYPE` — Which `status.addresses` entry to use (`PodIP`, `ExternalIP`, * `InternalIP`, `Hostname`). * - `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. */ +data class StaticServer(val name: String, val host: String, val port: Int) + data class DiscoveryConfig( val namespace: String, val labelSelector: String, @@ -29,6 +36,7 @@ data class DiscoveryConfig( val pollInterval: Duration, val addressType: String, val port: Int, + val staticServers: List, ) { companion object { const val DEFAULT_NAMESPACE = "games" @@ -59,6 +67,7 @@ data class DiscoveryConfig( ?: DEFAULT_POLL_INTERVAL, addressType = env["GROUNDS_AGONES_ADDRESS_TYPE"] ?: DEFAULT_ADDRESS_TYPE, port = env["GROUNDS_AGONES_PORT"]?.toIntOrNull() ?: DEFAULT_PORT, + staticServers = parseStaticServers(env["GROUNDS_STATIC_SERVERS"]), ) private val DURATION_PATTERN = Regex("""^(\d+)\s*(s|m|h)$""") @@ -77,5 +86,34 @@ data class DiscoveryConfig( else -> error("unreachable") } } + + private fun parseStaticServers(raw: String?): List { + if (raw.isNullOrBlank()) return emptyList() + + val names = mutableSetOf() + return raw.split(",").map { entry -> + val trimmedEntry = entry.trim() + val separator = trimmedEntry.indexOf('=') + require(separator >= 0) { "Invalid GROUNDS_STATIC_SERVERS entry '$trimmedEntry'" } + + val name = trimmedEntry.substring(0, separator).trim() + val address = trimmedEntry.substring(separator + 1).trim() + val portSeparator = address.lastIndexOf(':') + require(name.isNotEmpty() && portSeparator >= 0) { + "Invalid GROUNDS_STATIC_SERVERS entry '$trimmedEntry'" + } + + val host = address.substring(0, portSeparator).trim() + val port = address.substring(portSeparator + 1).trim().toIntOrNull() + require(host.isNotEmpty() && port != null && port in 1..65535) { + "Invalid GROUNDS_STATIC_SERVERS entry '$trimmedEntry'" + } + require(names.add(canonicalServerName(name))) { + "Invalid GROUNDS_STATIC_SERVERS entry '$trimmedEntry': duplicate name '$name'" + } + + StaticServer(name, host, port) + } + } } } diff --git a/velocity/src/main/kotlin/gg/grounds/discovery/DiscoveryPlayerListener.kt b/velocity/src/main/kotlin/gg/grounds/discovery/DiscoveryPlayerListener.kt index d4147a8..33be4ec 100644 --- a/velocity/src/main/kotlin/gg/grounds/discovery/DiscoveryPlayerListener.kt +++ b/velocity/src/main/kotlin/gg/grounds/discovery/DiscoveryPlayerListener.kt @@ -1,27 +1,77 @@ package gg.grounds.discovery +import com.velocitypowered.api.event.Continuation +import com.velocitypowered.api.event.EventTask 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.CookieReceiveEvent import com.velocitypowered.api.event.player.PlayerChooseInitialServerEvent +import com.velocitypowered.api.network.ProtocolVersion import com.velocitypowered.api.proxy.ProxyServer import com.velocitypowered.api.proxy.server.RegisteredServer +import com.velocitypowered.api.scheduler.ScheduledTask +import gg.grounds.drain.DrainTransferCookie +import java.util.UUID +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.TimeUnit import net.kyori.adventure.text.Component +internal fun selectDrainStaticServer( + serverName: String, + servers: Collection, + serverRole: (String) -> String?, +): RegisteredServer? = + servers.firstOrNull { server -> + canonicalServerName(server.serverInfo.name) == canonicalServerName(serverName) && + serverRole(server.serverInfo.name) == STATIC_SERVER_ROLE + } + +private const val STATIC_SERVER_ROLE = "static" + +internal fun shouldDenyInitialLogin( + hasLobby: Boolean, + hasStatic: Boolean, + protocolVersion: ProtocolVersion, +): Boolean = !hasLobby && !(hasStatic && protocolVersion >= ProtocolVersion.MINECRAFT_1_20_5) + +internal fun consumeDrainTransferCookie(clearCookie: () -> Unit) { + try { + clearCookie() + } catch (_: Exception) { + // The destination choice and its continuation must not depend on client cookie storage. + } +} + class DiscoveryPlayerListener( + private val plugin: Any, private val proxyServer: ProxyServer, private val lobbyServers: Set, + private val serverRole: (String) -> String?, /** * Network-wide players per backend server, or null when the network cannot be asked. Null falls * back to this proxy's own view — on a single proxy that is the same number, and with several * it still spreads, just per proxy rather than per network. */ private val networkCounts: () -> Map?, + private val drainTransferCookie: DrainTransferCookie = DrainTransferCookie(), + private val sourceCookiePending: (String) -> Boolean = { false }, ) { + private val pendingCookies = ConcurrentHashMap() @Subscribe fun onLogin(event: LoginEvent) { - if (findLobbyServer() != null) return + if ( + !shouldDenyInitialLogin( + findLobbyServer() != null, + proxyServer.allServers.any { server -> + serverRole(server.serverInfo.name) == STATIC_SERVER_ROLE + }, + event.player.protocolVersion, + ) + ) { + return + } event.result = ResultedEvent.ComponentResult.denied( @@ -32,13 +82,64 @@ class DiscoveryPlayerListener( } @Subscribe - fun onPlayerChooseInitialServer(event: PlayerChooseInitialServerEvent) { - if (event.initialServer.isPresent) return + fun onPlayerChooseInitialServer(event: PlayerChooseInitialServerEvent): EventTask = + EventTask.withContinuation { continuation -> + if (event.initialServer.isPresent) { + continuation.resume() + return@withContinuation + } + val player = event.player + if (player.protocolVersion < ProtocolVersion.MINECRAFT_1_20_5) { + chooseServer(event, null) + continuation.resume() + return@withContinuation + } - val lobby = findLobbyServer() - if (lobby != null) { - event.setInitialServer(lobby) + val pending = PendingCookieRequest(event, continuation) + pendingCookies.put(player.uniqueId, pending)?.complete(null) + try { + player.requestCookie(DrainTransferCookie.KEY) + pending.timeoutTask = + proxyServer.scheduler + .buildTask( + plugin, + Runnable { + if (pendingCookies.remove(player.uniqueId, pending)) + pending.complete(null) + }, + ) + .delay(COOKIE_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS) + .schedule() + } catch (error: IllegalArgumentException) { + if (pendingCookies.remove(player.uniqueId, pending)) pending.complete(null) + } } + + // This must run before DrainListener: source-proxy cookie echoes belong to its stager, not to + // initial-server selection. Velocity invokes higher priorities first. + @Subscribe(priority = DRAIN_COOKIE_SOURCE_SUPPRESSION_PRIORITY) + fun onCookieReceive(event: CookieReceiveEvent) { + if (event.originalKey != DrainTransferCookie.KEY) return + if (sourceCookiePending(event.player.uniqueId.toString())) return + event.result = CookieReceiveEvent.ForwardResult.handled() + val payload = event.originalData + consumeDrainTransferCookie { + event.player.storeCookie(DrainTransferCookie.KEY, byteArrayOf()) + } + pendingCookies.remove(event.player.uniqueId)?.complete(payload) + } + + private fun chooseServer(event: PlayerChooseInitialServerEvent, payload: ByteArray?) { + if (event.initialServer.isPresent) return + val preferred = + drainTransferCookie.decode(payload)?.let { serverName -> + selectDrainStaticServer(serverName, proxyServer.allServers, serverRole) + } + if (preferred != null) { + event.setInitialServer(preferred) + return + } + findLobbyServer()?.let(event::setInitialServer) } private fun findLobbyServer(): RegisteredServer? { @@ -57,4 +158,25 @@ class DiscoveryPlayerListener( val chosen = LobbySelection.pick(candidates) ?: return null return lobbies.firstOrNull { it.serverInfo.name == chosen } } + + private inner class PendingCookieRequest( + private val event: PlayerChooseInitialServerEvent, + private val continuation: Continuation, + ) { + var timeoutTask: ScheduledTask? = null + + fun complete(payload: ByteArray?) { + timeoutTask?.cancel() + try { + chooseServer(event, payload) + } finally { + continuation.resume() + } + } + } + + private companion object { + private const val COOKIE_TIMEOUT_MILLIS = 1_000L + private const val DRAIN_COOKIE_SOURCE_SUPPRESSION_PRIORITY: Short = 100 + } } diff --git a/velocity/src/main/kotlin/gg/grounds/discovery/DiscoveryService.kt b/velocity/src/main/kotlin/gg/grounds/discovery/DiscoveryService.kt index 3f9921b..1274603 100644 --- a/velocity/src/main/kotlin/gg/grounds/discovery/DiscoveryService.kt +++ b/velocity/src/main/kotlin/gg/grounds/discovery/DiscoveryService.kt @@ -16,11 +16,37 @@ import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.TimeUnit import org.slf4j.Logger +internal fun staleManagedServerNames( + runningAgonesServerNames: Set, + agonesManagedServerNames: Set, +): Set = agonesManagedServerNames - runningAgonesServerNames + +private fun createCustomObjectsApi(config: DiscoveryConfig, logger: Logger): CustomObjectsApi? = + try { + val client = Config.defaultClient() + Configuration.setDefaultApiClient(client) + CustomObjectsApi(client) + } catch (error: Throwable) { + logger.warn( + "Failed to initialize Agones discovery client (namespace={}, labelSelector={})", + config.namespace, + config.labelSelector, + error, + ) + null + } + class DiscoveryService( private val plugin: Any, private val proxyServer: ProxyServer, private val logger: Logger, private val config: DiscoveryConfig = DiscoveryConfig.fromEnv(), + private val kubernetesClientFactory: () -> CustomObjectsApi? = { + createCustomObjectsApi(config, logger) + }, + private val drainTransferCookie: gg.grounds.drain.DrainTransferCookie = + gg.grounds.drain.DrainTransferCookie(), + private val sourceCookiePending: (String) -> Boolean = { false }, ) { private val gson = Gson() private lateinit var customObjectsApi: CustomObjectsApi @@ -28,15 +54,19 @@ class DiscoveryService( private lateinit var pollTask: ScheduledTask private val lobbyServers: MutableSet = ConcurrentHashMap.newKeySet() private val serverRoles: MutableMap = ConcurrentHashMap() + private val agonesManagedServers: MutableMap = ConcurrentHashMap() + private val staticServerNames = + config.staticServers.mapTo(mutableSetOf()) { canonicalServerName(it.name) } @Volatile private var countsSnapshot: Pair?>? = null fun start() { - customObjectsApi = createCustomObjectsApi() ?: return + unregisterBakedInPlaceholders() + registerStaticServers() + registerListeners() + + customObjectsApi = kubernetesClientFactory() ?: return // Same client, already configured as the default above. coreApi = CoreV1Api() - - unregisterPreconfiguredServers() - registerListeners() schedulePolling() } @@ -46,31 +76,41 @@ class DiscoveryService( } } - fun getServerRole(serverName: String): String? = serverRoles[serverName] + fun getServerRole(serverName: String): String? = serverRoles[canonicalServerName(serverName)] - private fun createCustomObjectsApi(): CustomObjectsApi? { - return try { - val client = Config.defaultClient() - Configuration.setDefaultApiClient(client) - CustomObjectsApi(client) - } catch (error: Throwable) { - logger.warn( - "Failed to initialize Agones discovery client (namespace={}, labelSelector={})", - config.namespace, - config.labelSelector, - error, + private fun unregisterBakedInPlaceholders() { + for (server in proxyServer.allServers.filter(::isBakedInPlaceholder)) { + proxyServer.unregisterServer(server.serverInfo) + logger.info( + "Removed baked-in placeholder server successfully (serverName={})", + server.serverInfo.name, ) - null } } - private fun unregisterPreconfiguredServers() { - val configuredServers = proxyServer.allServers.toList() - for (server in configuredServers) { - proxyServer.unregisterServer(server.serverInfo) + private fun isBakedInPlaceholder(server: RegisteredServer): Boolean { + val serverInfo = server.serverInfo + return BAKED_IN_PLACEHOLDERS.any { placeholder -> + canonicalServerName(placeholder.name) == canonicalServerName(serverInfo.name) && + placeholder.host == serverInfo.address.hostString && + placeholder.port == serverInfo.address.port + } + } + + private fun registerStaticServers() { + for (server in config.staticServers) { + proxyServer.registerServer( + ServerInfo( + server.name, + InetSocketAddress.createUnresolved(server.host, server.port), + ) + ) + serverRoles[canonicalServerName(server.name)] = STATIC_SERVER_ROLE logger.info( - "Removed pre-configured server successfully (serverName={})", - server.serverInfo.name, + "Registered static proxy server successfully (serverName={}, host={}, port={})", + server.name, + server.host, + server.port, ) } } @@ -78,7 +118,15 @@ class DiscoveryService( private fun registerListeners() { proxyServer.eventManager.register( plugin, - DiscoveryPlayerListener(proxyServer, lobbyServers, this::networkCountsCached), + DiscoveryPlayerListener( + plugin, + proxyServer, + lobbyServers, + this::getServerRole, + this::networkCountsCached, + drainTransferCookie, + sourceCookiePending, + ), ) } @@ -109,7 +157,8 @@ class DiscoveryService( private fun updateRegisteredGameServers() { val runningGameServers = fetchRunningGameServers() - val currentServers = proxyServer.allServers.associateBy { it.serverInfo.name } + val currentServers = + proxyServer.allServers.associateBy { canonicalServerName(it.serverInfo.name) } registerRunningServers(runningGameServers, currentServers) unregisterServersThatAreNoLongerRunning(runningGameServers, currentServers) @@ -156,7 +205,7 @@ class DiscoveryService( } } - private fun registerRunningServers( + internal fun registerRunningServers( runningGameServers: List, currentServers: Map, ) { @@ -173,47 +222,67 @@ class DiscoveryService( continue } - val serverType = resolveServerType(metadata.labels) ?: continue - serverRoles[serverName] = serverType + val canonicalName = canonicalServerName(serverName) + if (canonicalName in staticServerNames) { + logger.warn( + "Skipping Agones GameServer because its name collides with a static server (serverName={})", + serverName, + ) + continue + } - if (serverType == config.lobbyValue) { - lobbyServers.add(serverName) - } else { - lobbyServers.remove(serverName) + val serverType = resolveServerType(metadata.labels) ?: continue + val currentServer = currentServers[canonicalName] + if (currentServer != null && agonesManagedServers[canonicalName] !== currentServer) { + agonesManagedServers.remove(canonicalName)?.let { ownedServer -> + lobbyServers.remove(ownedServer.serverInfo.name) + serverRoles.remove(canonicalName) + } + continue } - if (serverName in currentServers) continue + if (currentServer == null) { + // Agones does not always publish the pod's address in the GameServer's + // status. The bundle's lobby fleets carry Hostname, InternalIP AND + // PodIP; the fleets forge renders for a pushed gamemode carry only the + // first two. A proxy that insists on PodIP therefore throws away every + // pushed gamemode — the server runs, is Ready, and no player can ever + // reach it, which looks exactly like a broken game. + // + // The pod is the source of that address anyway, and Agones names it + // after the GameServer, so fall back to reading it directly. Never the + // node's InternalIP: that would route players to a machine instead of + // to their server. + val address = + gameServer.status + ?.addresses + ?.firstOrNull { it.type == config.addressType } + ?.address ?: podIp(serverName) + if (address == null) { + logger.error( + "Failed to register Agones GameServer (serverName={}, reason=missing_address, addressType={})", + serverName, + config.addressType, + ) + continue + } - // Agones does not always publish the pod's address in the GameServer's - // status. The bundle's lobby fleets carry Hostname, InternalIP AND - // PodIP; the fleets forge renders for a pushed gamemode carry only the - // first two. A proxy that insists on PodIP therefore throws away every - // pushed gamemode — the server runs, is Ready, and no player can ever - // reach it, which looks exactly like a broken game. - // - // The pod is the source of that address anyway, and Agones names it - // after the GameServer, so fall back to reading it directly. Never the - // node's InternalIP: that would route players to a machine instead of - // to their server. - val address = - gameServer.status?.addresses?.firstOrNull { it.type == config.addressType }?.address - ?: podIp(serverName) - if (address == null) { - logger.error( - "Failed to register Agones GameServer (serverName={}, reason=missing_address, addressType={})", + val serverInfo = ServerInfo(serverName, InetSocketAddress(address, config.port)) + val registeredServer = proxyServer.registerServer(serverInfo) + agonesManagedServers[canonicalName] = registeredServer + logger.info( + "Registered proxy server successfully (serverName={}, serverType={})", serverName, - config.addressType, + serverType, ) - continue } - val serverInfo = ServerInfo(serverName, InetSocketAddress(address, config.port)) - proxyServer.registerServer(serverInfo) - logger.info( - "Registered proxy server successfully (serverName={}, serverType={})", - serverName, - serverType, - ) + serverRoles[canonicalName] = serverType + if (serverType == config.lobbyValue) { + lobbyServers.add(serverName) + } else { + lobbyServers.remove(serverName) + } } } @@ -228,22 +297,30 @@ class DiscoveryService( else -> labels[config.lobbyLabel] } - private fun unregisterServersThatAreNoLongerRunning( + internal fun unregisterServersThatAreNoLongerRunning( runningGameServers: List, currentServers: Map, ) { - val runningServerNames = runningGameServers.mapNotNull { it.metadata?.name }.toSet() + val runningServerNames = + runningGameServers + .mapNotNull { it.metadata?.name } + .mapTo(mutableSetOf(), ::canonicalServerName) - for (server in currentServers.values) { - if (server.serverInfo.name !in runningServerNames) { - proxyServer.unregisterServer(server.serverInfo) - lobbyServers.remove(server.serverInfo.name) - serverRoles.remove(server.serverInfo.name) - logger.info( - "Unregistered proxy server successfully (serverName={})", - server.serverInfo.name, - ) - } + for (canonicalName in + staleManagedServerNames(runningServerNames, agonesManagedServers.keys)) { + val ownedServer = agonesManagedServers[canonicalName] ?: continue + currentServers[canonicalName] + ?.takeIf { it === ownedServer } + ?.let { server -> + proxyServer.unregisterServer(server.serverInfo) + logger.info( + "Unregistered proxy server successfully (serverName={})", + ownedServer.serverInfo.name, + ) + } + lobbyServers.remove(ownedServer.serverInfo.name) + serverRoles.remove(canonicalName) + agonesManagedServers.remove(canonicalName, ownedServer) } } @@ -251,6 +328,13 @@ class DiscoveryService( private const val GROUP = "agones.dev" private const val VERSION = "v1" private const val PLURAL = "gameservers" + private const val STATIC_SERVER_ROLE = "static" + private val BAKED_IN_PLACEHOLDERS = + setOf( + StaticServer("lobby", "127.0.0.1", 30066), + StaticServer("factions", "127.0.0.1", 30067), + StaticServer("minigames", "127.0.0.1", 30068), + ) private val COUNTS_TTL_NANOS = TimeUnit.SECONDS.toNanos(2) } } diff --git a/velocity/src/main/kotlin/gg/grounds/drain/DrainListener.kt b/velocity/src/main/kotlin/gg/grounds/drain/DrainListener.kt index a5787f4..ab4c609 100644 --- a/velocity/src/main/kotlin/gg/grounds/drain/DrainListener.kt +++ b/velocity/src/main/kotlin/gg/grounds/drain/DrainListener.kt @@ -3,6 +3,7 @@ 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.CookieReceiveEvent import com.velocitypowered.api.event.player.ServerPreConnectEvent class DrainListener(private val drainManager: DrainManager) { @@ -30,4 +31,12 @@ class DrainListener(private val drainManager: DrainManager) { event.result = ServerPreConnectEvent.ServerResult.denied() } } + + @Subscribe + fun onCookieReceive(event: CookieReceiveEvent) { + if (event.originalKey != DrainTransferCookie.KEY) return + if (drainManager.handleCookie(event.player, event.originalData)) { + event.result = CookieReceiveEvent.ForwardResult.handled() + } + } } diff --git a/velocity/src/main/kotlin/gg/grounds/drain/DrainManager.kt b/velocity/src/main/kotlin/gg/grounds/drain/DrainManager.kt index 45e9cd5..90dd74b 100644 --- a/velocity/src/main/kotlin/gg/grounds/drain/DrainManager.kt +++ b/velocity/src/main/kotlin/gg/grounds/drain/DrainManager.kt @@ -8,6 +8,20 @@ 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. * @@ -21,8 +35,8 @@ import org.slf4j.Logger * 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: anything that is not the - * lobby role defers the transfer. A server that discovery has no role for cannot be a protected + * "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 * round. */ class DrainManager( @@ -32,6 +46,10 @@ class DrainManager( private val config: DrainConfig, private val serverRole: (String) -> String?, private val lobbyValue: String, + private val drainTransferCookie: DrainTransferCookie = DrainTransferCookie(), + private val transferStager: DrainTransferStager = DrainTransferStager { action -> + proxy.scheduler.buildTask(plugin, Runnable(action)).delay(1, TimeUnit.SECONDS).schedule() + }, ) { @Volatile var isDraining: Boolean = false @@ -51,16 +69,21 @@ class DrainManager( config.transferHost?.let { "$it:${config.transferPort}" } ?: "", ) - proxy.allPlayers.forEach { player -> - if (!shouldDefer(roleOf(player), lobbyValue)) { - transferOut(player, force = false) - } - } - proxy.scheduler .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, + ) + }, + ) return true } @@ -83,7 +106,13 @@ class DrainManager( "Drain deadline reached; transferring {} players not inside a round", drainable.size, ) - drainable.forEach { transferOut(it, force = true) } + transferAllSafely( + drainable, + { player -> transferOut(player, force = true) }, + { player, error -> + logger.warn("Failed to transfer draining player {}", player.username, error) + }, + ) } if (inRound.isNotEmpty()) { logger.warn( @@ -102,16 +131,28 @@ class DrainManager( */ 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)) + if (host != null && player.protocolVersion >= ProtocolVersion.MINECRAFT_1_20_5) { + val transfer = { + logger.info( + "Draining player via transfer (player={}, target={}:{})", + player.username, + host, + config.transferPort, + ) + player.transferToHost(InetSocketAddress.createUnresolved(host, config.transferPort)) + } + val payload = currentStaticServerName(player)?.let(drainTransferCookie::encode) + if (payload != null) { + transferStager.stage( + player.uniqueId.toString(), + payload, + { player.storeCookie(DrainTransferCookie.KEY, payload) }, + { player.requestCookie(DrainTransferCookie.KEY) }, + transfer, + ) + } else { + transfer() + } return true } if (force) { @@ -124,16 +165,37 @@ class DrainManager( private fun roleOf(player: Player): String? = player.currentServer.map { it.serverInfo.name }.orElse(null)?.let(serverRole) + private fun currentStaticServerName(player: Player): String? = + player.currentServer + .map { it.serverInfo.name } + .orElse(null) + ?.takeIf { serverName -> shouldPreserveStaticBackend(serverRole(serverName)) } + + fun handleCookie(player: Player, payload: ByteArray?): Boolean { + val playerId = player.uniqueId.toString() + if (!transferStager.isPending(playerId)) return false + transferStager.onCookie(playerId, payload) + return true + } + + fun isCookiePending(playerId: String): Boolean = transferStager.isPending(playerId) + 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. + * A transfer is deferred only for players on a real round server. No server, an unknown + * role, a lobby, or a static server means nothing to protect. */ @JvmStatic - fun shouldDefer(role: String?, lobbyValue: String): Boolean = - role != null && role != lobbyValue + @Suppress("UNUSED_PARAMETER") + fun shouldDefer(role: String?, lobbyValue: String): Boolean = role in ROUND_ROLES + + @JvmStatic + fun shouldPreserveStaticBackend(role: String?): Boolean = role == STATIC_SERVER_ROLE + + private val ROUND_ROLES = setOf("game", "match") + private const val STATIC_SERVER_ROLE = "static" } } diff --git a/velocity/src/main/kotlin/gg/grounds/drain/DrainTransferCookie.kt b/velocity/src/main/kotlin/gg/grounds/drain/DrainTransferCookie.kt new file mode 100644 index 0000000..c89be71 --- /dev/null +++ b/velocity/src/main/kotlin/gg/grounds/drain/DrainTransferCookie.kt @@ -0,0 +1,83 @@ +package gg.grounds.drain + +import java.nio.ByteBuffer +import java.nio.charset.StandardCharsets +import java.security.MessageDigest +import java.time.Clock +import javax.crypto.Mac +import javax.crypto.spec.SecretKeySpec +import net.kyori.adventure.key.Key + +/** Signed, short-lived destination hint used only by an automatic proxy drain transfer. */ +class DrainTransferCookie(secret: String? = null, private val clock: Clock = Clock.systemUTC()) { + private val secret = secret?.takeIf { it.isNotBlank() }?.toByteArray(StandardCharsets.UTF_8) + + fun encode( + serverName: String, + expiresAtMillis: Long = clock.millis() + LIFETIME_MILLIS, + ): ByteArray? { + val key = secret ?: return null + val name = serverName.toByteArray(StandardCharsets.UTF_8) + if (name.isEmpty() || name.size > MAX_SERVER_NAME_BYTES) return null + val issuedAtMillis = clock.millis() + val body = + ByteBuffer.allocate(HEADER_BYTES + name.size) + .put(VERSION) + .putLong(issuedAtMillis) + .putLong(expiresAtMillis) + .put(name) + .array() + return body + sign(body, key) + } + + /** Returns null for expired, malformed, or unsupported client-controlled payloads. */ + fun decode(payload: ByteArray?): String? { + val key = secret ?: return null + if (payload == null || payload.size !in (HEADER_BYTES + MAC_BYTES + 1)..MAX_PAYLOAD_BYTES) { + return null + } + val body = payload.copyOfRange(0, payload.size - MAC_BYTES) + val signature = payload.copyOfRange(payload.size - MAC_BYTES, payload.size) + if (!MessageDigest.isEqual(sign(body, key), signature)) return null + + val bytes = ByteBuffer.wrap(body) + if (bytes.get() != VERSION) return null + val issuedAtMillis = bytes.long + val expiresAtMillis = bytes.long + val now = clock.millis() + if ( + issuedAtMillis > now + CLOCK_SKEW_MILLIS || + expiresAtMillis <= now || + expiresAtMillis > now + LIFETIME_MILLIS + CLOCK_SKEW_MILLIS || + expiresAtMillis <= issuedAtMillis || + expiresAtMillis > issuedAtMillis + LIFETIME_MILLIS + CLOCK_SKEW_MILLIS + ) { + return null + } + + val name = ByteArray(bytes.remaining()) + bytes.get(name) + val serverName = name.toString(StandardCharsets.UTF_8) + return serverName.takeIf { + it.isNotBlank() && it.toByteArray(StandardCharsets.UTF_8).size == name.size + } + } + + private fun sign(body: ByteArray, key: ByteArray): ByteArray = + Mac.getInstance("HmacSHA256").run { + init(SecretKeySpec(key, algorithm)) + doFinal(body) + } + + companion object { + val KEY: Key = Key.key("grounds", "drain-static-server") + const val SECRET_ENV = "VELOCITY_FORWARDING_SECRET" + private const val VERSION: Byte = 1 + private const val HEADER_BYTES = 1 + Long.SIZE_BYTES + Long.SIZE_BYTES + private const val MAC_BYTES = 32 + private const val MAX_SERVER_NAME_BYTES = 64 + private const val MAX_PAYLOAD_BYTES = HEADER_BYTES + MAX_SERVER_NAME_BYTES + MAC_BYTES + private const val LIFETIME_MILLIS = 30_000L + private const val CLOCK_SKEW_MILLIS = 5_000L + } +} diff --git a/velocity/src/main/kotlin/gg/grounds/drain/DrainTransferStager.kt b/velocity/src/main/kotlin/gg/grounds/drain/DrainTransferStager.kt new file mode 100644 index 0000000..dd27c43 --- /dev/null +++ b/velocity/src/main/kotlin/gg/grounds/drain/DrainTransferStager.kt @@ -0,0 +1,50 @@ +package gg.grounds.drain + +import java.security.MessageDigest +import java.util.concurrent.ConcurrentHashMap + +/** + * Coordinates store/request/echo before a drain transfer, with a bounded normal-transfer fallback. + */ +class DrainTransferStager(private val scheduleTimeout: ((() -> Unit) -> Unit)) { + private val pending = ConcurrentHashMap() + + fun stage( + playerId: String, + payload: ByteArray, + store: () -> Unit, + request: () -> Unit, + transfer: () -> Unit, + ) { + val stage = Pending(payload, request, transfer) + if (pending.putIfAbsent(playerId, stage) != null) return + try { + store() + request() + scheduleTimeout { complete(playerId, stage) } + } catch (_: Exception) { + complete(playerId, stage) + } + } + + fun onCookie(playerId: String, payload: ByteArray?) { + val stage = pending[playerId] ?: return + if (payload != null && MessageDigest.isEqual(stage.payload, payload)) { + complete(playerId, stage) + } else { + try { + stage.request() + } catch (_: Exception) { + complete(playerId, stage) + } + } + } + + fun isPending(playerId: String): Boolean = pending.containsKey(playerId) + + private fun complete(playerId: String, stage: Pending) { + if (pending.remove(playerId, stage)) stage.transfer() + } + + private class Pending(val payload: ByteArray, val request: () -> Unit, val transfer: () -> Unit) +} diff --git a/velocity/src/test/kotlin/gg/grounds/discovery/DiscoveryConfigTest.kt b/velocity/src/test/kotlin/gg/grounds/discovery/DiscoveryConfigTest.kt index 4de0bc1..fa2f54f 100644 --- a/velocity/src/test/kotlin/gg/grounds/discovery/DiscoveryConfigTest.kt +++ b/velocity/src/test/kotlin/gg/grounds/discovery/DiscoveryConfigTest.kt @@ -19,6 +19,7 @@ class DiscoveryConfigTest { assertEquals(Duration.ofSeconds(2), cfg.pollInterval) assertEquals("PodIP", cfg.addressType) assertEquals(25565, cfg.port) + assertEquals(emptyList(), cfg.staticServers) } @Test @@ -121,4 +122,96 @@ class DiscoveryConfigTest { assertEquals(25565, cfg.port) assertEquals(Duration.ofSeconds(2), cfg.pollInterval) } + + @Test + fun `static servers parse comma separated name host and port entries`() { + assertEquals( + listOf( + StaticServer("buildserver", "buildserver", 25565), + StaticServer("metrics", "metrics.stage.svc.cluster.local", 25566), + ), + DiscoveryConfig.fromEnv( + env = + mapOf( + "GROUNDS_STATIC_SERVERS" to + " buildserver=buildserver:25565, metrics=metrics.stage.svc.cluster.local:25566 " + ) + ) + .staticServers, + ) + } + + @Test + fun `static servers reject entries without a name address separator`() { + assertThrows(IllegalArgumentException::class.java) { + DiscoveryConfig.fromEnv(env = mapOf("GROUNDS_STATIC_SERVERS" to "buildserver:25565")) + } + } + + @Test + fun `static servers reject entries with an empty name`() { + assertThrows(IllegalArgumentException::class.java) { + DiscoveryConfig.fromEnv(env = mapOf("GROUNDS_STATIC_SERVERS" to "=buildserver:25565")) + } + } + + @Test + fun `static servers reject entries with an empty host`() { + assertThrows(IllegalArgumentException::class.java) { + DiscoveryConfig.fromEnv(env = mapOf("GROUNDS_STATIC_SERVERS" to "buildserver=:25565")) + } + } + + @Test + fun `static servers reject entries with port zero`() { + assertThrows(IllegalArgumentException::class.java) { + DiscoveryConfig.fromEnv( + env = mapOf("GROUNDS_STATIC_SERVERS" to "buildserver=buildserver:0") + ) + } + } + + @Test + fun `static servers reject entries with ports above 65535`() { + assertThrows(IllegalArgumentException::class.java) { + DiscoveryConfig.fromEnv( + env = mapOf("GROUNDS_STATIC_SERVERS" to "buildserver=buildserver:65536") + ) + } + } + + @Test + fun `static servers reject entries with non numeric ports`() { + assertThrows(IllegalArgumentException::class.java) { + DiscoveryConfig.fromEnv( + env = mapOf("GROUNDS_STATIC_SERVERS" to "buildserver=buildserver:abc") + ) + } + } + + @Test + fun `static servers reject duplicate names`() { + assertThrows(IllegalArgumentException::class.java) { + DiscoveryConfig.fromEnv( + env = + mapOf( + "GROUNDS_STATIC_SERVERS" to + "buildserver=buildserver:25565,buildserver=other:25566" + ) + ) + } + } + + @Test + fun `static servers reject duplicate names that differ only by case`() { + assertThrows(IllegalArgumentException::class.java) { + DiscoveryConfig.fromEnv( + env = + mapOf( + "GROUNDS_STATIC_SERVERS" to + "BuildServer=buildserver:25565,buildserver=other:25566" + ) + ) + } + } } diff --git a/velocity/src/test/kotlin/gg/grounds/discovery/DiscoveryServiceTest.kt b/velocity/src/test/kotlin/gg/grounds/discovery/DiscoveryServiceTest.kt new file mode 100644 index 0000000..eebcb3d --- /dev/null +++ b/velocity/src/test/kotlin/gg/grounds/discovery/DiscoveryServiceTest.kt @@ -0,0 +1,163 @@ +package gg.grounds.discovery + +import com.velocitypowered.api.event.EventManager +import com.velocitypowered.api.proxy.ProxyServer +import com.velocitypowered.api.proxy.server.RegisteredServer +import com.velocitypowered.api.proxy.server.ServerInfo +import java.lang.reflect.Proxy +import java.net.InetSocketAddress +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import org.slf4j.LoggerFactory + +class DiscoveryServiceTest { + + @Test + fun `startup removes only the exact baked image placeholders`() { + val removed = mutableListOf() + val placeholder = registeredServer("lobby", "127.0.0.1", 30066) + val sameNameDifferentAddress = registeredServer("lobby", "10.0.0.10", 25565) + val external = registeredServer("external", "10.0.0.11", 25565) + val service = + DiscoveryService( + plugin = Any(), + proxyServer = + proxyServer(listOf(placeholder, sameNameDifferentAddress, external), removed), + logger = LoggerFactory.getLogger(javaClass), + config = DiscoveryConfig.fromEnv(emptyMap()), + kubernetesClientFactory = { null }, + ) + + service.start() + + assertEquals(listOf(placeholder.serverInfo), removed) + } + + @Test + fun `static servers register when Kubernetes initialization fails`() { + val registrations = mutableListOf() + val service = + DiscoveryService( + plugin = Any(), + proxyServer = proxyServer(emptyList(), registrations = registrations), + logger = LoggerFactory.getLogger(javaClass), + config = + DiscoveryConfig.fromEnv( + mapOf("GROUNDS_STATIC_SERVERS" to "buildserver=buildserver:25565") + ), + kubernetesClientFactory = { null }, + ) + + service.start() + + assertEquals( + listOf( + ServerInfo("buildserver", InetSocketAddress.createUnresolved("buildserver", 25565)) + ), + registrations, + ) + assertEquals("static", service.getServerRole("buildserver")) + } + + @Test + fun `replacement of an Agones registration is never unregistered as owned`() { + val agonesRegistration = registeredServer("game", "10.0.0.1", 25565) + val externalReplacement = registeredServer("game", "10.0.0.2", 25565) + val removed = mutableListOf() + val service = + DiscoveryService( + plugin = Any(), + proxyServer = + proxyServer( + emptyList(), + removed, + registrations = mutableListOf(), + registered = agonesRegistration, + ), + logger = LoggerFactory.getLogger(javaClass), + config = DiscoveryConfig.fromEnv(emptyMap()), + kubernetesClientFactory = { null }, + ) + val gameServer = + GameServer( + metadata = + Metadata(name = "game", labels = mapOf("grounds/server-type" to "lobby")), + status = + Status( + state = "Ready", + addresses = listOf(GameServerAddress("10.0.0.1", "PodIP")), + ), + ) + + service.registerRunningServers(listOf(gameServer), emptyMap()) + service.registerRunningServers( + listOf( + gameServer.copy( + metadata = + Metadata(name = "game", labels = mapOf("grounds/server-type" to "game")) + ) + ), + mapOf("game" to externalReplacement), + ) + assertEquals(null, service.getServerRole("game")) + service.unregisterServersThatAreNoLongerRunning( + emptyList(), + mapOf("game" to externalReplacement), + ) + + assertTrue(removed.isEmpty()) + } + + private fun proxyServer( + servers: List, + removed: MutableList = mutableListOf(), + registrations: MutableList = mutableListOf(), + registered: RegisteredServer? = null, + ): ProxyServer = + proxy( + mapOf( + "getAllServers" to servers, + "getEventManager" to proxy(), + "unregisterServer" to + { args: Array -> + removed.add(args.single() as ServerInfo) + }, + "registerServer" to + { args: Array -> + val serverInfo = args.single() as ServerInfo + registrations.add(serverInfo) + registered + ?: registeredServer( + serverInfo.name, + serverInfo.address.hostString, + serverInfo.address.port, + ) + }, + ) + ) + + private fun registeredServer(name: String, host: String, port: Int): RegisteredServer = + proxy( + mapOf( + "getServerInfo" to ServerInfo(name, InetSocketAddress.createUnresolved(host, port)) + ) + ) + + @Suppress("UNCHECKED_CAST") + private inline fun proxy(responses: Map = emptyMap()): T = + Proxy.newProxyInstance(T::class.java.classLoader, arrayOf(T::class.java)) { _, method, args + -> + val response = responses[method.name] + when { + response is Function1<*, *> -> + (response as (Array) -> Any?)(args.orEmpty()) + response != null -> response + method.returnType == Boolean::class.javaPrimitiveType -> false + method.returnType == Int::class.javaPrimitiveType -> 0 + method.returnType == Long::class.javaPrimitiveType -> 0L + method.returnType == Void.TYPE -> null + else -> null + } + } as T +} diff --git a/velocity/src/test/kotlin/gg/grounds/discovery/DrainStaticServerSelectionTest.kt b/velocity/src/test/kotlin/gg/grounds/discovery/DrainStaticServerSelectionTest.kt new file mode 100644 index 0000000..1e7f030 --- /dev/null +++ b/velocity/src/test/kotlin/gg/grounds/discovery/DrainStaticServerSelectionTest.kt @@ -0,0 +1,67 @@ +package gg.grounds.discovery + +import com.velocitypowered.api.network.ProtocolVersion +import com.velocitypowered.api.proxy.server.RegisteredServer +import com.velocitypowered.api.proxy.server.ServerInfo +import java.lang.reflect.Proxy +import java.net.InetSocketAddress +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Assertions.assertSame +import org.junit.jupiter.api.Test + +class DrainStaticServerSelectionTest { + + @Test + fun `selects a registered static server named by the drain cookie`() { + val buildserver = registeredServer("buildserver") + + val selected = + selectDrainStaticServer("buildserver", listOf(buildserver)) { name -> + if (name == "buildserver") "static" else null + } + + assertSame(buildserver, selected) + } + + @Test + fun `matches a static server name from the drain cookie case insensitively`() { + val buildserver = registeredServer("BuildServer") + + val selected = selectDrainStaticServer("buildserver", listOf(buildserver)) { "static" } + + assertSame(buildserver, selected) + } + + @Test + fun `does not select an Agones round server named by the drain cookie`() { + val game = registeredServer("game-7") + + val selected = selectDrainStaticServer("game-7", listOf(game)) { "game" } + + assertNull(selected) + } + + @Test + fun `allows a cookie-capable login to reach static selection without a lobby`() { + assertFalse( + shouldDenyInitialLogin( + hasLobby = false, + hasStatic = true, + protocolVersion = ProtocolVersion.MINECRAFT_1_20_5, + ) + ) + } + + private fun registeredServer(name: String): RegisteredServer = + Proxy.newProxyInstance( + RegisteredServer::class.java.classLoader, + arrayOf(RegisteredServer::class.java), + ) { _, method, _ -> + when (method.name) { + "getServerInfo" -> + ServerInfo(name, InetSocketAddress.createUnresolved("$name.internal", 25565)) + else -> null + } + } as RegisteredServer +} diff --git a/velocity/src/test/kotlin/gg/grounds/discovery/DrainTransferCookieConsumptionTest.kt b/velocity/src/test/kotlin/gg/grounds/discovery/DrainTransferCookieConsumptionTest.kt new file mode 100644 index 0000000..992660b --- /dev/null +++ b/velocity/src/test/kotlin/gg/grounds/discovery/DrainTransferCookieConsumptionTest.kt @@ -0,0 +1,22 @@ +package gg.grounds.discovery + +import org.junit.jupiter.api.Assertions.assertDoesNotThrow +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +class DrainTransferCookieConsumptionTest { + + @Test + fun `attempts to clear a received drain cookie even when the client clear fails`() { + var attempts = 0 + + assertDoesNotThrow { + consumeDrainTransferCookie { + attempts++ + throw IllegalStateException("client disconnected") + } + } + + assertEquals(1, attempts) + } +} diff --git a/velocity/src/test/kotlin/gg/grounds/discovery/ServerOwnershipTest.kt b/velocity/src/test/kotlin/gg/grounds/discovery/ServerOwnershipTest.kt new file mode 100644 index 0000000..f0ed7db --- /dev/null +++ b/velocity/src/test/kotlin/gg/grounds/discovery/ServerOwnershipTest.kt @@ -0,0 +1,20 @@ +package gg.grounds.discovery + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +class ServerOwnershipTest { + + @Test + fun `only missing Agones owned servers become stale`() { + assertEquals( + setOf("old-game"), + staleManagedServerNames(setOf("live-game"), setOf("live-game", "old-game")), + ) + } + + @Test + fun `an unowned static server cannot enter the removal result`() { + assertEquals(emptySet(), staleManagedServerNames(emptySet(), emptySet())) + } +} diff --git a/velocity/src/test/kotlin/gg/grounds/drain/DrainCookieListenerOrderingTest.kt b/velocity/src/test/kotlin/gg/grounds/drain/DrainCookieListenerOrderingTest.kt new file mode 100644 index 0000000..cb6082c --- /dev/null +++ b/velocity/src/test/kotlin/gg/grounds/drain/DrainCookieListenerOrderingTest.kt @@ -0,0 +1,31 @@ +package gg.grounds.drain + +import com.velocitypowered.api.event.Subscribe +import com.velocitypowered.api.event.player.CookieReceiveEvent +import gg.grounds.discovery.DiscoveryPlayerListener +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +class DrainCookieListenerOrderingTest { + + @Test + fun `source cookie suppression runs before drain cookie completion`() { + val discoveryPriority = + DiscoveryPlayerListener::class + .java + .getDeclaredMethod("onCookieReceive", CookieReceiveEvent::class.java) + .getAnnotation(Subscribe::class.java) + .priority + val drainPriority = + DrainListener::class + .java + .getDeclaredMethod("onCookieReceive", CookieReceiveEvent::class.java) + .getAnnotation(Subscribe::class.java) + .priority + + assertTrue( + discoveryPriority > drainPriority, + "Discovery must suppress source drain cookies before DrainListener completes them", + ) + } +} diff --git a/velocity/src/test/kotlin/gg/grounds/drain/DrainDecisionTest.kt b/velocity/src/test/kotlin/gg/grounds/drain/DrainDecisionTest.kt index b5fadc2..ce04d52 100644 --- a/velocity/src/test/kotlin/gg/grounds/drain/DrainDecisionTest.kt +++ b/velocity/src/test/kotlin/gg/grounds/drain/DrainDecisionTest.kt @@ -19,8 +19,20 @@ class DrainDecisionTest { } @Test - fun `no server or unknown role is nothing to protect`() { + fun `only real round roles defer a drain`() { + assertFalse(DrainManager.shouldDefer(role = "static", lobbyValue = "lobby")) + assertFalse(DrainManager.shouldDefer(role = "lobby", lobbyValue = "lobby")) assertFalse(DrainManager.shouldDefer(role = null, lobbyValue = "lobby")) + assertFalse(DrainManager.shouldDefer(role = "unknown", lobbyValue = "lobby")) + } + + @Test + fun `only static backends are preserved across an automatic drain transfer`() { + assertTrue(DrainManager.shouldPreserveStaticBackend("static")) + assertFalse(DrainManager.shouldPreserveStaticBackend("lobby")) + assertFalse(DrainManager.shouldPreserveStaticBackend("game")) + assertFalse(DrainManager.shouldPreserveStaticBackend("match")) + assertFalse(DrainManager.shouldPreserveStaticBackend(null)) } @Test diff --git a/velocity/src/test/kotlin/gg/grounds/drain/DrainTransferCookieTest.kt b/velocity/src/test/kotlin/gg/grounds/drain/DrainTransferCookieTest.kt new file mode 100644 index 0000000..166db8f --- /dev/null +++ b/velocity/src/test/kotlin/gg/grounds/drain/DrainTransferCookieTest.kt @@ -0,0 +1,68 @@ +package gg.grounds.drain + +import java.time.Clock +import java.time.Instant +import java.time.ZoneOffset +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Test + +class DrainTransferCookieTest { + + private val now = Instant.parse("2026-08-23T16:00:00Z") + private val clock = Clock.fixed(now, ZoneOffset.UTC) + + @Test + fun `round trips a static server name before its expiry`() { + val cookie = DrainTransferCookie("secret", clock) + + val payload = cookie.encode("buildserver") + + assertEquals("buildserver", cookie.decode(payload)) + } + + @Test + fun `rejects a cookie at its expiry`() { + val cookie = DrainTransferCookie("secret", clock) + val payload = cookie.encode("buildserver", now.toEpochMilli()) + + assertNull(cookie.decode(payload)) + } + + @Test + fun `rejects malformed client payloads`() { + val cookie = DrainTransferCookie("secret", clock) + + assertNull(cookie.decode(byteArrayOf(1, 2, 3))) + } + + @Test + fun `rejects a cookie signed with another secret`() { + val payload = DrainTransferCookie("source-secret", clock).encode("buildserver") + + assertNull(DrainTransferCookie("target-secret", clock).decode(payload)) + } + + @Test + fun `rejects a forged cookie signature`() { + val cookie = DrainTransferCookie("secret", clock) + val payload = cookie.encode("buildserver")!! + payload[10] = (payload[10].toInt() xor 1).toByte() + + assertNull(cookie.decode(payload)) + } + + @Test + fun `rejects an expiry beyond the allowed lifetime`() { + val cookie = DrainTransferCookie("secret", clock) + val payload = cookie.encode("buildserver", now.plusSeconds(600).toEpochMilli()) + + assertNull(cookie.decode(payload)) + } + + @Test + fun `blank or missing secrets disable cookie preservation`() { + assertNull(DrainTransferCookie(null, clock).encode("buildserver")) + assertNull(DrainTransferCookie(" ", clock).encode("buildserver")) + } +} diff --git a/velocity/src/test/kotlin/gg/grounds/drain/DrainTransferSafetyTest.kt b/velocity/src/test/kotlin/gg/grounds/drain/DrainTransferSafetyTest.kt new file mode 100644 index 0000000..67a6b0c --- /dev/null +++ b/velocity/src/test/kotlin/gg/grounds/drain/DrainTransferSafetyTest.kt @@ -0,0 +1,25 @@ +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/DrainTransferStagerTest.kt b/velocity/src/test/kotlin/gg/grounds/drain/DrainTransferStagerTest.kt new file mode 100644 index 0000000..ffe18bd --- /dev/null +++ b/velocity/src/test/kotlin/gg/grounds/drain/DrainTransferStagerTest.kt @@ -0,0 +1,31 @@ +package gg.grounds.drain + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +class DrainTransferStagerTest { + @Test + fun `transfers only after the stored cookie echo is confirmed`() { + val actions = mutableListOf() + val stager = DrainTransferStager { actions += "retry" } + stager.stage("player", byteArrayOf(1), { actions += "store" }, { actions += "request" }) { + actions += "transfer" + } + stager.onCookie("player", byteArrayOf(1)) + assertEquals(listOf("store", "request", "retry", "transfer"), actions) + } + + @Test + fun `mismatch retries and timeout transfers once`() { + val actions = mutableListOf() + lateinit var timeout: () -> Unit + val stager = DrainTransferStager { timeout = it } + stager.stage("player", byteArrayOf(1), { actions += "store" }, { actions += "request" }) { + actions += "transfer" + } + stager.onCookie("player", byteArrayOf(2)) + timeout() + timeout() + assertEquals(listOf("store", "request", "request", "transfer"), actions) + } +}