Fix dropped connections never saving the player

engine/src/main/kotlin/world/gregs/voidps/engine/client/PlayerAccountLoader.kt

@@ -11,6 +11,7 @@ import world.gregs.voidps.engine.data.Storage
 import world.gregs.voidps.engine.data.definition.AccountDefinitions
 import world.gregs.voidps.engine.entity.World
 import world.gregs.voidps.engine.entity.character.player.Player
+import world.gregs.voidps.engine.entity.character.player.Players
 import world.gregs.voidps.engine.entity.character.player.name
 import world.gregs.voidps.engine.entity.character.player.rights
 import world.gregs.voidps.engine.event.AuditLog
@@ -86,13 +87,20 @@ class PlayerAccountLoader(
     }
 
     suspend fun connect(player: Player, client: Client, displayMode: Int = 0, viewport: Boolean = true) {
-        if (!accounts.setup(player, client, displayMode, viewport)) {
-            logger.warn { "Error setting up account" }
-            client.disconnect(Response.WORLD_FULL)
-            return
-        }
         withContext(gameContext) {
             queue.await()
+            val existing = Players.findByAccount(player.accountName)
+            if (existing != null) {
+                logger.warn { "Logging out stale session for ${player.accountName} before login." }
+                accounts.logout(existing, safely = false)
+                client.disconnect(Response.ACCOUNT_ONLINE)
+                return@withContext
+            }
+            if (!accounts.setup(player, client, displayMode, viewport)) {
+                logger.warn { "Error setting up account" }
+                client.disconnect(Response.WORLD_FULL)
+                return@withContext
+            }
             logger.info { "${if (viewport) "Player" else "Bot"} logged in ${player.accountName} index ${player.index}." }
             client.login(player.name, player.index, player.rights.ordinal, member = World.members, membersWorld = World.members)
             accounts.spawn(player, client)

engine/src/main/kotlin/world/gregs/voidps/engine/data/AccountManager.kt

@@ -103,7 +103,7 @@ class AccountManager(
             player.message("You need to wait a few moments before you can log out.")
             return
         }
-        if (!Despawn.logout(player)) {
+        if (safely && !Despawn.logout(player)) {
             return
         }
         player["logged_out"] = true

engine/src/main/kotlin/world/gregs/voidps/engine/data/SaveQueue.kt

@@ -38,15 +38,22 @@ class SaveQueue(
         this.job = scope.save(pending.values.toList())
     }
 
-    fun direct(): Job = scope.save(Players.filter { !it.contains("bot") }.map { it.copy() })
+    fun direct(): Job {
+        val online = Players.filter { !it.contains("bot") }.map { it.copy() }
+        val names = online.mapTo(HashSet()) { it.name }
+        val queued = pending.values.filter { it.name !in names }
+        return scope.save(online + queued)
+    }
+
+    suspend fun awaitInFlight() {
+        job?.join()
+    }
 
     private fun CoroutineScope.save(accounts: List<PlayerSave>) = launch(handler) {
         val took = measureTimeMillis {
             withContext(NonCancellable) {
                 storage.save(accounts)
-                for (account in accounts) {
-                    pending.remove(account.name)
-                }
+                clearPending(accounts)
             }
         }
         logger.info { "Saved ${accounts.size} ${"account".plural(accounts.size)} in ${took}ms" }
@@ -55,8 +62,14 @@ class SaveQueue(
     private fun CoroutineScope.fallback(accounts: List<PlayerSave>) = launch(fallbackHandler) {
         withContext(NonCancellable) {
             fallback.save(accounts)
-            for (account in accounts) {
-                pending.remove(account.name)
+            clearPending(accounts)
+        }
+    }
+
+    private fun clearPending(accounts: List<PlayerSave>) {
+        for (account in accounts) {
+            pending.computeIfPresent(account.name) { _, current ->
+                if (current === account) null else current
             }
         }
     }

engine/src/test/kotlin/world/gregs/voidps/engine/client/PlayerAccountLoaderTest.kt

@@ -14,6 +14,7 @@ import world.gregs.voidps.engine.data.exchange.Claim
 import world.gregs.voidps.engine.data.exchange.OpenOffers
 import world.gregs.voidps.engine.data.exchange.PriceHistory
 import world.gregs.voidps.engine.entity.character.player.Player
+import world.gregs.voidps.engine.entity.character.player.Players
 import world.gregs.voidps.engine.entity.character.player.chat.clan.Clan
 import world.gregs.voidps.engine.script.KoinMock
 import world.gregs.voidps.network.Response
@@ -139,6 +140,28 @@ internal class PlayerAccountLoaderTest : KoinMock() {
         }
     }
 
+    @Test
+    fun `Can't login while an earlier session is still in the world`() = runTest {
+        mockkStatic("world.gregs.voidps.network.login.protocol.encode.LoginEncoderKt")
+        mockkObject(Players)
+        val client: Client = mockk(relaxed = true)
+        val ghost = Player(index = 7, accountName = "name")
+        every { Players.findByAccount("name") } returns ghost
+        try {
+            val player = Player(index = 4, accountName = "name", variables = mutableMapOf("display_name" to "name"))
+
+            loader.connect(player, client, 2)
+
+            coVerify {
+                accounts.logout(ghost, safely = false)
+                client.disconnect(Response.ACCOUNT_ONLINE)
+            }
+            coVerify(exactly = 0) { accounts.spawn(player, client) }
+        } finally {
+            unmockkObject(Players)
+        }
+    }
+
     @Test
     fun `World full`() = runTest {
         mockkStatic("world.gregs.voidps.network.login.protocol.encode.LoginEncoderKt")

engine/src/test/kotlin/world/gregs/voidps/engine/data/AccountManagerTest.kt

@@ -34,6 +34,7 @@ class AccountManagerTest : KoinMock() {
 
     private lateinit var manager: AccountManager
     private lateinit var connectionQueue: ConnectionQueue
+    private lateinit var saveQueue: SaveQueue
 
     override val modules = listOf(
         module {
@@ -78,9 +79,10 @@ class AccountManagerTest : KoinMock() {
             override fun load(accountName: String): PlayerSave? = null
         }
         Settings.load(mapOf("world.home.x" to "1234", "world.home.y" to "5432", "world.experienceRate" to "1.0"))
+        saveQueue = SaveQueue(storage)
         manager = AccountManager(
             accountDefinitions = AccountDefinitions(),
-            saveQueue = SaveQueue(storage),
+            saveQueue = saveQueue,
             connectionQueue = connectionQueue,
             overrides = AppearanceOverrides(),
         )
@@ -133,6 +135,23 @@ class AccountManagerTest : KoinMock() {
         }
     }
 
+    @Test
+    fun `Dropped connection still saves the session`() = runTest {
+        val client = DummyClient()
+        val player = Player(accountName = "name", tile = Tile(3200, 3200))
+        manager.setup(player, client, 0, viewport = false)
+        manager.spawn(player, client)
+
+        client.disconnect()
+        client.exit()
+        connectionQueue.run()
+        GameLoop.tick = 2
+        World.run()
+
+        assertTrue(saveQueue.saving("name"), "Dropped connection never queued a save, losing the session")
+        assertTrue(player["logged_out", false], "Dropped connection left the player logged in as a ghost")
+    }
+
     @AfterEach
     fun teardown() {
         Settings.clear()

engine/src/test/kotlin/world/gregs/voidps/engine/data/SaveQueueTest.kt

@@ -1,5 +1,6 @@
 package world.gregs.voidps.engine.data
 
+import kotlinx.coroutines.runBlocking
 import org.junit.jupiter.api.Test
 import world.gregs.voidps.engine.data.config.AccountDefinition
 import world.gregs.voidps.engine.data.exchange.Claim
@@ -8,11 +9,15 @@ import world.gregs.voidps.engine.data.exchange.PriceHistory
 import world.gregs.voidps.engine.entity.character.player.Player
 import world.gregs.voidps.engine.entity.character.player.chat.clan.Clan
 import world.gregs.voidps.engine.script.KoinMock
+import world.gregs.voidps.type.Tile
 import java.io.IOException
+import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicInteger
 import kotlin.test.assertEquals
+import kotlin.test.assertFalse
 import kotlin.test.assertTrue
 
 internal class SaveQueueTest : KoinMock() {
@@ -108,6 +113,69 @@ internal class SaveQueueTest : KoinMock() {
         waitFor("pending to drain") { queue.empty() }
     }
 
+    @Test
+    fun `Save queued during a write isn't dropped`() {
+        val started = CountDownLatch(1)
+        val release = CountDownLatch(1)
+        val blocked = AtomicBoolean(true)
+        val written = CopyOnWriteArrayList<Tile>()
+        val storage = object : TestStorage() {
+            override fun save(accounts: List<PlayerSave>) {
+                accounts.mapTo(written) { it.tile }
+                if (!blocked.getAndSet(false)) {
+                    return
+                }
+                started.countDown()
+                release.await(5, TimeUnit.SECONDS)
+            }
+        }
+        val queue = SaveQueue(storage)
+        queue.save(Player(accountName = "player", tile = Tile(1, 1)))
+        queue.run()
+        assertTrue(started.await(5, TimeUnit.SECONDS), "First save didn't start")
+        queue.save(Player(accountName = "player", tile = Tile(2, 2)))
+        release.countDown()
+        waitFor("newer snapshot to be written") {
+            queue.run()
+            written.contains(Tile(2, 2))
+        }
+        waitFor("pending to drain") { queue.empty() }
+    }
+
+    @Test
+    fun `Completed save clears pending when nothing superseded it`() {
+        val written = CopyOnWriteArrayList<String>()
+        val storage = object : TestStorage() {
+            override fun save(accounts: List<PlayerSave>) {
+                accounts.mapTo(written) { it.name }
+            }
+        }
+        val queue = SaveQueue(storage)
+        queue.save(Player(accountName = "player"))
+        waitFor("save to complete") {
+            queue.run()
+            written.contains("player")
+        }
+        waitFor("pending to drain") { queue.empty() }
+        assertFalse(queue.saving("player"))
+    }
+
+    @Test
+    fun `Shutdown save includes accounts pending from a logout`() {
+        val written = CopyOnWriteArrayList<String>()
+        val storage = object : TestStorage() {
+            override fun save(accounts: List<PlayerSave>) {
+                accounts.mapTo(written) { it.name }
+            }
+        }
+        val queue = SaveQueue(storage)
+        queue.save(Player(accountName = "logged_out_player"))
+
+        runBlocking { queue.direct().join() }
+
+        assertTrue(written.contains("logged_out_player"), "Shutdown dropped a save left pending by a logout")
+    }
+
     private fun waitFor(description: String, condition: () -> Boolean) {
         val deadline = System.currentTimeMillis() + 5000
         while (!condition()) {

game/src/main/kotlin/content/entity/player/AutoSave.kt

@@ -22,6 +22,7 @@ class AutoSave(
 
         worldDespawn {
             runBlocking {
+                saveQueue.awaitInFlight()
                 saveQueue.direct().join()
                 exchange.save()
             }

game/src/test/kotlin/content/area/karamja/tzhaar_city/TzhaarFightCaveTest.kt

@@ -18,6 +18,7 @@ import world.gregs.voidps.engine.entity.character.move.tele
 import world.gregs.voidps.engine.entity.character.npc.NPCs
 import world.gregs.voidps.engine.entity.character.player.Player
 import world.gregs.voidps.engine.entity.character.player.PlayerRights
+import world.gregs.voidps.engine.entity.character.player.Players
 import world.gregs.voidps.engine.entity.character.player.rights
 import world.gregs.voidps.engine.entity.character.player.skill.Skill
 import world.gregs.voidps.engine.entity.obj.GameObjects
@@ -183,6 +184,26 @@ class TzhaarFightCaveTest : WorldTest() {
         assertEquals(Tile(2413, 5113), player.tile)
     }
 
+    @Test
+    fun `Connection loss mid wave still logs the player out`() = runTest {
+        setRandom(object : FakeRandom() {
+            override fun nextBits(bitCount: Int): Int = 0
+        })
+        val player = createPlayer(Tile(2438, 5168), "JalYt-11")
+        player["god_mode"] = true
+        val entrance = GameObjects.find(Tile(2437, 5166), "cave_entrance_fight_cave")
+        player.interactObject(entrance, "Enter")
+        tick(5)
+        player["fight_cave_wave"] = 10
+
+        get<AccountManager>().logout(player, false)
+        tick(3)
+
+        assertTrue(player["logged_out", false], "Involuntary disconnect was vetoed instead of logging out")
+        assertNull(Players.findByAccount(player.accountName), "Involuntary disconnect left a ghost in the world")
+        assertFalse(Instances.reserved(player.tile.region), "Player should be saved on real map")
+    }
+
     @Test
     fun `Server shutdown keeps the wave and moves the player out of the instance`() {
         setRandom(object : FakeRandom() {

network/src/main/kotlin/world/gregs/voidps/network/client/Client.kt

@@ -49,15 +49,16 @@ open class Client(
         }
         disconnected = true
         write.flushAndClose()
-        state = ClientState.Disconnected
         disconnect?.invoke()
     }
 
     suspend fun exit() {
-        if (state == ClientState.Connected) {
-            state = ClientState.Disconnecting
-            disconnecting?.invoke()
+        if (state != ClientState.Connected) {
+            return
         }
+        state = ClientState.Disconnecting
+        disconnecting?.invoke()
+        state = ClientState.Disconnected
     }
 
     open fun flush() {

network/src/test/kotlin/world/gregs/voidps/network/client/ClientTest.kt

@@ -0,0 +1,55 @@
+package world.gregs.voidps.network.client
+
+import io.ktor.utils.io.*
+import kotlinx.coroutines.test.runTest
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+
+internal class ClientTest {
+
+    private fun client() = Client(ByteChannel(false), IsaacCipher(IntArray(4)), null, "127.0.0.1")
+
+    @Test
+    fun `Exit still logs out after a write error disconnected the client`() = runTest {
+        val client = client()
+        var loggedOut = false
+        client.onDisconnecting {
+            loggedOut = true
+        }
+
+        client.disconnect()
+        client.exit()
+
+        assertTrue(loggedOut, "Logout skipped because disconnect() poisoned the client state")
+    }
+
+    @Test
+    fun `Logout only runs once`() = runTest {
+        val client = client()
+        var count = 0
+        client.onDisconnecting {
+            count++
+        }
+
+        client.exit()
+        client.exit()
+
+        assertEquals(1, count)
+    }
+
+    @Test
+    fun `Disconnect callback only runs once`() = runTest {
+        val client = client()
+        var count = 0
+        client.onDisconnected {
+            count++
+        }
+
+        client.disconnect()
+        client.disconnect()
+
+        assertEquals(1, count)
+        assertTrue(client.disconnected)
+    }
+}