Retry failed player saves instead of draining them to SafeStorage

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

@@ -7,24 +7,22 @@ import world.gregs.voidps.engine.entity.character.player.Player
 import world.gregs.voidps.engine.entity.character.player.Players
 import java.lang.Runnable
 import java.util.concurrent.ConcurrentHashMap
-import kotlin.system.measureTimeMillis
+import java.util.concurrent.TimeUnit
 
 class SaveQueue(
     private val storage: Storage,
     private val fallback: Storage = storage,
     // SupervisorJob so a failed save doesn't cancel the scope and kill future saves
     private val scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob()),
+    private val retryMillis: Long = TimeUnit.MINUTES.toMillis(Settings["storage.save.retryMinutes", 5].toLong()),
 ) : Runnable {
     private val pending = ConcurrentHashMap<String, PlayerSave>()
+    private val failing = ConcurrentHashMap<String, Long>()
     private val logger = InlineLogger()
     private var job: Job? = null
 
     private val handler = CoroutineExceptionHandler { _, exception ->
-        logger.error(exception) { "Error saving players!" }
-        scope.fallback(pending.values.toList())
-    }
-    private val fallbackHandler = CoroutineExceptionHandler { _, exception ->
-        logger.error(exception) { "Fallback save failed!" }
+        logger.error(exception) { "Unexpected error in the save queue!" }
     }
 
     override fun run() {
@@ -42,27 +40,52 @@ class SaveQueue(
         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)
+        return scope.save(online + queued, retry = false)
     }
 
     suspend fun awaitInFlight() {
         job?.join()
     }
 
-    private fun CoroutineScope.save(accounts: List<PlayerSave>) = launch(handler) {
-        val took = measureTimeMillis {
-            withContext(NonCancellable) {
+    private fun CoroutineScope.save(accounts: List<PlayerSave>, retry: Boolean = true) = launch(handler) {
+        withContext(NonCancellable) {
+            val start = System.currentTimeMillis()
+            try {
                 storage.save(accounts)
-                clearPending(accounts)
+            } catch (e: Exception) {
+                failed(accounts, e, retry)
+                return@withContext
             }
+            clearPending(accounts)
+            clearFailing(accounts)
+            logger.info { "Saved ${accounts.size} ${"account".plural(accounts.size)} in ${System.currentTimeMillis() - start}ms" }
         }
-        logger.info { "Saved ${accounts.size} ${"account".plural(accounts.size)} in ${took}ms" }
     }
 
-    private fun CoroutineScope.fallback(accounts: List<PlayerSave>) = launch(fallbackHandler) {
-        withContext(NonCancellable) {
-            fallback.save(accounts)
-            clearPending(accounts)
+    private fun failed(accounts: List<PlayerSave>, exception: Exception, retry: Boolean) {
+        val exhausted = if (retry) exhausted(accounts) else accounts
+        if (exhausted.isEmpty()) {
+            logger.error(exception) { "Error saving ${accounts.size} ${"account".plural(accounts.size)}, retrying next tick." }
+            return
+        }
+        logger.error(exception) { "Giving up on ${exhausted.size} ${"account".plural(exhausted.size)}, writing to fallback storage: ${exhausted.joinToString { it.name }}" }
+        try {
+            fallback.save(exhausted)
+        } catch (e: Exception) {
+            logger.error(e) { "Fallback save failed!" }
+        }
+        clearPending(exhausted)
+        clearFailing(exhausted)
+    }
+
+    private fun exhausted(accounts: List<PlayerSave>): List<PlayerSave> {
+        val now = System.currentTimeMillis()
+        return accounts.filter { now - (failing.putIfAbsent(it.name, now) ?: now) >= retryMillis }
+    }
+
+    private fun clearFailing(accounts: List<PlayerSave>) {
+        for (account in accounts) {
+            failing.remove(account.name)
         }
     }
 

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

@@ -54,34 +54,85 @@ internal class SaveQueueTest : KoinMock() {
     }
 
     @Test
-    fun `Failed save falls back and doesn't kill the queue`() {
-        val fallbackSaved = CountDownLatch(1)
+    fun `Failed save retries against real storage and doesn't kill the queue`() {
+        val attempted = AtomicInteger()
         val saved = CountDownLatch(1)
         var fail = true
         val storage = object : TestStorage() {
             override fun save(accounts: List<PlayerSave>) {
+                attempted.incrementAndGet()
                 if (fail) {
                     throw IOException("Disk full")
                 }
                 saved.countDown()
             }
         }
+        val dumped = CopyOnWriteArrayList<String>()
         val fallback = object : TestStorage() {
             override fun save(accounts: List<PlayerSave>) {
-                fallbackSaved.countDown()
+                accounts.mapTo(dumped) { it.name }
             }
         }
         val queue = SaveQueue(storage, fallback)
         queue.save(Player(accountName = "player"))
         queue.run()
-        assertTrue(fallbackSaved.await(5, TimeUnit.SECONDS), "Fallback didn't run after failed save")
-        waitFor("fallback to clear pending") { queue.empty() }
+        waitFor("first attempt") { attempted.get() >= 1 }
         fail = false
-        queue.save(Player(accountName = "player"))
-        waitFor("save after a failure") {
+        waitFor("retry to succeed") {
             queue.run()
             saved.count == 0L
         }
+        waitFor("pending to drain") { queue.empty() }
+        assertTrue(dumped.isEmpty(), "Transient failure gave up on the first attempt")
+    }
+
+    @Test
+    fun `Failure only touches the accounts that were attempted`() {
+        val started = CountDownLatch(1)
+        val release = CountDownLatch(1)
+        val dumped = CopyOnWriteArrayList<String>()
+        val storage = object : TestStorage() {
+            override fun save(accounts: List<PlayerSave>) {
+                started.countDown()
+                release.await(5, TimeUnit.SECONDS)
+                throw IOException("Disk full")
+            }
+        }
+        val fallback = object : TestStorage() {
+            override fun save(accounts: List<PlayerSave>) {
+                accounts.mapTo(dumped) { it.name }
+            }
+        }
+        val queue = SaveQueue(storage, fallback, retryMillis = 0)
+        queue.save(Player(accountName = "attempted"))
+        queue.run()
+        assertTrue(started.await(5, TimeUnit.SECONDS), "Save didn't start")
+        queue.save(Player(accountName = "queued_later"))
+        release.countDown()
+        waitFor("attempted account to be dumped") { dumped.contains("attempted") }
+        assertFalse(dumped.contains("queued_later"), "Failure dumped an account storage was never asked to write")
+        assertTrue(queue.saving("queued_later"), "Failure dropped an account storage was never asked to write")
+    }
+
+    @Test
+    fun `Shutdown save writes to the fallback before the job completes`() {
+        val dumped = CopyOnWriteArrayList<String>()
+        val storage = object : TestStorage() {
+            override fun save(accounts: List<PlayerSave>) {
+                throw IOException("Disk full")
+            }
+        }
+        val fallback = object : TestStorage() {
+            override fun save(accounts: List<PlayerSave>) {
+                accounts.mapTo(dumped) { it.name }
+            }
+        }
+        val queue = SaveQueue(storage, fallback)
+        queue.save(Player(accountName = "player"))
+
+        runBlocking { queue.direct().join() }
+
+        assertTrue(dumped.contains("player"), "Shutdown left a failed save nowhere on disk")
     }
 
     @Test

game/src/main/resources/game.properties

@@ -422,6 +422,10 @@ storage.players.logs=./data/saves/logs/
 # How frequently to save logs to file
 storage.players.logs.seconds=10
 
+# How long to keep retrying a failed player save against real storage before
+# giving up and writing it to the failed saves directory instead
+storage.save.retryMinutes=5
+
 # The directory where failed player save files are stored
 storage.players.errors=./data/errors/