fix(drain): retry bounded handoffs before the shutdown deadline

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

@@ -9,7 +9,6 @@
 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 @@
      */
     @Volatile private var deadlinePassed = false
 
+    private val retryQueue =
+        RetryingTransfers<Player, UUID>(
+            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<UUID>()
+
     fun playersRemaining(): Int = proxy.allPlayers.size
 
     /** Starts the drain. Idempotent — a second call reports `false` and changes nothing. */
@@ -135,24 +151,18 @@
      * 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<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)
+        players.forEach(retryQueue::offer)
+        if (tickerRunning) return
+        tickerRunning = true
+        ticker.everySecondUntilDone {
+            synchronized(this) {
+                val done = retryQueue.tick()
+                if (done) tickerRunning = false
+                done
+            }
+        }
     }
 
     /**
@@ -161,9 +171,12 @@
      * 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 {
+        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 @@
                     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 @@
             } else {
                 transfer()
             }
-            return true
+            return issued
         }
         if (force) {
             player.disconnect(restartMessage(player))
@@ -251,18 +267,15 @@
 /**
  * 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()
 }

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

@@ -0,0 +1,37 @@
+package gg.grounds.drain
+
+/** A bounded-per-tick, deduplicated queue. A refused handoff remains at its source. */
+internal class RetryingTransfers<T, K>(
+    private val perSecond: Int,
+    private val key: (T) -> K,
+    private val stillDue: (T) -> Boolean,
+    private val transfer: (T) -> Boolean,
+) {
+    private val queue = LinkedHashMap<K, T>()
+
+    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)
+            else {
+                queue.remove(id)
+                queue[id] = player
+            }
+        }
+        return queue.isEmpty()
+    }
+}

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)
     }
 }

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

@@ -0,0 +1,80 @@
+package gg.grounds.drain
+
+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<Int>()
+        val queue =
+            RetryingTransfers<Int, Int>(
+                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
+        val sent = mutableListOf<Int>()
+        val queue =
+            RetryingTransfers<Int, Int>(
+                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<Int, Int>(
+                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<Int, Int>(2, { it }, { false }, { error("not due") })
+        queue.offer(1)
+        assertTrue(queue.tick())
+    }
+}