Persist Shardok battle updates asynchronously (#8718)

src/main/scala/net/eagle0/eagle/service/BUILD.bazel

@@ -651,6 +651,7 @@ scala_library(
         "GameHistoryFactory.scala",
         "HistoryBackendConfig.scala",
         "HistoryBackendStartupValidation.scala",
+        "OrderedAsyncWriter.scala",
         "PostgresGameSchemaPool.scala",
         "PostgresHistory.scala",
         "ShardokLiveBattleCache.scala",

src/main/scala/net/eagle0/eagle/service/FullGameHistory.scala

@@ -64,6 +64,9 @@ trait FullGameHistory extends GameHistory {
   /** Clear all Shardok results for a battle (used when tutorial battle resets). */
   def withResetShardokResults(shardokGameId: ShardokGameId): FullGameHistory
 
+  /** Wait until all Shardok updates accepted by this history are durable. */
+  def flushShardokWrites(): Unit = ()
+
   /**
    * Releases any in-process Shardok read projection after a battle ends. Durable histories may lazily reload it if a
    * later caller needs completed-battle data; histories without a read cache can use this default no-op.

src/main/scala/net/eagle0/eagle/service/OrderedAsyncWriter.scala

@@ -0,0 +1,73 @@
+package net.eagle0.eagle.service
+
+import java.util.concurrent.{CompletableFuture, CompletionException, Executor}
+
+/**
+ * A bounded, ordered write-behind queue. Work for one writer is serialized without occupying an executor thread while
+ * it waits for earlier work. A failed write poisons the queue so callers cannot continue advancing in-memory state
+ * after persistence has stopped.
+ */
+private[service] final class OrderedAsyncWriter[A](
+    write: A => Unit,
+    executor: Executor,
+    maxPending: Int
+) {
+  require(maxPending > 0, s"maxPending must be positive, got $maxPending")
+
+  private val lock                          = new Object
+  private var tail: CompletableFuture[Void] = CompletableFuture.completedFuture(null)
+  private var pending                       = 0
+  private var accepting                     = true
+  private var failure: Option[Throwable]    = None
+
+  def enqueue(item: A): Unit = lock.synchronized {
+    ensureAcceptingAndHealthy()
+    while pending >= maxPending && accepting && failure.isEmpty do
+      try lock.wait()
+      catch {
+        case e: InterruptedException =>
+          Thread.currentThread().interrupt()
+          throw new RuntimeException("Interrupted while waiting for asynchronous persistence capacity", e)
+      }
+    ensureAcceptingAndHealthy()
+
+    pending += 1
+    val writeFuture = tail.thenRunAsync(() => write(item), executor)
+    tail = writeFuture.whenComplete { (_, error) =>
+      lock.synchronized {
+        pending -= 1
+        if error != null && failure.isEmpty then failure = Some(unwrap(error))
+        lock.notifyAll()
+      }
+    }
+  }
+
+  def flush(): Unit = await(lock.synchronized(tail))
+
+  def close(): Unit = {
+    val finalTail = lock.synchronized {
+      accepting = false
+      lock.notifyAll()
+      tail
+    }
+    await(finalTail)
+  }
+
+  private[service] def pendingCount: Int = lock.synchronized(pending)
+
+  private def ensureAcceptingAndHealthy(): Unit = {
+    failure.foreach(throw _)
+    if !accepting then throw new IllegalStateException("Asynchronous persistence queue is closed")
+  }
+
+  private def await(future: CompletableFuture[Void]): Unit =
+    try future.join(): Unit
+    catch {
+      case e: CompletionException => throw unwrap(e)
+    }
+
+  private def unwrap(error: Throwable): Throwable = error match {
+    case e: CompletionException if e.getCause != null => unwrap(e.getCause)
+    case other                                        => other
+  }
+}

src/main/scala/net/eagle0/eagle/service/PostgresHistory.scala

@@ -1,10 +1,12 @@
 package net.eagle0.eagle.service
 
-import java.sql.{Connection, Types}
+import java.sql.{Connection, SQLException, Statement, Types}
+import java.util.concurrent.{Executor, Executors, ThreadFactory}
+import java.util.concurrent.atomic.AtomicInteger
 
 import scala.collection.mutable
 
-import net.eagle0.common.JfrEvents
+import net.eagle0.common.{JfrEvents, SimpleTimedLogger}
 import net.eagle0.eagle.{FactionId, GameId, RoundId, ShardokGameId}
 import net.eagle0.eagle.internal.game_state.GameState as GameStateProto
 import net.eagle0.eagle.library.actions.applier.ActionResultWithResultingState
@@ -23,7 +25,10 @@ class PostgresHistory private[service] (
     initialCount: Int,
     initialShardokResultCounts: Map[ShardokGameId, Int] = Map.empty,
     initialShardokGameStates: Map[ShardokGameId, com.google.protobuf.ByteString] = Map.empty,
-    initialShardokLiveBattleCaches: Map[ShardokGameId, ShardokLiveBattleCache] = Map.empty
+    initialShardokLiveBattleCaches: Map[ShardokGameId, ShardokLiveBattleCache] = Map.empty,
+    shardokWriteOverride: Option[PostgresHistory.ShardokWriteBatch => Unit] = None,
+    shardokWriteExecutor: Executor = PostgresHistory.shardokWriteExecutor,
+    maxPendingShardokWrites: Int = PostgresHistory.MaxPendingShardokWrites
 ) extends FullGameHistory {
   import PostgresHistory.{applier, SnapshotInterval}
 
@@ -32,15 +37,20 @@ class PostgresHistory private[service] (
   private val dbLock                  = new Object
   private val actionCache             = mutable.Map.empty[Int, ActionResultT]
   private val stateCache              = mutable.Map.empty[Int, GameState]
-  // Shardok reconnects need these values while assembling a subscription. Keep them synchronized with committed
-  // writes so reconnect setup never waits for the shared Postgres connection pool.
+  // Shardok reconnects need these values while assembling a subscription. Keep them synchronized with accepted
+  // write-behind updates so reconnect setup never waits for the shared Postgres connection pool.
   private val shardokResultCounts     = mutable.Map.from(initialShardokResultCounts)
   private val shardokGameStates       = mutable.Map.from(initialShardokGameStates)
   // Player views and current commands are a live read projection. Cache misses hydrate one battle from PostgreSQL;
-  // committed Shardok updates then keep the projection current until battle resolution evicts it.
+  // accepted Shardok updates then keep the projection current until battle resolution flushes and evicts it.
   private val shardokLiveBattleCaches = mutable.Map.from(initialShardokLiveBattleCaches)
   private var activeConnection        = Option.empty[Connection]
   private var closed                  = false
+  private val shardokWrites           = new OrderedAsyncWriter[PostgresHistory.ShardokWriteBatch](
+    write = shardokWriteOverride.getOrElse(persistShardokWriteWithRetry),
+    executor = shardokWriteExecutor,
+    maxPending = maxPendingShardokWrites
+  )
 
   private def connection: Connection =
     activeConnection.getOrElse(
@@ -70,6 +80,11 @@ class PostgresHistory private[service] (
 
   private def withDbLock[A](f: => A): A = withLock(withConnection(f))
 
+  private def withDbLockAfterShardokWrites[A](f: => A): A = {
+    flushShardokWrites()
+    withDbLock(f)
+  }
+
   override def count: Int = dbLock.synchronized(cachedCount)
 
   override def emittedTutorialDialogues(factionId: FactionId): Map[String, RoundId] = withDbLock {
@@ -113,19 +128,30 @@ class PostgresHistory private[service] (
     ActionResultWithResultingState(readActionResult(lastSeq), stateAfter(cachedCount))
   }
 
-  override def saveNow: PostgresHistory = this
+  override def saveNow: PostgresHistory = {
+    flushShardokWrites()
+    this
+  }
 
-  override def close(): Unit =
+  override def close(): Unit = {
     dbLock.synchronized {
       closed = true
-      shardokLiveBattleCaches.clear()
     }
+    try shardokWrites.close()
+    finally
+      dbLock.synchronized {
+        closed = true
+        shardokLiveBattleCaches.clear()
+      }
+  }
+
+  override def flushShardokWrites(): Unit = shardokWrites.flush()
 
   private[service] def createSqliteSaveExport(
       dbFile: java.io.File,
       startingState: GameState,
       finalState: GameState
-  ): SqliteHistory = withDbLock {
+  ): SqliteHistory = withDbLockAfterShardokWrites {
     val sqlite = SqliteHistory.createForSaveExport(dbFile, startingState)
     sqlite.importSaveExportRowsFromJdbc(connection, finalState)
     sqlite
@@ -133,7 +159,7 @@ class PostgresHistory private[service] (
 
   override def withNewResults(
       newResults: Vector[ActionResultWithResultingState]
-  ): PostgresHistory = withDbLock {
+  ): PostgresHistory = withDbLockAfterShardokWrites {
     if newResults.isEmpty then this
     else
       JfrEvents.historyAppend(
@@ -208,7 +234,7 @@ class PostgresHistory private[service] (
       }
   }
 
-  override def truncateTo(targetActionCount: Int): PostgresHistory = withDbLock {
+  override def truncateTo(targetActionCount: Int): PostgresHistory = withDbLockAfterShardokWrites {
     if targetActionCount < 0 || targetActionCount > cachedCount then
       throw new EagleInternalException(s"truncateTo($targetActionCount) out of range; count is $cachedCount")
     if targetActionCount == cachedCount then this
@@ -401,8 +427,9 @@ class PostgresHistory private[service] (
   ): Option[ShardokAvailableCommands] =
     withLock(shardokLiveBattleCache(shardokGameId).availableCommandsFor(factionId))
 
-  override def evictShardokReadCache(shardokGameId: ShardokGameId): Unit = withLock {
-    shardokLiveBattleCaches.remove(shardokGameId): Unit
+  override def evictShardokReadCache(shardokGameId: ShardokGameId): Unit = {
+    flushShardokWrites()
+    withLock(shardokLiveBattleCaches.remove(shardokGameId): Unit)
   }
 
   override def withNewShardokResults(
@@ -412,152 +439,248 @@ class PostgresHistory private[service] (
       eagleRoundId: RoundId,
       currentGameState: Array[Byte],
       availableCommands: Vector[(FactionId, Option[ShardokAvailableCommands])]
-  ): PostgresHistory = withDbLock {
-    val originalAutoCommit = connection.getAutoCommit
-    connection.setAutoCommit(false)
-    try {
-      val existingResultCount = shardokResultCounts.getOrElse(shardokGameId, 0)
-      val existingLiveCache   = shardokLiveBattleCache(shardokGameId)
-      if newResults.nonEmpty then {
-        val insertStmt = connection.prepareStatement(
-          "INSERT INTO shardok_results (shardok_game_id, action_seq, payload) VALUES (?, ?, ?)"
-        )
-        try {
-          newResults.zipWithIndex.foreach {
-            case (result, i) =>
-              insertStmt.setString(1, shardokGameId)
-              insertStmt.setInt(2, existingResultCount + i)
-              insertStmt.setBytes(3, result.toByteArray)
-              insertStmt.addBatch()
-          }
-          insertStmt.executeBatch(): Unit
-        } finally insertStmt.close()
-      }
-
-      val allFids = (existingLiveCache.registeredFactionIds ++ newPlayerResults.map(_._1)).toVector.sorted
-
-      val insertViewStmt = connection.prepareStatement(
-        "INSERT INTO shardok_player_results (shardok_game_id, faction_id, seq, payload) VALUES (?, ?, ?, ?)"
+  ): PostgresHistory = withLock {
+    val existingResultCount = shardokResultCounts.getOrElse(shardokGameId, 0)
+    val existingLiveCache   = shardokLiveBattleCache(shardokGameId)
+    val allFids             = (existingLiveCache.registeredFactionIds ++ newPlayerResults.map(_._1)).toVector.sorted
+    val playerWrites        = allFids.map { factionId =>
+      PostgresHistory.ShardokPlayerWrite(
+        factionId = factionId,
+        firstSequence = existingLiveCache.exactPlayerResults(factionId).size,
+        results = newPlayerResults.find(_._1 == factionId).fold(Vector.empty)(_._2)
       )
-      val upsertCmdStmt  = connection.prepareStatement(
+    }
+    val commandWrites       = allFids.map { factionId =>
+      factionId -> availableCommands.find(_._1 == factionId).flatMap(_._2)
+    }
+    val batch               = PostgresHistory.ShardokWriteBatch(
+      shardokGameId = shardokGameId,
+      firstResultSequence = existingResultCount,
+      results = newResults,
+      playerWrites = playerWrites,
+      eagleRoundId = eagleRoundId,
+      currentGameState = currentGameState.clone(),
+      commandWrites = commandWrites
+    )
+
+    shardokWrites.enqueue(batch)
+    shardokResultCounts(shardokGameId) = existingResultCount + newResults.size
+    shardokGameStates(shardokGameId) = com.google.protobuf.ByteString.copyFrom(currentGameState)
+    shardokLiveBattleCaches(shardokGameId) = existingLiveCache.appended(newPlayerResults, availableCommands)
+    this
+  }
+
+  private def persistShardokWriteWithRetry(batch: PostgresHistory.ShardokWriteBatch): Unit = {
+    var attempt = 1
+    var written = false
+    while !written do
+      try {
+        persistShardokWrite(batch)
+        written = true
+      } catch {
+        case e: SQLException =>
+          if attempt >= PostgresHistory.ShardokWriteMaxAttempts then {
+            SimpleTimedLogger.printLogger.logLine(
+              s"Asynchronous Shardok persistence failed for ${batch.shardokGameId} after $attempt attempts: $e"
+            )
+            throw e
+          } else {
+            SimpleTimedLogger.printLogger.logLine(
+              s"Retrying asynchronous Shardok persistence for ${batch.shardokGameId} after attempt $attempt: $e"
+            )
+            try Thread.sleep(PostgresHistory.ShardokWriteRetryDelayMillis * attempt)
+            catch {
+              case interrupted: InterruptedException =>
+                Thread.currentThread().interrupt()
+                throw interrupted
+            }
+            attempt += 1
+          }
+      }
+    end while
+  }
+
+  private def persistShardokWrite(batch: PostgresHistory.ShardokWriteBatch): Unit = {
+    val writeConnection = PostgresHistory.openConnection()
+    try {
+      PostgresHistory.useSchema(writeConnection, schemaName)
+      val originalAutoCommit = writeConnection.getAutoCommit
+      writeConnection.setAutoCommit(false)
+      try {
+        persistShardokResults(writeConnection, batch)
+        persistShardokPlayerResults(writeConnection, batch)
+        persistShardokCommands(writeConnection, batch)
+        persistShardokState(writeConnection, batch)
+        writeConnection.commit()
+      } catch {
+        case e: Exception =>
+          try writeConnection.rollback()
+          catch {
+            case rollbackFailure: Exception => e.addSuppressed(rollbackFailure)
+          }
+          throw e
+      } finally writeConnection.setAutoCommit(originalAutoCommit)
+    } finally PostgresHistory.closeOpenedConnection(writeConnection)
+    end try
+  }
+
+  private def persistShardokResults(
+      writeConnection: Connection,
+      batch: PostgresHistory.ShardokWriteBatch
+  ): Unit =
+    if batch.results.nonEmpty then {
+      val stmt = writeConnection.prepareStatement(
+        """INSERT INTO shardok_results (shardok_game_id, action_seq, payload) VALUES (?, ?, ?)
+          |ON CONFLICT (shardok_game_id, action_seq) DO UPDATE SET payload = EXCLUDED.payload
+          |WHERE shardok_results.payload = EXCLUDED.payload""".stripMargin
+      )
+      try {
+        batch.results.zipWithIndex.foreach {
+          case (result, index) =>
+            stmt.setString(1, batch.shardokGameId)
+            stmt.setInt(2, batch.firstResultSequence + index)
+            stmt.setBytes(3, result.toByteArray)
+            stmt.addBatch()
+        }
+        PostgresHistory.requireIdempotentBatch("shardok_results", stmt.executeBatch(), batch.results.size)
+      } finally stmt.close()
+    }
+
+  private def persistShardokPlayerResults(
+      writeConnection: Connection,
+      batch: PostgresHistory.ShardokWriteBatch
+  ): Unit = {
+    val rows = batch.playerWrites.flatMap { playerWrite =>
+      playerWrite.results.zipWithIndex.map { case (result, index) => (playerWrite, index, result) }
+    }
+    if rows.nonEmpty then {
+      val stmt = writeConnection.prepareStatement(
+        """INSERT INTO shardok_player_results (shardok_game_id, faction_id, seq, payload) VALUES (?, ?, ?, ?)
+          |ON CONFLICT (shardok_game_id, faction_id, seq) DO UPDATE SET payload = EXCLUDED.payload
+          |WHERE shardok_player_results.payload = EXCLUDED.payload""".stripMargin
+      )
+      try {
+        rows.foreach {
+          case (playerWrite, index, result) =>
+            stmt.setString(1, batch.shardokGameId)
+            stmt.setInt(2, playerWrite.factionId)
+            stmt.setInt(3, playerWrite.firstSequence + index)
+            stmt.setBytes(4, result.toByteArray)
+            stmt.addBatch()
+        }
+        PostgresHistory.requireIdempotentBatch("shardok_player_results", stmt.executeBatch(), rows.size)
+      } finally stmt.close()
+    }
+  }
+
+  private def persistShardokCommands(
+      writeConnection: Connection,
+      batch: PostgresHistory.ShardokWriteBatch
+  ): Unit =
+    if batch.commandWrites.nonEmpty then {
+      val stmt = writeConnection.prepareStatement(
         """INSERT INTO shardok_player_commands (shardok_game_id, faction_id, payload)
           |VALUES (?, ?, ?)
           |ON CONFLICT (shardok_game_id, faction_id) DO UPDATE SET payload = EXCLUDED.payload""".stripMargin
       )
       try {
-        allFids.foreach { fid =>
-          val existingViewCount = existingLiveCache.exactPlayerResults(fid).size
-          val viewsForFid       = newPlayerResults.find(_._1 == fid).fold(Vector.empty)(_._2)
-          viewsForFid.zipWithIndex.foreach {
-            case (view, i) =>
-              insertViewStmt.setString(1, shardokGameId)
-              insertViewStmt.setInt(2, fid)
-              insertViewStmt.setInt(3, existingViewCount + i)
-              insertViewStmt.setBytes(4, view.toByteArray)
-              insertViewStmt.addBatch()
-          }
-          upsertCmdStmt.setString(1, shardokGameId)
-          upsertCmdStmt.setInt(2, fid)
-          availableCommands.find(_._1 == fid).flatMap(_._2) match {
-            case Some(cmd) => upsertCmdStmt.setBytes(3, cmd.toByteArray)
-            case None      => upsertCmdStmt.setNull(3, Types.BINARY)
-          }
-          upsertCmdStmt.addBatch()
+        batch.commandWrites.foreach {
+          case (factionId, command) =>
+            stmt.setString(1, batch.shardokGameId)
+            stmt.setInt(2, factionId)
+            command match {
+              case Some(value) => stmt.setBytes(3, value.toByteArray)
+              case None        => stmt.setNull(3, Types.BINARY)
+            }
+            stmt.addBatch()
         }
-        insertViewStmt.executeBatch(): Unit
-        upsertCmdStmt.executeBatch(): Unit
-      } finally {
-        insertViewStmt.close()
-        upsertCmdStmt.close()
-      }
+        PostgresHistory.requireSuccessfulBatch("shardok_player_commands", stmt.executeBatch(), batch.commandWrites.size)
+      } finally stmt.close()
+    }
 
-      val stateStmt = connection.prepareStatement(
-        """INSERT INTO shardok_state (shardok_game_id, game_state, last_eagle_round_id)
-          |VALUES (?, ?, ?)
-          |ON CONFLICT (shardok_game_id) DO UPDATE SET game_state = EXCLUDED.game_state""".stripMargin
-      )
-      try {
-        stateStmt.setString(1, shardokGameId)
-        stateStmt.setBytes(2, currentGameState)
-        stateStmt.setInt(3, eagleRoundId)
-        stateStmt.executeUpdate(): Unit
-      } finally stateStmt.close()
-
-      connection.commit()
-      shardokResultCounts(shardokGameId) = existingResultCount + newResults.size
-      shardokGameStates(shardokGameId) = com.google.protobuf.ByteString.copyFrom(currentGameState)
-      shardokLiveBattleCaches(shardokGameId) = existingLiveCache.appended(newPlayerResults, availableCommands)
-    } catch {
-      case e: Exception =>
-        connection.rollback()
-        throw e
-    } finally connection.setAutoCommit(originalAutoCommit)
-    end try
-    this
-  }
-
-  override def withResetShardokResults(shardokGameId: ShardokGameId): PostgresHistory = withDbLock {
-    val originalAutoCommit = connection.getAutoCommit
-    connection.setAutoCommit(false)
-    try {
-      Vector(
-        "DELETE FROM shardok_results WHERE shardok_game_id = ?",
-        "DELETE FROM shardok_player_results WHERE shardok_game_id = ?",
-        "DELETE FROM shardok_player_commands WHERE shardok_game_id = ?",
-        "DELETE FROM shardok_state WHERE shardok_game_id = ?"
-      ).foreach { sql =>
-        val stmt = connection.prepareStatement(sql)
-        try {
-          stmt.setString(1, shardokGameId)
-          stmt.executeUpdate(): Unit
-        } finally stmt.close()
-      }
-      connection.commit()
-      shardokResultCounts.remove(shardokGameId): Unit
-      shardokGameStates.remove(shardokGameId): Unit
-      shardokLiveBattleCaches.remove(shardokGameId): Unit
-    } catch {
-      case e: Exception =>
-        connection.rollback()
-        throw e
-    } finally connection.setAutoCommit(originalAutoCommit)
-    end try
-    this
-  }
-
-  private[service] def importShardokTables(shardokExport: SqliteHistory.ShardokTableExport): Unit = withDbLock {
-    val originalAutoCommit = connection.getAutoCommit
-    connection.setAutoCommit(false)
-    try {
-      PostgresHistory.insertShardokResultRows(connection, shardokExport.results)
-      PostgresHistory.insertShardokStateRows(connection, shardokExport.states)
-      PostgresHistory.insertShardokPlayerResultRows(connection, shardokExport.playerResults)
-      PostgresHistory.insertShardokPlayerCommandRows(connection, shardokExport.playerCommands)
-      connection.commit()
-      shardokExport.results
-        .groupMapReduce(_.shardokGameId)(_ => 1)(_ + _)
-        .foreach { case (shardokGameId, count) => shardokResultCounts(shardokGameId) = count }
-      shardokExport.states.foreach { state =>
-        shardokGameStates(state.shardokGameId) = com.google.protobuf.ByteString.copyFrom(state.gameState)
-      }
-      shardokLiveBattleCaches.clear()
-    } catch {
-      case e: Exception =>
-        connection.rollback()
-        throw e
-    } finally connection.setAutoCommit(originalAutoCommit)
-    end try
-  }
-
-  private[service] def exportShardokTablesForMigration(): SqliteHistory.ShardokTableExport = withDbLock {
-    SqliteHistory.ShardokTableExport(
-      results = PostgresHistory.readShardokResultRows(connection),
-      states = PostgresHistory.readShardokStateRows(connection),
-      playerResults = PostgresHistory.readShardokPlayerResultRows(connection),
-      playerCommands = PostgresHistory.readShardokPlayerCommandRows(connection)
+  private def persistShardokState(
+      writeConnection: Connection,
+      batch: PostgresHistory.ShardokWriteBatch
+  ): Unit = {
+    val stmt = writeConnection.prepareStatement(
+      """INSERT INTO shardok_state (shardok_game_id, game_state, last_eagle_round_id)
+        |VALUES (?, ?, ?)
+        |ON CONFLICT (shardok_game_id) DO UPDATE SET game_state = EXCLUDED.game_state""".stripMargin
     )
+    try {
+      stmt.setString(1, batch.shardokGameId)
+      stmt.setBytes(2, batch.currentGameState)
+      stmt.setInt(3, batch.eagleRoundId)
+      if stmt.executeUpdate() != 1 then
+        throw new EagleInternalException(s"Failed to persist Shardok state for ${batch.shardokGameId}")
+    } finally stmt.close()
   }
 
+  override def withResetShardokResults(shardokGameId: ShardokGameId): PostgresHistory =
+    withDbLockAfterShardokWrites {
+      val originalAutoCommit = connection.getAutoCommit
+      connection.setAutoCommit(false)
+      try {
+        Vector(
+          "DELETE FROM shardok_results WHERE shardok_game_id = ?",
+          "DELETE FROM shardok_player_results WHERE shardok_game_id = ?",
+          "DELETE FROM shardok_player_commands WHERE shardok_game_id = ?",
+          "DELETE FROM shardok_state WHERE shardok_game_id = ?"
+        ).foreach { sql =>
+          val stmt = connection.prepareStatement(sql)
+          try {
+            stmt.setString(1, shardokGameId)
+            stmt.executeUpdate(): Unit
+          } finally stmt.close()
+        }
+        connection.commit()
+        shardokResultCounts.remove(shardokGameId): Unit
+        shardokGameStates.remove(shardokGameId): Unit
+        shardokLiveBattleCaches.remove(shardokGameId): Unit
+      } catch {
+        case e: Exception =>
+          connection.rollback()
+          throw e
+      } finally connection.setAutoCommit(originalAutoCommit)
+      end try
+      this
+    }
+
+  private[service] def importShardokTables(shardokExport: SqliteHistory.ShardokTableExport): Unit =
+    withDbLockAfterShardokWrites {
+      val originalAutoCommit = connection.getAutoCommit
+      connection.setAutoCommit(false)
+      try {
+        PostgresHistory.insertShardokResultRows(connection, shardokExport.results)
+        PostgresHistory.insertShardokStateRows(connection, shardokExport.states)
+        PostgresHistory.insertShardokPlayerResultRows(connection, shardokExport.playerResults)
+        PostgresHistory.insertShardokPlayerCommandRows(connection, shardokExport.playerCommands)
+        connection.commit()
+        shardokExport.results
+          .groupMapReduce(_.shardokGameId)(_ => 1)(_ + _)
+          .foreach { case (shardokGameId, count) => shardokResultCounts(shardokGameId) = count }
+        shardokExport.states.foreach { state =>
+          shardokGameStates(state.shardokGameId) = com.google.protobuf.ByteString.copyFrom(state.gameState)
+        }
+        shardokLiveBattleCaches.clear()
+      } catch {
+        case e: Exception =>
+          connection.rollback()
+          throw e
+      } finally connection.setAutoCommit(originalAutoCommit)
+      end try
+    }
+
+  private[service] def exportShardokTablesForMigration(): SqliteHistory.ShardokTableExport =
+    withDbLockAfterShardokWrites {
+      SqliteHistory.ShardokTableExport(
+        results = PostgresHistory.readShardokResultRows(connection),
+        states = PostgresHistory.readShardokStateRows(connection),
+        playerResults = PostgresHistory.readShardokPlayerResultRows(connection),
+        playerCommands = PostgresHistory.readShardokPlayerCommandRows(connection)
+      )
+    }
+
   private def readLatestSnapshotAtOrBefore(seq: Int): (Int, GameState) = {
     val stmt = connection.prepareStatement(
       "SELECT boundary_action_seq, payload FROM state_snapshots WHERE boundary_action_seq <= ? ORDER BY boundary_action_seq DESC LIMIT 1"
@@ -658,12 +781,52 @@ class PostgresHistory private[service] (
 }
 
 object PostgresHistory {
-  private val SnapshotInterval             = 25
+  private val SnapshotInterval                      = 25
   // Keep v1 during the additive phase so the old blue/green container can still lazy-load expanded schemas.
-  private[service] val SchemaVersion       = 1
+  private[service] val SchemaVersion                = 1
   // Prepared schemas need a new namespace even though the additive layout remains readable by schema-v1 servers.
-  private[service] val SchemaLayoutVersion = 3
-  private val applier                      = SqliteHistory.applier
+  private[service] val SchemaLayoutVersion          = 3
+  private val applier                               = SqliteHistory.applier
+  private[service] val MaxPendingShardokWrites      = 8
+  private[service] val ShardokWriteMaxAttempts      = 3
+  private[service] val ShardokWriteRetryDelayMillis = 50L
+
+  private val shardokWriterThreadCounter              = new AtomicInteger(0)
+  private[service] val shardokWriteExecutor: Executor = Executors.newFixedThreadPool(
+    4,
+    new ThreadFactory {
+      override def newThread(runnable: Runnable): Thread = {
+        val thread = new Thread(runnable, s"postgres-shardok-writer-${shardokWriterThreadCounter.incrementAndGet()}")
+        thread.setDaemon(true)
+        thread
+      }
+    }
+  )
+
+  private[service] case class ShardokPlayerWrite(
+      factionId: FactionId,
+      firstSequence: Int,
+      results: Vector[ShardokActionResultView]
+  )
+
+  private[service] case class ShardokWriteBatch(
+      shardokGameId: ShardokGameId,
+      firstResultSequence: Int,
+      results: Vector[ShardokActionResult],
+      playerWrites: Vector[ShardokPlayerWrite],
+      eagleRoundId: RoundId,
+      currentGameState: Array[Byte],
+      commandWrites: Vector[(FactionId, Option[ShardokAvailableCommands])]
+  )
+
+  private[service] def requireSuccessfulBatch(table: String, results: Array[Int], expected: Int): Unit =
+    if results.length != expected || results.exists(result => result <= 0 && result != Statement.SUCCESS_NO_INFO) then
+      throw new EagleInternalException(
+        s"Failed to persist an ordered $table batch: expected $expected successful rows, got ${results.toVector}"
+      )
+
+  private[service] def requireIdempotentBatch(table: String, results: Array[Int], expected: Int): Unit =
+    requireSuccessfulBatch(table, results, expected)
 
   val ActionResultInsertSql: String =
     """INSERT INTO action_results

src/main/scala/net/eagle0/eagle/service/controller/GameController.scala

@@ -722,6 +722,9 @@ final case class GameController(
   ): GameControllerWithPostResults = {
     val shardokGameId = battleResolution.battle.shardokGameId
 
+    // A completed strategic battle must never become durable before its final tactical updates.
+    fullHistory.flushShardokWrites()
+
     // Record the winner in the battle progress tracker
     val winnerFid = battleResolution.resolvedPlayers
       .find(_.endGameCondition.isVictory)

src/test/scala/net/eagle0/eagle/service/BUILD.bazel

@@ -40,6 +40,7 @@ scala_test(
     deps = [
         "//src/main/protobuf/net/eagle0/shardok/api:action_result_view_scala_proto",
         "//src/main/protobuf/net/eagle0/shardok/api:command_descriptor_scala_proto",
+        "//src/main/protobuf/net/eagle0/shardok/storage:action_result_scala_proto",
         "//src/main/scala/net/eagle0/eagle:eagle_pkg",
         "//src/main/scala/net/eagle0/eagle/service:changed_province_table_writer",
         "//src/main/scala/net/eagle0/eagle/service:sqlite_history",

src/test/scala/net/eagle0/eagle/service/PostgresHistorySchemaBatchTest.scala

@@ -1,10 +1,15 @@
 package net.eagle0.eagle.service
 
 import java.sql.{Connection, PreparedStatement, ResultSet, Statement}
+import java.util.concurrent.{ConcurrentLinkedQueue, CountDownLatch, Executors, TimeUnit}
+import java.util.concurrent.atomic.AtomicReference
+
+import scala.jdk.CollectionConverters.*
 
 import com.google.protobuf.ByteString
 import net.eagle0.shardok.api.action_result_view.ActionResultView as ShardokActionResultView
 import net.eagle0.shardok.api.command_descriptor.AvailableCommands as ShardokAvailableCommands
+import net.eagle0.shardok.storage.action_result.ActionResult as ShardokActionResult
 import org.scalamock.scalatest.MockFactory
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
@@ -296,6 +301,161 @@ class PostgresHistorySchemaBatchTest extends AnyFlatSpec with Matchers with Mock
     history.shardokPlayerAvailableCommands("battle", 1) shouldBe Some(availableCommand)
   }
 
+  "PostgresHistory Shardok write-behind" should "publish accepted updates before a blocked database write completes" in {
+    val writeStarted     = new CountDownLatch(1)
+    val allowWrite       = new CountDownLatch(1)
+    val acceptedBatch    = new AtomicReference[PostgresHistory.ShardokWriteBatch]()
+    val executor         = Executors.newSingleThreadExecutor()
+    val playerResult     = ShardokActionResultView()
+    val availableCommand = ShardokAvailableCommands()
+    val history          = new PostgresHistory(
+      gameId = 0x1234L,
+      initialCount = 1,
+      initialShardokLiveBattleCaches = Map("battle" -> ShardokLiveBattleCache(Map.empty, Map.empty)),
+      shardokWriteOverride = Some { batch =>
+        acceptedBatch.set(batch)
+        writeStarted.countDown()
+        allowWrite.await()
+      },
+      shardokWriteExecutor = executor
+    )
+
+    try {
+      history.withNewShardokResults(
+        shardokGameId = "battle",
+        newResults = Vector(ShardokActionResult()),
+        newPlayerResults = Vector(1 -> Vector(playerResult)),
+        eagleRoundId = 7,
+        currentGameState = Array[Byte](1, 2, 3),
+        availableCommands = Vector(1 -> Some(availableCommand))
+      ): Unit
+
+      writeStarted.await(5, TimeUnit.SECONDS) shouldBe true
+      history.shardokCount("battle") shouldBe 1
+      history.shardokGameState("battle") shouldBe ByteString.copyFrom(Array[Byte](1, 2, 3))
+      history.shardokPlayerResultsSince("battle", 1, 0) shouldBe Vector(playerResult)
+      history.shardokPlayerAvailableCommands("battle", 1) shouldBe Some(availableCommand)
+      acceptedBatch.get().firstResultSequence shouldBe 0
+
+      allowWrite.countDown()
+      history.flushShardokWrites()
+    } finally {
+      allowWrite.countDown()
+      history.close()
+      executor.shutdownNow(): Unit
+    }
+  }
+
+  it should "persist accepted updates in sequence order" in {
+    val sequences = new ConcurrentLinkedQueue[Int]()
+    val executor  = Executors.newFixedThreadPool(2)
+    val history   = new PostgresHistory(
+      gameId = 0x1234L,
+      initialCount = 1,
+      initialShardokLiveBattleCaches = Map("battle" -> ShardokLiveBattleCache(Map.empty, Map.empty)),
+      shardokWriteOverride = Some(batch => sequences.add(batch.firstResultSequence): Unit),
+      shardokWriteExecutor = executor
+    )
+
+    try {
+      Vector(1, 2).foreach { value =>
+        history.withNewShardokResults(
+          shardokGameId = "battle",
+          newResults = Vector(ShardokActionResult()),
+          newPlayerResults = Vector.empty,
+          eagleRoundId = 7,
+          currentGameState = Array(value.toByte),
+          availableCommands = Vector.empty
+        )
+      }
+
+      history.flushShardokWrites()
+      sequences.asScala.toVector shouldBe Vector(0, 1)
+      history.shardokCount("battle") shouldBe 2
+    } finally {
+      history.close()
+      executor.shutdownNow(): Unit
+    }
+  }
+
+  it should "surface a failed write at flush and reject later updates" in {
+    val executor = Executors.newSingleThreadExecutor()
+    val history  = new PostgresHistory(
+      gameId = 0x1234L,
+      initialCount = 1,
+      initialShardokLiveBattleCaches = Map("battle" -> ShardokLiveBattleCache(Map.empty, Map.empty)),
+      shardokWriteOverride = Some(_ => throw new RuntimeException("write failed")),
+      shardokWriteExecutor = executor
+    )
+
+    try {
+      history.withNewShardokResults(
+        shardokGameId = "battle",
+        newResults = Vector(ShardokActionResult()),
+        newPlayerResults = Vector.empty,
+        eagleRoundId = 7,
+        currentGameState = Array[Byte](1),
+        availableCommands = Vector.empty
+      ): Unit
+
+      val flushFailure   = intercept[RuntimeException](history.flushShardokWrites())
+      flushFailure.getMessage shouldBe "write failed"
+      val enqueueFailure = intercept[RuntimeException] {
+        history.withNewShardokResults(
+          shardokGameId = "battle",
+          newResults = Vector(ShardokActionResult()),
+          newPlayerResults = Vector.empty,
+          eagleRoundId = 7,
+          currentGameState = Array[Byte](2),
+          availableCommands = Vector.empty
+        )
+      }
+      enqueueFailure.getMessage shouldBe "write failed"
+      val closeFailure   = intercept[RuntimeException](history.close())
+      closeFailure.getMessage shouldBe "write failed"
+    } finally executor.shutdownNow(): Unit
+    end try
+  }
+
+  "OrderedAsyncWriter" should "apply backpressure when its bounded queue is full" in {
+    val firstWriteStarted = new CountDownLatch(1)
+    val allowFirstWrite   = new CountDownLatch(1)
+    val secondEnqueue     = new CountDownLatch(1)
+    val secondReturned    = new CountDownLatch(1)
+    val executor          = Executors.newSingleThreadExecutor()
+    val writer            = new OrderedAsyncWriter[Int](
+      write = { value =>
+        if value == 1 then {
+          firstWriteStarted.countDown()
+          allowFirstWrite.await()
+        }
+      },
+      executor = executor,
+      maxPending = 1
+    )
+
+    try {
+      writer.enqueue(1)
+      firstWriteStarted.await(5, TimeUnit.SECONDS) shouldBe true
+      val producer = new Thread(() => {
+        secondEnqueue.countDown()
+        writer.enqueue(2)
+        secondReturned.countDown()
+      })
+      producer.start()
+
+      secondEnqueue.await(5, TimeUnit.SECONDS) shouldBe true
+      secondReturned.await(100, TimeUnit.MILLISECONDS) shouldBe false
+      allowFirstWrite.countDown()
+      secondReturned.await(5, TimeUnit.SECONDS) shouldBe true
+      writer.flush()
+    } finally {
+      allowFirstWrite.countDown()
+      writer.close()
+      executor.shutdownNow(): Unit
+    }
+  }
+
   "readShardokSubscriptionState" should "load counts and states in one query" in {
     val connection = mock[Connection]
     val statement  = mock[Statement]

src/test/scala/net/eagle0/eagle/service/controller/GameControllerTest.scala

@@ -25,6 +25,7 @@ import net.eagle0.eagle.model.state.province.concrete.ProvinceC
 import net.eagle0.eagle.model.state.run_status.RunStatus
 import net.eagle0.eagle.model.state.shardok_battle.{BattleType, ShardokBattle}
 import net.eagle0.eagle.service.{FullGameHistory, InMemoryHistory, PostResults, SyncResponseObserver}
+import net.eagle0.eagle.shardok_interface.BattleResolution
 import net.eagle0.eagle.UserId
 import org.scalamock.scalatest.MockFactory
 import org.scalatest.flatspec.AnyFlatSpec
@@ -116,6 +117,36 @@ class GameControllerTest extends AnyFlatSpec with MockFactory with Matchers with
     ) shouldBe Some(ServerGameStatus(status = ServerGameStatus.Status.BATTLE_IN_PROGRESS))
   }
 
+  "postBattleResolution" should "flush tactical writes before resolving the strategic battle" in {
+    val engine     = mock[Engine]
+    val history    = mock[FullGameHistory]
+    val battle     = ShardokBattle(
+      hexMapName = "test-map",
+      players = Vector.empty,
+      roundStarted = 0,
+      defenderProvince = 1,
+      eagleGameId = eagleGameId,
+      shardokGameId = "test-battle",
+      battleIndex = 0,
+      battleType = BattleType.AssaultProvince
+    )
+    val resolution = BattleResolution(battle = battle, resolvedPlayers = Vector.empty)
+    val controller = GameController(
+      engine = engine,
+      fullHistory = history,
+      userIdToFactionId = Map.empty,
+      humanClients = Vector.empty,
+      aiClients = Vector.empty,
+      clientTextStore = mock[ClientTextStore]
+    )
+
+    (() => history.flushShardokWrites()).expects().once(): Unit
+    engine.resolveBattle.expects(resolution).throws(new RuntimeException("resolved after flush")): Unit
+
+    val failure = intercept[RuntimeException](controller.postBattleResolution(resolution))
+    failure.getMessage shouldBe "resolved after flush"
+  }
+
   "postStreamingLlmFailure" should "reset visible clients before retrying the request" in {
     val requestId      = "chronicle"
     val mockObserver   = mock[SyncResponseObserver]