admin/eagle0
1
0
Fork 0
mirror of https://github.com/nolen777/eagle0.git synced 2026-07-28 21:35:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

Defer completed text recovery during stream setup

Browse Source
This commit is contained in:
admin
2026-07-19 12:58:44 -07:00
parent c6b73d24be
commit 5a27526871
4 changed files with 240 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/scala/net/eagle0/eagle/service/GamesManager.scala
+31 -6
View File
@@ -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(
src/main/scala/net/eagle0/eagle/service/controller/GameController.scala
+60 -46
View File
@@ -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))
src/test/scala/net/eagle0/eagle/service/GamesManagerTest.scala
+86
View File
@@ -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]
src/test/scala/net/eagle0/eagle/service/controller/GameControllerTest.scala
+63
View File
@@ -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()