From 56dc67f732c5fdf81d5c49d424953fd33650008a Mon Sep 17 00:00:00 2001 From: Lukas Jost Date: Sat, 31 Oct 2026 20:35:20 +0100 Subject: [PATCH 1/2] fix(drain): retry bounded handoffs before the shutdown deadline --- .../kotlin/gg/grounds/drain/DrainManager.kt | 81 +++++++++++-------- .../gg/grounds/drain/RetryingTransfers.kt | 33 ++++++++ .../gg/grounds/drain/DrainHandoffTest.kt | 5 +- .../gg/grounds/drain/RetryingTransfersTest.kt | 58 +++++++++++++ 4 files changed, 140 insertions(+), 37 deletions(-) create mode 100644 velocity/src/main/kotlin/gg/grounds/drain/RetryingTransfers.kt create mode 100644 velocity/src/test/kotlin/gg/grounds/drain/RetryingTransfersTest.kt diff --git a/velocity/src/main/kotlin/gg/grounds/drain/DrainManager.kt b/velocity/src/main/kotlin/gg/grounds/drain/DrainManager.kt index 88a2392..fcfe7cc 100644 --- a/velocity/src/main/kotlin/gg/grounds/drain/DrainManager.kt +++ b/velocity/src/main/kotlin/gg/grounds/drain/DrainManager.kt @@ -9,7 +9,6 @@ import gg.grounds.i18n.Translations import gg.grounds.presentation.VelocitySystemMessages import gg.grounds.proxy.api.PlayerSessionTransfer import gg.grounds.proxy.api.ProxyServiceRegistry -import java.net.InetSocketAddress import java.util.UUID import java.util.concurrent.TimeUnit import org.slf4j.Logger @@ -78,6 +77,23 @@ class DrainManager( */ @Volatile private var deadlinePassed = false + private val retryQueue = + RetryingTransfers( + config.transfersPerSecond, + { it.uniqueId }, + { it.isActive && !shouldDefer(roleOf(it), lobbyValue) }, + { player -> + runCatching { transferOut(player, deadlinePassed) } + .getOrElse { + logger.warn("Failed to transfer draining player {}", player.username, it) + if (deadlinePassed) player.disconnect(restartMessage(player)) + deadlinePassed + } + }, + ) + private var tickerRunning = false + private val sentTransfers = java.util.concurrent.ConcurrentHashMap.newKeySet() + fun playersRemaining(): Int = proxy.allPlayers.size /** Starts the drain. Idempotent — a second call reports `false` and changes nothing. */ @@ -135,24 +151,18 @@ 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. */ + @Synchronized 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) + players.forEach(retryQueue::offer) + if (tickerRunning) return + tickerRunning = true + ticker.everySecondUntilDone { + synchronized(this) { + val done = retryQueue.tick() + if (done) tickerRunning = false + done + } + } } /** @@ -162,8 +172,11 @@ class DrainManager( * message rather than the raw proxy-shutdown kick. */ private fun transferOut(player: Player, force: Boolean): Boolean { + if (sentTransfers.contains(player.uniqueId)) return true + if (transferStager.isPending(player.uniqueId.toString())) return true val host = config.transferHost if (host != null && player.protocolVersion >= ProtocolVersion.MINECRAFT_1_20_5) { + var issued = true val transfer = { logger.info( "Draining player via transfer (player={}, target={}:{})", @@ -176,19 +189,22 @@ class DrainManager( player.uniqueId, host, config.transferPort, - direct = { - player.transferToHost( - InetSocketAddress.createUnresolved(host, config.transferPort) - ) - }, refuse = { - logger.warn( - "Session handoff unavailable for draining player {}; disconnecting", - player.username, - ) - player.disconnect(restartMessage(player)) + if (deadlinePassed) { + logger.warn( + "Drain deadline reached without transfer capacity (player={})", + player.username, + ) + player.disconnect(restartMessage(player)) + } else { + issued = false + // Global issuance budget is shared by all draining sources. Retry + // later; do not mint a ticket until this actual send attempt. + pace(listOf(player), force = false) + } }, ) + if (issued) sentTransfers.add(player.uniqueId) } val payload = currentStaticServerName(player)?.let(drainTransferCookie::encode) if (payload != null) { @@ -202,7 +218,7 @@ class DrainManager( } else { transfer() } - return true + return issued } if (force) { player.disconnect(restartMessage(player)) @@ -251,18 +267,15 @@ class DrainManager( /** * Sends a draining player on through plugin-player's session handoff when it is installed: the * destination proxy admits a transfer only with the single-use token that handoff stores in a - * cookie, so a bare transfer packet ends in "This transfer could not be verified". Without - * plugin-player the bare transfer is all there is. When the handoff cannot be issued, the player is - * [refuse]d with the restart message rather than sent to a proxy that will turn them away. + * cookie. Missing plugin-player or capacity calls [refuse]; no bare transfer can bypass admission. */ internal fun transferWithHandoff( session: PlayerSessionTransfer?, playerId: UUID, host: String, port: Int, - direct: () -> Unit, refuse: () -> Unit, ) { - if (session == null) return direct() + if (session == null) return refuse() if (!session.transferToHost(playerId, host, port)) refuse() } diff --git a/velocity/src/main/kotlin/gg/grounds/drain/RetryingTransfers.kt b/velocity/src/main/kotlin/gg/grounds/drain/RetryingTransfers.kt new file mode 100644 index 0000000..aa6c788 --- /dev/null +++ b/velocity/src/main/kotlin/gg/grounds/drain/RetryingTransfers.kt @@ -0,0 +1,33 @@ +package gg.grounds.drain + +/** A bounded-per-tick, deduplicated queue. A refused handoff remains at its source. */ +internal class RetryingTransfers( + private val perSecond: Int, + private val key: (T) -> K, + private val stillDue: (T) -> Boolean, + private val transfer: (T) -> Boolean, +) { + private val queue = LinkedHashMap() + + init { + require(perSecond >= 1) + } + + val remaining: Int + @Synchronized get() = queue.size + + @Synchronized + fun offer(player: T) { + queue.putIfAbsent(key(player), player) + } + + @Synchronized + fun tick(): Boolean { + // Snapshot also prevents the one refused entry from spinning in the same tick. + val batch = queue.entries.take(perSecond).map { it.key to it.value } + for ((id, player) in batch) { + if (!stillDue(player) || transfer(player)) queue.remove(id) + } + return queue.isEmpty() + } +} diff --git a/velocity/src/test/kotlin/gg/grounds/drain/DrainHandoffTest.kt b/velocity/src/test/kotlin/gg/grounds/drain/DrainHandoffTest.kt index ba5a352..2d3e829 100644 --- a/velocity/src/test/kotlin/gg/grounds/drain/DrainHandoffTest.kt +++ b/velocity/src/test/kotlin/gg/grounds/drain/DrainHandoffTest.kt @@ -15,7 +15,6 @@ class DrainHandoffTest { player, "eu.geo.grnds.io", 25565, - direct = { calls += "direct" }, refuse = { calls += "refuse" }, ) @@ -35,8 +34,8 @@ class DrainHandoffTest { } @Test - fun `without plugin-player the bare transfer is used`() { + fun `without plugin-player the transfer is refused`() { drain(null) - assertEquals(listOf("direct"), calls) + assertEquals(listOf("refuse"), calls) } } diff --git a/velocity/src/test/kotlin/gg/grounds/drain/RetryingTransfersTest.kt b/velocity/src/test/kotlin/gg/grounds/drain/RetryingTransfersTest.kt new file mode 100644 index 0000000..86d1344 --- /dev/null +++ b/velocity/src/test/kotlin/gg/grounds/drain/RetryingTransfersTest.kt @@ -0,0 +1,58 @@ +package gg.grounds.drain + +import org.junit.jupiter.api.Assertions.* +import org.junit.jupiter.api.Test + +class RetryingTransfersTest { + @Test + fun `full shared budget queues once and retries without minting another ticket`() { + var available = false + val sent = mutableListOf() + val queue = + RetryingTransfers( + 2, + { it }, + { true }, + { + if (available) { + sent += it + true + } else false + }, + ) + queue.offer(1) + queue.offer(1) + queue.offer(2) + assertFalse(queue.tick()) + assertEquals(2, queue.remaining) + assertTrue(sent.isEmpty()) + available = true + assertTrue(queue.tick()) + assertEquals(listOf(1, 2), sent) + } + + @Test + fun `one tick never spins indefinitely on a refused transfer`() { + var attempts = 0 + val queue = + RetryingTransfers( + 20, + { it }, + { true }, + { + attempts++ + false + }, + ) + queue.offer(1) + assertFalse(queue.tick()) + assertEquals(1, attempts) + } + + @Test + fun `left players and newly protected rounds are removed without sending`() { + val queue = RetryingTransfers(2, { it }, { false }, { error("not due") }) + queue.offer(1) + assertTrue(queue.tick()) + } +} From 7066cb2c2de4b9bd44e2f14515c2423908fd817c Mon Sep 17 00:00:00 2001 From: Lukas Jost Date: Sat, 31 Oct 2026 20:44:30 +0100 Subject: [PATCH 2/2] fix: address transfer admission review regressions --- .../gg/grounds/drain/RetryingTransfers.kt | 4 ++++ .../gg/grounds/drain/RetryingTransfersTest.kt | 22 +++++++++++++++++++ 2 files changed, 26 insertions(+) diff --git a/velocity/src/main/kotlin/gg/grounds/drain/RetryingTransfers.kt b/velocity/src/main/kotlin/gg/grounds/drain/RetryingTransfers.kt index aa6c788..7e6f676 100644 --- a/velocity/src/main/kotlin/gg/grounds/drain/RetryingTransfers.kt +++ b/velocity/src/main/kotlin/gg/grounds/drain/RetryingTransfers.kt @@ -27,6 +27,10 @@ internal class RetryingTransfers( val batch = queue.entries.take(perSecond).map { it.key to it.value } for ((id, player) in batch) { if (!stillDue(player) || transfer(player)) queue.remove(id) + else { + queue.remove(id) + queue[id] = player + } } return queue.isEmpty() } diff --git a/velocity/src/test/kotlin/gg/grounds/drain/RetryingTransfersTest.kt b/velocity/src/test/kotlin/gg/grounds/drain/RetryingTransfersTest.kt index 86d1344..c58f45c 100644 --- a/velocity/src/test/kotlin/gg/grounds/drain/RetryingTransfersTest.kt +++ b/velocity/src/test/kotlin/gg/grounds/drain/RetryingTransfersTest.kt @@ -4,6 +4,28 @@ import org.junit.jupiter.api.Assertions.* import org.junit.jupiter.api.Test class RetryingTransfersTest { + @Test + fun `persistently refused heads do not starve eligible players behind them`() { + val sent = mutableListOf() + val queue = + RetryingTransfers( + 20, + { it }, + { true }, + { + if (it <= 20) false + else { + sent += it + true + } + }, + ) + (1..25).forEach(queue::offer) + queue.tick() + queue.tick() + assertEquals((21..25).toList(), sent) + } + @Test fun `full shared budget queues once and retries without minting another ticket`() { var available = false