From 5a27526871a8044058ea1c9e995285ca71d4e289 Mon Sep 17 00:00:00 2001 From: Dan Crosby Date: Sun, 19 Jul 2026 12:58:44 -0700 Subject: [PATCH] Defer completed text recovery during stream setup --- .../eagle0/eagle/service/GamesManager.scala | 37 +++++- .../service/controller/GameController.scala | 106 ++++++++++-------- .../eagle/service/GamesManagerTest.scala | 86 ++++++++++++++ .../controller/GameControllerTest.scala | 63 +++++++++++ 4 files changed, 240 insertions(+), 52 deletions(-) diff --git a/src/main/scala/net/eagle0/eagle/service/GamesManager.scala b/src/main/scala/net/eagle0/eagle/service/GamesManager.scala index f91cdd476b..106f46b14b 100644 --- a/src/main/scala/net/eagle0/eagle/service/GamesManager.scala +++ b/src/main/scala/net/eagle0/eagle/service/GamesManager.scala @@ -2117,7 +2117,7 @@ class GamesManager( s"[TIMING] ensureGameLoaded for $userId took ${loadEnd - loadStart}ms" ) - this.synchronized { + val controllerForCompleteTextRecovery = this.synchronized { val filterStart = System.currentTimeMillis() updateController( gameId, @@ -2136,12 +2136,37 @@ class GamesManager( s"[TIMING] streamUpdates filtering for $userId took ${filterEnd - filterStart}ms" ) - val totalTime = System.currentTimeMillis() - streamStart - if totalTime > 500 then - SimpleTimedLogger.printLogger.logLine( - s"[TIMING] Total streamUpdates for $userId took ${totalTime}ms (load: ${loadEnd - loadStart}ms, filter: ${filterEnd - filterStart}ms)" - ) + requiredController(gameId) } + + val completeTextStart = System.currentTimeMillis() + val completeTexts = controllerForCompleteTextRecovery.completeTextsAccessibleTo( + userId = userId, + streamingTextStatuses = streamingTextStatuses + ) + val completeTextEnd = System.currentTimeMillis() + if completeTextEnd - completeTextStart > 500 then + SimpleTimedLogger.printLogger.logLine( + s"[TIMING] completed-text recovery for $userId took ${completeTextEnd - completeTextStart}ms" + ) + + this.synchronized { + updateController( + gameId, + _.withCompleteTextsForStream( + clientId = clientId, + responseObserver = responseObserver, + completeTexts = completeTexts + ) + ) + } + + val totalTime = System.currentTimeMillis() - streamStart + if totalTime > 500 then + SimpleTimedLogger.printLogger.logLine( + s"[TIMING] Total streamUpdates for $userId took ${totalTime}ms " + + s"(load: ${loadEnd - loadStart}ms, completed-text recovery: ${completeTextEnd - completeTextStart}ms)" + ) } def stopStreaming( diff --git a/src/main/scala/net/eagle0/eagle/service/controller/GameController.scala b/src/main/scala/net/eagle0/eagle/service/controller/GameController.scala index f256185f48..3b10acddca 100644 --- a/src/main/scala/net/eagle0/eagle/service/controller/GameController.scala +++ b/src/main/scala/net/eagle0/eagle/service/controller/GameController.scala @@ -1231,54 +1231,42 @@ final case class GameController( 0 } else knownResultCount - val knownCompleteTextByteCounts = streamingTextStatuses - .filter(_.knownComplete) - .map(status => status.llmIdentifier -> status.knownByteCount) - .toMap - this.withHumanClient( - clientTextStore - .getCompleteTextsAccessibleTo(fid, knownCompleteTextByteCounts) - .foldLeft( - HumanPlayerClientConnectionState( - eagleGameId = gameId, - factionId = fid, - clientId = clientId, - grpcObserver = responseObserver, - unfilteredKnownHistoryCount = safeKnownResultCount, - knownShardokResultCounts = knownShardokResultCounts, - history = fullHistory, - availableCommands = GameController.toProtoAvailableCommands( - engine.getAvailablePlayerCommands(fid), - engine, - fid - ), - serverGameStatus = GameController.computeServerGameStatus( - engine = engine, - factionId = fid, - otherFactionIds = engine.factionIds.filterNot(_ == fid), - clientTextStore = clientTextStore - ), - activeShardokGameIds = GameStateViewFilter.activeObservableBattleIds(engine.currentState, fid), - shardokServerStatusFor = sgid => - GameController.computeShardokServerStatus( - shardokGameId = sgid, - myFactionId = fid, - fullHistory = fullHistory, - allFactionIds = engine.factionIds, - aiFactionIds = aiClientFactionIds.toSet - ), - streamingTextStatuses = streamingTextStatuses, - logWriter = { - val path = s"/tmp/eagle_log/$gameId/${fid}_bytes.txt" - Paths.get(path).getParent.toFile.mkdirs() - new PrintWriter(new FileOutputStream(new File(path), true)) - } - ) - ) { - case (client, clientText) => - client.update(clientText = clientText) + HumanPlayerClientConnectionState( + eagleGameId = gameId, + factionId = fid, + clientId = clientId, + grpcObserver = responseObserver, + unfilteredKnownHistoryCount = safeKnownResultCount, + knownShardokResultCounts = knownShardokResultCounts, + history = fullHistory, + availableCommands = GameController.toProtoAvailableCommands( + engine.getAvailablePlayerCommands(fid), + engine, + fid + ), + serverGameStatus = GameController.computeServerGameStatus( + engine = engine, + factionId = fid, + otherFactionIds = engine.factionIds.filterNot(_ == fid), + clientTextStore = clientTextStore + ), + activeShardokGameIds = GameStateViewFilter.activeObservableBattleIds(engine.currentState, fid), + shardokServerStatusFor = sgid => + GameController.computeShardokServerStatus( + shardokGameId = sgid, + myFactionId = fid, + fullHistory = fullHistory, + allFactionIds = engine.factionIds, + aiFactionIds = aiClientFactionIds.toSet + ), + streamingTextStatuses = streamingTextStatuses, + logWriter = { + val path = s"/tmp/eagle_log/$gameId/${fid}_bytes.txt" + Paths.get(path).getParent.toFile.mkdirs() + new PrintWriter(new FileOutputStream(new File(path), true)) } + ) ) case None => SimpleTimedLogger.printLogger.logLine( @@ -1292,6 +1280,32 @@ final case class GameController( this } + private[service] def completeTextsAccessibleTo( + userId: UserId, + streamingTextStatuses: Vector[StreamingTextStatus] + ): Vector[CompleteClientText] = + userIdToFactionId.get(userId).fold(Vector.empty) { factionId => + val knownCompleteTextByteCounts = streamingTextStatuses + .filter(_.knownComplete) + .map(status => status.llmIdentifier -> status.knownByteCount) + .toMap + clientTextStore.getCompleteTextsAccessibleTo(factionId, knownCompleteTextByteCounts) + } + + private[service] def withCompleteTextsForStream( + clientId: String, + responseObserver: SyncResponseObserver, + completeTexts: Vector[CompleteClientText] + ): GameController = + copy(humanClients = humanClients.map { client => + if client.clientId == clientId && (client.grpcObserver eq responseObserver) then + completeTexts.foldLeft(client) { + case (updatedClient, clientText) => + updatedClient.update(clientText = clientText) + } + else client + }) + def stopStreaming(userId: UserId): GameController = copy( humanClients = this.humanClients.filterNot(hc => userIdToFactionId.get(userId).contains(hc.factionId)) diff --git a/src/test/scala/net/eagle0/eagle/service/GamesManagerTest.scala b/src/test/scala/net/eagle0/eagle/service/GamesManagerTest.scala index ba986a920f..12b6a27cea 100644 --- a/src/test/scala/net/eagle0/eagle/service/GamesManagerTest.scala +++ b/src/test/scala/net/eagle0/eagle/service/GamesManagerTest.scala @@ -557,6 +557,92 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B mergedController.humanClients.map(_.clientId).shouldBe(Vector("live-client")) } + "streamUpdates" should "send the initial update and release the manager lock before recovering completed texts" in { + val userId = UserId("alice") + val factionId = 7 + val engine = mock[Engine] + val textStore = mock[ClientTextStore] + val observer = mock[SyncResponseObserver] + val state = emptyGameState(Vector.empty) + val history = mock[FullGameHistory] + val initialUpdateSent = new CountDownLatch(1) + val recoveryStarted = new CountDownLatch(1) + val allowRecovery = new CountDownLatch(1) + val updateBeforeLookup = new AtomicBoolean(false) + + (() => engine.gameId).expects().returning(gameId.toLong).anyNumberOfTimes(): Unit + (() => engine.factionIds).expects().returning(Vector(factionId)).anyNumberOfTimes(): Unit + (() => engine.currentState).expects().returning(state).anyNumberOfTimes(): Unit + engine.getAvailablePlayerCommands.expects(factionId).returning(SortedMap.empty).anyNumberOfTimes(): Unit + (() => history.count).expects().returning(0).anyNumberOfTimes(): Unit + history.stateAfter.expects(0).returning(state).anyNumberOfTimes(): Unit + history.sinceFromState.expects(0, state).returning(Vector.empty).anyNumberOfTimes(): Unit + textStore.hasIncompleteTextsAccessibleTo.expects(factionId).returning(false).anyNumberOfTimes(): Unit + observer.onNext + .expects(*) + .onCall { _ => + updateBeforeLookup.set(true) + initialUpdateSent.countDown() + } + .anyNumberOfTimes(): Unit + (textStore + .getCompleteTextsAccessibleTo(_: Int, _: Map[String, Int])) + .expects(factionId, Map.empty) + .onCall { (_, _) => + recoveryStarted.countDown() + if !allowRecovery.await(5, TimeUnit.SECONDS) then + throw new RuntimeException("timed out waiting to finish completed-text recovery") + Vector.empty + }: Unit + + val controller = GameController( + engine = engine, + fullHistory = history, + userIdToFactionId = Map(userId -> factionId), + humanClients = Vector.empty, + aiClients = Vector.empty, + clientTextStore = textStore + ) + val manager = new GamesManager( + pregeneratedClientText = pregeneratedClientText, + shardokInternalInterface = mockShardokInterface, + initialGameControllerInfos = Map( + gameId.toLong -> ControllerInfo(controller, isAiGame = false, maxPlayers = 1) + ), + initialGamesAwaitingPlayers = Vector.empty, + randomGenerator = mockRandom, + persister = new RecordingPersister, + gameCreation = mockGameCreation, + gamePersisterCreation = mockGamePersisterCreation, + hexMaps = Map.empty, + userService = new NoOpUserService, + heroNameCache = new HeroNameCache(targetPoolSize = 0), + currentSchemaVersion = 3 + ) + + val streaming = Future { + manager.streamUpdates( + userId = userId, + gameId = gameId.toLong, + clientId = "client", + knownResultCount = 0, + knownShardokResultCounts = Vector.empty, + streamingTextStatuses = Vector.empty, + responseObserver = observer + ) + } + + try { + recoveryStarted.await(2, TimeUnit.SECONDS) shouldBe true + initialUpdateSent.await(0, TimeUnit.SECONDS) shouldBe true + updateBeforeLookup.get() shouldBe true + + Await.result(Future(manager.stopStreaming(userId, gameId.toLong)), 1.second) + } finally allowRecovery.countDown() + + Await.result(streaming, 5.seconds) + } + it should "not immediately retry unrequested text after an LLM failure" in { val testGameId = gameId.toLong val initialTextStore = mock[ClientTextStore] diff --git a/src/test/scala/net/eagle0/eagle/service/controller/GameControllerTest.scala b/src/test/scala/net/eagle0/eagle/service/controller/GameControllerTest.scala index 5663c099ec..326ca3dfe2 100644 --- a/src/test/scala/net/eagle0/eagle/service/controller/GameControllerTest.scala +++ b/src/test/scala/net/eagle0/eagle/service/controller/GameControllerTest.scala @@ -245,6 +245,69 @@ class GameControllerTest extends AnyFlatSpec with MockFactory with Matchers with controller.withResolvedTextsPushedToClients(texts).humanClients shouldBe empty } + "completeTextsAccessibleTo" should "forward only completed client byte counts" in { + val userId = UserId("joe") + val completed = CompleteClientText("completed", "Recovered text", 7) + val textStatuses = Vector( + StreamingTextStatus(llmIdentifier = "completed", knownByteCount = 14, knownComplete = true), + StreamingTextStatus(llmIdentifier = "in-progress", knownByteCount = 9, knownComplete = false) + ) + val controller = GameController( + engine = mockEngine, + fullHistory = mockFullHistory, + userIdToFactionId = Map(userId -> 7), + humanClients = Vector.empty, + aiClients = Vector.empty, + clientTextStore = mockClientTextStore + ) + (mockClientTextStore + .getCompleteTextsAccessibleTo(_: Int, _: Map[String, Int])) + .expects(7, Map("completed" -> 14)) + .returning(Vector(completed)): Unit + + controller.completeTextsAccessibleTo(userId, textStatuses) shouldBe Vector(completed) + } + + "withCompleteTextsForStream" should "ignore recovery results for a replaced stream" in { + val clientId = "test-client" + val staleObserver = mock[SyncResponseObserver] + val currentObserver = mock[SyncResponseObserver] + val client = new HumanPlayerClientConnectionState( + eagleGameId = eagleGameId, + factionId = 7, + clientId = clientId, + grpcObserver = currentObserver, + unfilteredKnownHistoryCount = 0, + knownShardokResultCounts = Vector.empty, + streamingTextStatusMap = Map.empty, + logWriter = mockLogWriter + ) + val controller = GameController( + engine = mockEngine, + fullHistory = mockFullHistory, + userIdToFactionId = Map(UserId("joe") -> 7), + humanClients = Vector(client), + aiClients = Vector.empty, + clientTextStore = mockClientTextStore + ) + val completed = CompleteClientText("completed", "Recovered text", 7) + + val afterStaleRecovery = controller.withCompleteTextsForStream( + clientId = clientId, + responseObserver = staleObserver, + completeTexts = Vector(completed) + ) + afterStaleRecovery.humanClients.head.streamingTextStatusMap shouldBe empty + + currentObserver.onNext.expects(*): Unit + val afterCurrentRecovery = afterStaleRecovery.withCompleteTextsForStream( + clientId = clientId, + responseObserver = currentObserver, + completeTexts = Vector(completed) + ) + afterCurrentRecovery.humanClients.head.streamingTextStatusMap("completed").knownComplete shouldBe true + } + "gameId" should "return the gameId of the engine" in { (() => mockEngine.gameId) .expects()