fix(velocity): consume the verified runtime transfer admission

common/src/main/kotlin/gg/grounds/player/presence/HttpPlayerPresenceClient.kt

@@ -196,8 +196,15 @@ class HttpPlayerPresenceClient(
         return response.statusCode() == 204
     }
 
-    fun issueHandoff(playerId: UUID, sessionId: UUID): SessionHandoff? {
-        val body = mapper.writeValueAsString(mapOf("sessionId" to sessionId.toString()))
+    fun issueHandoff(
+        playerId: UUID,
+        sessionId: UUID,
+        boundedAdmission: Boolean = false,
+    ): SessionHandoff? {
+        val body =
+            mapper.writeValueAsString(
+                mapOf("sessionId" to sessionId.toString(), "boundedAdmission" to boundedAdmission)
+            )
         val response =
             send(
                 request("/v2/players/$playerId/session/transfer")
@@ -211,13 +218,31 @@ class HttpPlayerPresenceClient(
         return SessionHandoff(token, parseInstantMillis(expiry))
     }
 
+    private fun abortHandoff(playerId: UUID, token: String, connectionId: UUID, proxyId: String) {
+        val body =
+            mapper.writeValueAsString(
+                mapOf(
+                    "playerId" to playerId.toString(),
+                    "handoffToken" to token,
+                    "connectionId" to connectionId.toString(),
+                    "destinationProxyId" to proxyId,
+                )
+            )
+        send(
+            request("/v2/players/sessions/transfer/admission/abort")
+                .header("Content-Type", APPLICATION_JSON)
+                .POST(HttpRequest.BodyPublishers.ofString(body))
+        )
+    }
+
     fun consumeHandoff(
         playerId: UUID,
         sessionId: UUID,
         token: String,
         playerName: String,
         proxyId: String,
         region: String,
+        admissionConnectionId: UUID? = null,
     ): ExactPlayerLoginResult {
         val body =
             mapper.writeValueAsString(
@@ -228,14 +253,23 @@ class HttpPlayerPresenceClient(
                     "playerName" to playerName,
                     "proxyId" to proxyId,
                     "region" to region,
+                    "connectionId" to admissionConnectionId?.toString(),
                 )
             )
         val response =
             send(
                 request("/v2/players/sessions/transfer")
                     .header("Content-Type", APPLICATION_JSON)
                     .POST(HttpRequest.BodyPublishers.ofString(body))
-            ) ?: return ExactPlayerLoginResult.Unavailable("presence service did not answer")
+            )
+        if (response == null || response.statusCode() in 500..599) {
+            if (admissionConnectionId != null)
+                abortHandoff(playerId, token, admissionConnectionId, proxyId)
+            logoutExactSession(playerId, sessionId)
+            return ExactPlayerLoginResult.Unavailable(
+                response?.let { problemDetail(it.body()) } ?: "presence service did not answer"
+            )
+        }
         return when (response.statusCode()) {
             201 -> {
                 val json = parse(response.body())
@@ -246,10 +280,14 @@ class HttpPlayerPresenceClient(
                     ExactPlayerLoginResult.Accepted(
                         AdmittedPlayerSession(sessionId, parseInstantMillis(expiry))
                     )
-                else
+                else {
+                    if (admissionConnectionId != null)
+                        abortHandoff(playerId, token, admissionConnectionId, proxyId)
+                    logoutExactSession(playerId, sessionId)
                     ExactPlayerLoginResult.Error(
                         "presence service returned an invalid transferred lease"
                     )
+                }
             }
             409 -> ExactPlayerLoginResult.AlreadyOnline
             in 500..599 -> ExactPlayerLoginResult.Unavailable(problemDetail(response.body()))

common/src/test/kotlin/gg/grounds/player/presence/HttpPlayerPresenceClientTest.kt

@@ -177,6 +177,72 @@ class HttpPlayerPresenceClientTest {
         )
     }
 
+    @Test
+    fun `reserved consume carries the runtime connection binding`() {
+        routes["/v2/players/sessions/transfer"] =
+            201 to
+                """{"playerId":"$PLAYER_ID","sessionId":"$DESTINATION_SESSION_ID","leaseExpiresAt":"2026-08-04T10:01:00Z"}"""
+        client.consumeHandoff(
+            PLAYER_ID,
+            DESTINATION_SESSION_ID,
+            "token",
+            "Notch",
+            "target",
+            "eu",
+            DESTINATION_SESSION_ID,
+        )
+        assertTrue(
+            seen["/v2/players/sessions/transfer"]!!
+                .body
+                .contains("\"connectionId\":\"$DESTINATION_SESSION_ID\"")
+        )
+    }
+
+    @Test
+    fun `uncertain reserved consume fences its admission before exact cleanup`() {
+        routes["/v2/players/sessions/transfer"] = 503 to ""
+        routes["/v2/players/sessions/transfer/admission/abort"] = 204 to ""
+        routes["/v2/players/$PLAYER_ID/session"] = 204 to ""
+        client.consumeHandoff(
+            PLAYER_ID,
+            DESTINATION_SESSION_ID,
+            "opaque-token",
+            "Notch",
+            "target",
+            "eu",
+            DESTINATION_SESSION_ID,
+        )
+        val abort = seen["/v2/players/sessions/transfer/admission/abort"]!!
+        assertEquals("POST", abort.method)
+        assertTrue(abort.body.contains("\"connectionId\":\"$DESTINATION_SESSION_ID\""))
+        assertTrue(abort.body.contains("\"destinationProxyId\":\"target\""))
+        assertEquals(
+            "sessionId=$DESTINATION_SESSION_ID",
+            seen["/v2/players/$PLAYER_ID/session"]?.query,
+        )
+    }
+
+    @Test
+    fun `unknown transfer result cleans only its exact destination lease`() {
+        routes["/v2/players/sessions/transfer"] = 503 to ""
+        routes["/v2/players/$PLAYER_ID/session"] = 204 to ""
+        val result =
+            client.consumeHandoff(
+                PLAYER_ID,
+                DESTINATION_SESSION_ID,
+                "token",
+                "Notch",
+                "target",
+                "eu",
+            )
+        assertTrue(result is ExactPlayerLoginResult.Unavailable)
+        assertEquals("DELETE", seen["/v2/players/$PLAYER_ID/session"]?.method)
+        assertEquals(
+            "sessionId=$DESTINATION_SESSION_ID",
+            seen["/v2/players/$PLAYER_ID/session"]?.query,
+        )
+    }
+
     @Test
     fun `409 is already-online, not an error`() {
         routes["/v1/players/sessions"] =

velocity/src/main/kotlin/gg/grounds/listener/PlayerConnectionListener.kt

@@ -15,6 +15,7 @@ import gg.grounds.config.MessagesConfig
 import gg.grounds.player.presence.ExactPlayerLoginResult
 import gg.grounds.player.presence.PlayerLogoutResult
 import gg.grounds.presence.PlayerPresenceService
+import gg.grounds.presence.TransferAdmissionBridge
 import java.util.Collections
 import java.util.IdentityHashMap
 import java.util.concurrent.CompletableFuture
@@ -48,7 +49,18 @@ class PlayerConnectionListener(
             if (
                 player.handshakeIntent == com.velocitypowered.api.network.HandshakeIntent.TRANSFER
             ) {
-                // A transfer is admitted only after the destination consumes the cookie handoff.
+                if (
+                    TransferAdmissionBridge.required &&
+                        runCatching { TransferAdmissionBridge.admission(player) }.getOrNull() ==
+                            null
+                ) {
+                    deny(
+                        event,
+                        Component.text("This transfer could not be verified. Please reconnect."),
+                    )
+                    return@async
+                }
+                // Ownership still waits for the final authenticated profile and exact consume.
                 pendingTransfers.add(player)
                 return@async
             }
@@ -124,6 +136,25 @@ class PlayerConnectionListener(
         handoffCookies[player] = cookie
         return EventTask.async {
             try {
+                val admission = TransferAdmissionBridge.admission(player)
+                if (TransferAdmissionBridge.required && admission == null)
+                    throw IllegalStateException("transfer admission is missing")
+                if (admission != null) {
+                    val result =
+                        playerPresenceService.consumeHandoff(
+                            player,
+                            admission.token,
+                            admission.connectionId,
+                            admission.connectionId,
+                        )
+                    if (result is ExactPlayerLoginResult.Accepted)
+                        TransferAdmissionBridge.complete(player)
+                    if (result !is ExactPlayerLoginResult.Accepted)
+                        player.disconnect(
+                            Component.text("This transfer could not be verified. Please reconnect.")
+                        )
+                    return@async
+                }
                 player.requestCookie(HANDOFF_COOKIE)
                 // Null when the sending proxy stored no handoff: a bare transfer packet.
                 val bytes = cookie.get(COOKIE_TIMEOUT_SECONDS, TimeUnit.SECONDS)

velocity/src/main/kotlin/gg/grounds/presence/PlayerPresenceService.kt

@@ -28,6 +28,8 @@ class PlayerPresenceService : AutoCloseable {
     private var client: HttpPlayerPresenceClient? = null
     private val connections =
         Collections.synchronizedMap(IdentityHashMap<Player, ConnectionLease>())
+    private val outboundTransfers =
+        Collections.newSetFromMap(Collections.synchronizedMap(IdentityHashMap<Player, Boolean>()))
     private val mapper = JsonMapper.builder().build()
 
     data class ConnectionLease(
@@ -51,10 +53,17 @@ class PlayerPresenceService : AutoCloseable {
 
     fun issueHandoff(player: Player): SessionHandoff? {
         val lease = lease(player) ?: return null
-        return withClient(null) { it.issueHandoff(lease.playerId, lease.sessionId) }
+        return withClient(null) {
+            it.issueHandoff(lease.playerId, lease.sessionId, TransferAdmissionBridge.required)
+        }
     }
 
-    fun consumeHandoff(player: Player, token: String, sessionId: UUID): ExactPlayerLoginResult {
+    fun consumeHandoff(
+        player: Player,
+        token: String,
+        sessionId: UUID,
+        admissionConnectionId: UUID? = null,
+    ): ExactPlayerLoginResult {
         val result =
             withClient(ExactPlayerLoginResult.Unavailable("presence service is not configured")) {
                 it.consumeHandoff(
@@ -64,18 +73,25 @@ class PlayerPresenceService : AutoCloseable {
                     player.username,
                     proxyId(),
                     region(),
+                    admissionConnectionId,
                 )
             }
         if (result is ExactPlayerLoginResult.Accepted) {
-            connections[player] =
+            val lease =
                 ConnectionLease(player.uniqueId, sessionId, result.session.leaseExpiresAtMillis)
+            synchronized(connections) {
+                if (player.isActive) connections[player] = lease else logoutExact(lease)
+            }
         }
         return result
     }
 
     fun lease(player: Player): ConnectionLease? = connections[player]
 
-    fun remove(player: Player): ConnectionLease? = connections.remove(player)
+    fun remove(player: Player): ConnectionLease? {
+        outboundTransfers.remove(player)
+        return connections.remove(player)
+    }
 
     fun leasesFor(players: Collection<Player>): List<ConnectionLease> = players.mapNotNull(::lease)
 
@@ -113,15 +129,22 @@ class PlayerPresenceService : AutoCloseable {
 
     fun transferToHost(player: Player, host: String, port: Int): Boolean {
         val lease = lease(player) ?: return false
-        val handoff = issueHandoff(player) ?: return false
-        player.storeCookie(
-            HANDOFF_COOKIE,
-            mapper.writeValueAsBytes(
-                mapOf("playerId" to lease.playerId.toString(), "token" to handoff.token)
-            ),
-        )
-        player.transferToHost(InetSocketAddress.createUnresolved(host, port))
-        return true
+        if (!outboundTransfers.add(player)) return false
+        var sent = false
+        try {
+            val handoff = issueHandoff(player) ?: return false
+            player.storeCookie(
+                HANDOFF_COOKIE,
+                mapper.writeValueAsBytes(
+                    mapOf("playerId" to lease.playerId.toString(), "token" to handoff.token)
+                ),
+            )
+            player.transferToHost(InetSocketAddress.createUnresolved(host, port))
+            sent = true
+            return true
+        } finally {
+            if (!sent) outboundTransfers.remove(player)
+        }
     }
 
     fun configure(baseUrl: String) {

velocity/src/main/kotlin/gg/grounds/presence/TransferAdmissionBridge.kt

@@ -0,0 +1,40 @@
+package gg.grounds.presence
+
+import com.velocitypowered.api.proxy.Player
+import java.util.UUID
+
+/** Optional runtime bridge; enabling the gate requires the patched runtime, never a fallback. */
+object TransferAdmissionBridge {
+    val required: Boolean
+        get() = System.getenv("GROUNDS_TRANSFER_ADMISSION_ENABLED") == "true"
+
+    data class Admission(val token: String, val connectionId: UUID)
+
+    private fun runtime() = Class.forName("gg.grounds.runtime.transfer.TransferAdmission")
+
+    fun admission(player: Player): Admission? {
+        if (!required) return null
+        val raw = runtime().getMethod("admission", Player::class.java).invoke(null, player)
+        return parse(raw as? Map<*, *>, player.uniqueId, System.getenv("PROXY_ID"))
+    }
+
+    internal fun parse(raw: Map<*, *>?, profile: UUID, proxy: String?): Admission? {
+        if (
+            raw?.get("playerId") != profile.toString() ||
+                raw["destinationProxyId"] != proxy ||
+                proxy.isNullOrBlank()
+        )
+            return null
+        val token =
+            (raw["handoffToken"] as? String)?.takeIf { it.matches(Regex("[A-Za-z0-9_-]{43}")) }
+                ?: return null
+        val connection =
+            runCatching { UUID.fromString(raw["connectionId"] as? String) }.getOrNull()
+                ?: return null
+        return Admission(token, connection)
+    }
+
+    fun complete(player: Player) {
+        if (required) runtime().getMethod("complete", Player::class.java).invoke(null, player)
+    }
+}