Make text client subscriptions resilient (#8668)

2026-07-29 00:55:43 +00:00 · 2026-07-17 09:08:00 -07:00
parent 90bedd4be5
commit eca0289a43
2 changed files with 314 additions and 44 deletions
@@ -2,8 +2,8 @@ package net.eagle0.eagle.text_client

 import java.nio.charset.StandardCharsets
 import java.nio.file.{Files, Path, Paths, StandardCopyOption, StandardOpenOption}
-import java.util.concurrent.atomic.AtomicBoolean
-import java.util.concurrent.TimeUnit
+import java.util.concurrent.{Executors, TimeUnit}
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong}
 import java.util.UUID

 import scala.io.StdIn
@@ -55,6 +55,9 @@ import scalapb.{GeneratedMessage, GeneratedMessageCompanion}
 import scalapb.json4s.{JsonFormat, Printer}

 object EagleTextClient {
+  private val PostAcknowledgementTimeoutSeconds      = 30L
+  private val SubscriptionTimeoutSeconds             = 15L
+  private val ReconnectDelayMillis                   = 1000L
  private val AuthorizationKey: Metadata.Key[String] =
    Metadata.Key.of("Authorization", Metadata.ASCII_STRING_MARSHALLER)
  private val WarmupUserKey: Metadata.Key[String]    =
@@ -193,6 +196,10 @@ private[text_client] object TextClientCommandParser {
 }

 final class EagleTextClient(config: Config) {
+  private final case class ActiveStream(generation: Long, observer: StreamObserver[UpdateStreamRequest])
+  private final case class PendingPost(id: Long, generation: Long, gameId: Long, description: String)
+  private final case class PendingSubscription(id: Long, generation: Long, gameId: Long)
+
  private val channel: ManagedChannel = {
    val builder = ManagedChannelBuilder.forAddress(config.host, config.port)
    if config.useTls then builder.useTransportSecurity().build()
@@ -211,24 +218,19 @@ final class EagleTextClient(config: Config) {
    baseStub.withInterceptors(MetadataUtils.newAttachHeadersInterceptor(metadata))
  }

-  private val state        = new ClientState(TutorialDialogueHistory.default)
-  private val running      = new AtomicBoolean(true)
-  private val reconnecting = new AtomicBoolean(false)
-  private var requestObserver: Option[StreamObserver[UpdateStreamRequest]] = None
-  private val responseObserver: StreamObserver[UpdateStreamResponse]       = new StreamObserver[UpdateStreamResponse] {
-    override def onNext(response: UpdateStreamResponse): Unit =
-      state.handle(response)
-
-    override def onError(t: Throwable): Unit = {
-      println(s"stream error: ${t.getMessage}")
-      reconnect()
-    }
-
-    override def onCompleted(): Unit = {
-      println("stream completed")
-      reconnect()
-    }
+  private val state                                 = new ClientState(TutorialDialogueHistory.default)
+  private val running                               = new AtomicBoolean(true)
+  private val reconnecting                          = new AtomicBoolean(false)
+  private val nextStreamGeneration                  = new AtomicLong(0L)
+  private val nextRequestId                         = new AtomicLong(0L)
+  private val watchdog                              = Executors.newSingleThreadScheduledExecutor { runnable =>
+    val thread = new Thread(runnable, "eagle-text-client-watchdog")
+    thread.setDaemon(true)
+    thread
  }
+  private var requestObserver: Option[ActiveStream] = None
+  private var pendingPost: Option[PendingPost]      = None
+  private var pendingSubscription: Option[PendingSubscription] = None

  openRequestStream()

@@ -240,9 +242,16 @@ final class EagleTextClient(config: Config) {
      println("Type 'help' for commands.")
      repl()
    } finally {
-      requestObserver.foreach(observer => Try(observer.onCompleted()))
+      running.set(false)
+      val activeObserver = synchronized {
+        val observer = requestObserver.map(_.observer)
+        requestObserver = None
+        observer
+      }
+      activeObserver.foreach(observer => Try(observer.onCompleted()))
+      watchdog.shutdownNow()
      channel.shutdown()
-      val _ = channel.awaitTermination(2, TimeUnit.SECONDS)
+      val _              = channel.awaitTermination(2, TimeUnit.SECONDS)
    }

  private def repl(): Unit =
@@ -407,7 +416,7 @@ final class EagleTextClient(config: Config) {

  private def stream(gameId: Long): Unit = {
    state.setCurrentGame(gameId)
-    send(
+    sendSubscription(
      UpdateStreamRequest(
        UpdateStreamRequest.RequestDetails.StreamGameRequest(
          UpdateStreamRequest.StreamGameRequest(
@@ -417,7 +426,8 @@ final class EagleTextClient(config: Config) {
            clientId = UUID.randomUUID().toString
          )
        )
-      )
+      ),
+      gameId
    )
  }

@@ -465,7 +475,7 @@ final class EagleTextClient(config: Config) {
              println("Repeat with post-json --force to submit this march.")
            } else {
              warnings.foreach(warning => println(s"March warning (forced): $warning"))
-              send(
+              sendPost(
                UpdateStreamRequest(
                  UpdateStreamRequest.RequestDetails.PostCommandRequest(
                    PostCommandRequest(
@@ -477,7 +487,9 @@ final class EagleTextClient(config: Config) {
                      )
                    )
                  )
-                )
+                ),
+                gameId = gameId,
+                description = s"Eagle province=$provinceId token=$token"
              )
            }
          case None        => println("No command token yet. Stream a game and wait for available commands.")
@@ -490,7 +502,7 @@ final class EagleTextClient(config: Config) {
      case Some(eagleGameId) =>
        state.shardokCommandPost(shardokGameId, index) match {
          case Some(post) =>
-            send(
+            sendPost(
              UpdateStreamRequest(
                UpdateStreamRequest.RequestDetails.PostCommandRequest(
                  PostCommandRequest(
@@ -504,7 +516,9 @@ final class EagleTextClient(config: Config) {
                    )
                  )
                )
-              )
+              ),
+              gameId = eagleGameId,
+              description = s"Shardok battle=${post.shardokGameId} index=$index token=${post.token}"
            )
          case None       =>
            println("No matching Shardok command. Use shardok-commands first.")
@@ -513,19 +527,83 @@ final class EagleTextClient(config: Config) {
    }

  private def send(request: UpdateStreamRequest): Unit =
-    requestObserver match {
-      case Some(observer) =>
-        Try(observer.onNext(request)) match {
+    synchronized(requestObserver) match {
+      case Some(active) =>
+        Try(active.observer.onNext(request)) match {
          case Success(_)     => ()
-          case Failure(error) =>
-            println(s"send failed: ${error.getMessage}")
-            reconnect()
+          case Failure(error) => handleStreamTermination(active.generation, s"send failed: ${safeMessage(error)}")
        }
-      case None           => println("No active stream. Waiting for reconnection.")
+      case None         => println("No active stream. Waiting for reconnection.")
    }

-  private def openRequestStream(): Unit = synchronized {
-    requestObserver = Some(stub.streamUpdates(responseObserver))
+  private def sendPost(request: UpdateStreamRequest, gameId: Long, description: String): Unit = {
+    val prepared = synchronized {
+      pendingPost match {
+        case Some(existing) => Left(s"Still awaiting acknowledgement for ${existing.description}.")
+        case None if pendingSubscription.nonEmpty =>
+          Left("The current game subscription has not been acknowledged yet.")
+        case None                                 =>
+          requestObserver match {
+            case Some(active) =>
+              val post = PendingPost(nextRequestId.incrementAndGet(), active.generation, gameId, description)
+              pendingPost = Some(post)
+              Right((active, post))
+            case None         => Left("No active stream. Waiting for reconnection.")
+          }
+      }
+    }
+
+    prepared match {
+      case Left(error)           => println(s"post-command: not sent; $error")
+      case Right((active, post)) =>
+        Try(active.observer.onNext(request)) match {
+          case Success(_)     =>
+            if isPendingPost(post.id) then {
+              println(s"post-command: sent ${post.description}; awaiting server acknowledgement")
+              schedulePostTimeout(post.id)
+            }
+          case Failure(error) =>
+            clearPendingPost(post.id)
+            handleStreamTermination(active.generation, s"send failed: ${safeMessage(error)}")
+        }
+    }
+  }
+
+  private def sendSubscription(request: UpdateStreamRequest, gameId: Long): Unit = {
+    val prepared = synchronized {
+      pendingPost match {
+        case Some(post) => Left(s"Still awaiting acknowledgement for ${post.description}.")
+        case None       =>
+          requestObserver match {
+            case Some(active) =>
+              val subscription = PendingSubscription(nextRequestId.incrementAndGet(), active.generation, gameId)
+              pendingSubscription = Some(subscription)
+              Right((active, subscription))
+            case None         => Left("No active stream. Waiting for reconnection.")
+          }
+      }
+    }
+
+    prepared match {
+      case Left(error)                   => println(s"subscription: not sent; $error")
+      case Right((active, subscription)) =>
+        Try(active.observer.onNext(request)) match {
+          case Success(_)     =>
+            if isPendingSubscription(subscription.id) then {
+              println(s"subscription: requested game=$gameId")
+              scheduleSubscriptionTimeout(subscription.id)
+            }
+          case Failure(error) =>
+            clearPendingSubscription(subscription.id)
+            handleStreamTermination(active.generation, s"send failed: ${safeMessage(error)}")
+        }
+    }
+  }
+
+  private def openRequestStream(): Unit = {
+    val generation = nextStreamGeneration.incrementAndGet()
+    val observer   = stub.streamUpdates(responseObserverFor(generation))
+    synchronized { requestObserver = Some(ActiveStream(generation, observer)) }
    requestHexMaps()
  }

@@ -536,13 +614,145 @@ final class EagleTextClient(config: Config) {
      )
    )

-  private def reconnect(): Unit =
+  private def responseObserverFor(generation: Long): StreamObserver[UpdateStreamResponse] =
+    new StreamObserver[UpdateStreamResponse] {
+      override def onNext(response: UpdateStreamResponse): Unit = synchronized {
+        if requestObserver.exists(_.generation == generation) then {
+          response.responseDetails match {
+            case UpdateStreamResponse.ResponseDetails.PostCommandResponse(value) =>
+              acknowledgePost(value.gameId)
+            case UpdateStreamResponse.ResponseDetails.SubscriptionAck(value)     =>
+              acknowledgeSubscription(value.gameId)
+            case _                                                               => ()
+          }
+          state.handle(response)
+        }
+      }
+
+      override def onError(error: Throwable): Unit =
+        handleStreamTermination(generation, s"stream error: ${safeMessage(error)}")
+
+      override def onCompleted(): Unit =
+        handleStreamTermination(generation, "stream completed")
+    }
+
+  private def acknowledgePost(gameId: Long): Unit = synchronized {
+    if pendingPost.exists(_.gameId == gameId) then pendingPost = None
+  }
+
+  private def acknowledgeSubscription(gameId: Long): Unit = synchronized {
+    if pendingSubscription.exists(_.gameId == gameId) then pendingSubscription = None
+  }
+
+  private def clearPendingPost(id: Long): Unit = synchronized {
+    if pendingPost.exists(_.id == id) then pendingPost = None
+  }
+
+  private def clearPendingSubscription(id: Long): Unit = synchronized {
+    if pendingSubscription.exists(_.id == id) then pendingSubscription = None
+  }
+
+  private def isPendingPost(id: Long): Boolean = synchronized {
+    pendingPost.exists(_.id == id)
+  }
+
+  private def isPendingSubscription(id: Long): Boolean = synchronized {
+    pendingSubscription.exists(_.id == id)
+  }
+
+  private def schedulePostTimeout(id: Long): Unit = {
+    val _ = watchdog.schedule(
+      new Runnable {
+        override def run(): Unit = handlePostTimeout(id)
+      },
+      EagleTextClient.PostAcknowledgementTimeoutSeconds,
+      TimeUnit.SECONDS
+    )
+  }
+
+  private def scheduleSubscriptionTimeout(id: Long): Unit = {
+    val _ = watchdog.schedule(
+      new Runnable {
+        override def run(): Unit = handleSubscriptionTimeout(id)
+      },
+      EagleTextClient.SubscriptionTimeoutSeconds,
+      TimeUnit.SECONDS
+    )
+  }
+
+  private def handlePostTimeout(id: Long): Unit = {
+    val timedOut = synchronized {
+      pendingPost.filter(_.id == id).map { post =>
+        pendingPost = None
+        post
+      }
+    }
+    timedOut.foreach { post =>
+      println(
+        s"post-command: no acknowledgement for ${post.description}; delivery is uncertain. " +
+          "Reconnecting without resending. Inspect the refreshed state before retrying."
+      )
+      restartGeneration(post.generation, "post acknowledgement timed out")
+    }
+  }
+
+  private def handleSubscriptionTimeout(id: Long): Unit = {
+    val timedOut = synchronized {
+      pendingSubscription.filter(_.id == id).map { subscription =>
+        pendingSubscription = None
+        subscription
+      }
+    }
+    timedOut.foreach { subscription =>
+      println(s"subscription: no acknowledgement for game=${subscription.gameId}; reconnecting")
+      restartGeneration(subscription.generation, "subscription acknowledgement timed out")
+    }
+  }
+
+  private def restartGeneration(generation: Long, reason: String): Unit = {
+    val observer = synchronized {
+      requestObserver.filter(_.generation == generation).map { active =>
+        requestObserver = None
+        pendingSubscription = pendingSubscription.filterNot(_.generation == generation)
+        active.observer
+      }
+    }
+    observer.foreach { activeObserver =>
+      println(s"stream reconnect: $reason")
+      val _ = Try(activeObserver.onCompleted())
+      scheduleReconnect()
+    }
+  }
+
+  private def handleStreamTermination(generation: Long, message: String): Unit = {
+    val (wasActive, uncertainPost) = synchronized {
+      if requestObserver.exists(_.generation == generation) then {
+        requestObserver = None
+        val post = pendingPost.filter(_.generation == generation)
+        pendingPost = pendingPost.filterNot(_.generation == generation)
+        pendingSubscription = pendingSubscription.filterNot(_.generation == generation)
+        (true, post)
+      } else (false, None)
+    }
+
+    if wasActive then {
+      println(message)
+      uncertainPost.foreach { post =>
+        println(
+          s"post-command: connection ended while awaiting ${post.description}; delivery is uncertain. " +
+            "Inspect the refreshed state before retrying."
+        )
+      }
+      scheduleReconnect()
+    }
+  }
+
+  private def scheduleReconnect(): Unit =
    if running.get() && reconnecting.compareAndSet(false, true) then {
-      synchronized { requestObserver = None }
      val reconnectThread = new Thread(
        () =>
          try {
-            Thread.sleep(1000)
+            Thread.sleep(EagleTextClient.ReconnectDelayMillis)
            openRequestStream()
            state.currentGameId.foreach { gameId =>
              println(s"Reconnecting to game $gameId with saved text state.")
@@ -551,13 +761,19 @@ final class EagleTextClient(config: Config) {
          } catch {
            case _: InterruptedException => Thread.currentThread.interrupt()
            case error: Throwable        => println(s"reconnect failed: ${error.getMessage}")
-          } finally reconnecting.set(false),
+          } finally {
+            reconnecting.set(false)
+            if running.get() && synchronized(requestObserver.isEmpty) then scheduleReconnect()
+          },
        "eagle-text-client-reconnect"
      )
      reconnectThread.setDaemon(true)
      reconnectThread.start()
    }

+  private def safeMessage(error: Throwable): String =
+    Option(error.getMessage).filter(_.nonEmpty).getOrElse(error.getClass.getSimpleName)
+
  private def parseSelected(commandType: String, json: String): Try[SelectedCommandProto] =
    commandType match {
      case "AlmsSelectedCommand"                    => parseJson[sc.AlmsSelectedCommand](json)
@@ -731,9 +947,9 @@ final class ClientState(
        println(s"join-game: ${value.result} game=${value.gameId}")
      case UpdateStreamResponse.ResponseDetails.SubscriptionAck(value)       =>
        setCurrentGame(value.gameId)
-        latestUnfilteredResultCount = value.confirmedResultCount
+        latestUnfilteredResultCount = latestUnfilteredResultCount.max(value.confirmedResultCount)
        println(
-          s"subscription: success=${value.success} game=${value.gameId} count=${value.confirmedResultCount} ${value.errorMessage}"
+          s"subscription: success=${value.success} game=${value.gameId} confirmedHistoryCount=${value.confirmedResultCount} ${value.errorMessage}"
        )
      case UpdateStreamResponse.ResponseDetails.PostCommandResponse(value)   =>
        println(s"post-command: ${value.status} game=${value.gameId} ${value.errorMessage}")
@@ -771,7 +987,7 @@ final class ClientState(
            )
            printReadyTutorialDialogues()
            println(
-              s"game-update: ${ar.actionResultViews.size} eagle results, count=${ar.unfilteredResultCountAfter}, commands=${ar.availableCommands.exists(_.commandsByProvince.nonEmpty)}"
+              s"game-update: ${ar.actionResultViews.size} eagle results, historyCount=${ar.unfilteredResultCountAfter}, commands=${ar.availableCommands.exists(_.commandsByProvince.nonEmpty)}"
            )
          case GameUpdate.GameUpdateDetails.ShardokActionResultResponse(sr)   =>
            sr.shardokGameResponses.foreach(updateShardokBattle)
@@ -1131,6 +1347,11 @@ final class ClientState(
      textStore.clear()
      pendingTutorialDialogues = Vector.empty
      gameOverAnnounced = false
+      latestState = None
+      latestCommands = None
+      latestStatus = None
+      latestUnfilteredResultCount = 0
+      latestShardokBattles = Map.empty
    }
    if !latestGameId.contains(gameId) then seenTutorialDialogueIds = tutorialDialogueHistory.displayedIds(gameId)
    latestGameId = Some(gameId)
@@ -1318,7 +1539,7 @@ final class ClientState(
        latestStatus
          .map(s => s"${s.status} waiting=${s.waitingForFactionIds.mkString("[", ",", "]")}")
          .getOrElse("unknown")
-    s"game=${latestGameId.getOrElse("-")} count=$latestUnfilteredResultCount token=${commandToken.getOrElse("-")} status=$status"
+    s"game=${latestGameId.getOrElse("-")} historyCount=$latestUnfilteredResultCount token=${commandToken.getOrElse("-")} status=$status"
  }

  private def announceGameOver(): Unit = {
@@ -2156,6 +2377,8 @@ object Help {
      |  shardok-map [shardok-game-id]
      |  shardok-map-json [shardok-game-id]
      |  post-json [--force] <province-id> <SelectedCommandType> <JSON>
+      |    --force only overrides local march warnings; it does not bypass tokens or connection checks
+      |    historyCount is the subscription replay cursor; command posts use the displayed token, not historyCount
      |  post-shardok <index> [integer-roll]
      |  post-shardok <shardok-game-id> <index> [integer-roll]
      |  post-rest <province-id>
@@ -44,6 +44,7 @@ import net.eagle0.eagle.api.eagle.{
  HexMapResponse,
  HexMapsRequest,
  ShardokActionResultResponse,
+  SubscriptionAck,
  UpdateStreamResponse
 }
 import net.eagle0.eagle.api.selected_command as sc
@@ -124,6 +125,52 @@ class ClientStateTest extends AnyFlatSpec with Matchers {
    Help.text should include("example: create-normal 4 2  (four total players, up to two human players)")
  }

+  it should "distinguish the subscription history cursor from command posting" in {
+    Help.text should include("--force only overrides local march warnings")
+    Help.text should include("historyCount is the subscription replay cursor")
+    Help.text should include("command posts use the displayed token, not historyCount")
+  }
+
+  "ClientState" should "not regress its history cursor when a subscription acknowledgement arrives late" in {
+    val state = new ClientState
+    state.handle(
+      UpdateStreamResponse(
+        UpdateStreamResponse.ResponseDetails.GameUpdate(
+          GameUpdate(
+            gameId = 1L,
+            gameUpdateDetails = GameUpdate.GameUpdateDetails.ActionResultResponse(
+              ActionResultResponse(unfilteredResultCountAfter = 4)
+            )
+          )
+        )
+      )
+    )
+    state.handle(
+      UpdateStreamResponse(
+        UpdateStreamResponse.ResponseDetails.SubscriptionAck(
+          SubscriptionAck(gameId = 1L, success = true, confirmedResultCount = 0)
+        )
+      )
+    )
+
+    state.describeStatus should include("historyCount=4")
+  }
+
+  it should "reset its history cursor when switching games" in {
+    val state = new ClientState
+    state.handle(
+      UpdateStreamResponse(
+        UpdateStreamResponse.ResponseDetails.SubscriptionAck(
+          SubscriptionAck(gameId = 1L, success = true, confirmedResultCount = 4)
+        )
+      )
+    )
+
+    state.setCurrentGame(2L)
+
+    state.unfilteredResultCount shouldBe 0
+  }
+
  private def startingState(battleIds: String*): UpdateStreamResponse =
    UpdateStreamResponse(
      UpdateStreamResponse.ResponseDetails.GameUpdate(