Run dropped game cleanup in background (#8756)

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

@@ -1,6 +1,7 @@
 package net.eagle0.eagle.service
 
 import java.util.concurrent.{ConcurrentHashMap, Executor, Executors, TimeUnit}
+import java.util.concurrent.atomic.AtomicBoolean
 
 import scala.concurrent.{ExecutionContext, Future}
 import scala.util.{Failure, Random, Success}
@@ -115,6 +116,11 @@ object GamesManager {
     thread.setDaemon(true)
     thread
   }
+  private[service] val defaultDroppedGameCleanupExecutor  = Executors.newSingleThreadExecutor { runnable =>
+    val thread = new Thread(runnable, "dropped-game-cleanup")
+    thread.setDaemon(true)
+    thread
+  }
   private val LoadedGameIdleTimeoutMillis: Long           = envLong(
     name = "EAGLE_LOADED_GAME_IDLE_TIMEOUT_MILLIS",
     defaultValue = TimeUnit.MINUTES.toMillis(
@@ -610,7 +616,8 @@ class GamesManager(
     val userService: UserService,
     heroNameCache: HeroNameCache,
     currentSchemaVersion: Int,
-    initialTextDispatchExecutor: Executor = GamesManager.defaultInitialTextDispatchExecutor
+    initialTextDispatchExecutor: Executor = GamesManager.defaultInitialTextDispatchExecutor,
+    droppedGameCleanupExecutor: Executor = GamesManager.defaultDroppedGameCleanupExecutor
 ) extends BattleUpdateReceiver
     with LlmUpdateReceiver {
 
@@ -646,7 +653,9 @@ class GamesManager(
   // Per-game locks prevent concurrent loads of the same game. Lock order is always
   // per-game lock -> GamesManager monitor; callers must finish lazy loading before
   // entering a GamesManager synchronized section.
-  private val gameLoadingLocks = new ConcurrentHashMap[GameId, AnyRef]()
+  private val gameLoadingLocks          = new ConcurrentHashMap[GameId, AnyRef]()
+  private val droppedGamesStorageLock   = new Object()
+  private val droppedGameCleanupRunning = new AtomicBoolean(false)
 
   def gameControllerInfos: Map[GameId, ControllerInfo] =
     readState(_.controllerInfosSnapshot)
@@ -913,16 +922,18 @@ class GamesManager(
   private def saveDroppedGames(droppedGames: Vector[DroppedGame]): Unit =
     persister.save(GamesManager.DroppedGamesKey, DroppedGames(droppedGames = droppedGames).toByteArray): Unit
 
-  private def recordDroppedGame(gameId: GameId, droppedAtMillis: Long): Unit = {
-    val withoutExisting = readDroppedGamesFromStorage().filterNot(_.gameId == gameId)
-    saveDroppedGames(withoutExisting :+ DroppedGame(gameId = gameId, droppedTimestampMillis = droppedAtMillis))
-  }
+  private def recordDroppedGame(gameId: GameId, droppedAtMillis: Long): Unit =
+    droppedGamesStorageLock.synchronized {
+      val withoutExisting = readDroppedGamesFromStorage().filterNot(_.gameId == gameId)
+      saveDroppedGames(withoutExisting :+ DroppedGame(gameId = gameId, droppedTimestampMillis = droppedAtMillis))
+    }
 
-  private def removeDroppedGame(gameId: GameId): Unit = {
-    val existing = readDroppedGamesFromStorage()
-    val updated  = existing.filterNot(_.gameId == gameId)
-    if updated.size != existing.size then saveDroppedGames(updated)
-  }
+  private def removeDroppedGame(gameId: GameId): Unit =
+    droppedGamesStorageLock.synchronized {
+      val existing = readDroppedGamesFromStorage()
+      val updated  = existing.filterNot(_.gameId == gameId)
+      if updated.size != existing.size then saveDroppedGames(updated)
+    }
 
   private def deleteS3Files(gameId: GameId): Unit =
     if S3Credentials.isEnabled then {
@@ -941,30 +952,64 @@ class GamesManager(
     !saveDir.exists() || deleteDirectory(saveDir)
   }
 
-  private def cleanupExpiredDroppedGames(nowMillis: Long): Unit =
-    if transitionState(
-        _.withLastDroppedGameCleanupCheckIfDue(nowMillis, GamesManager.DroppedGameCleanupIntervalMillis)
-      )
-    then {
-      val droppedGames        = readDroppedGamesFromStorage()
-      val (expired, retained) =
-        droppedGames.partition(droppedGame =>
-          nowMillis - droppedGame.droppedTimestampMillis >= GamesManager.DroppedGameRetentionMillis
+  private def requestExpiredDroppedGameCleanup(nowMillis: Long): Unit = {
+    val cleanupDue = transitionState(
+      _.withLastDroppedGameCleanupCheckIfDue(nowMillis, GamesManager.DroppedGameCleanupIntervalMillis)
+    )
+
+    if cleanupDue && droppedGameCleanupRunning.compareAndSet(false, true) then
+      try
+        droppedGameCleanupExecutor.execute(() =>
+          try cleanupExpiredDroppedGames(nowMillis)
+          catch {
+            case NonFatal(exception) =>
+              SimpleTimedLogger.printLogger.logLine(s"Dropped-game cleanup failed: $exception")
+              Sentry.captureException(exception): Unit
+          } finally droppedGameCleanupRunning.set(false)
         )
-
-      if expired.nonEmpty then {
-        val failedDeletes = expired.filterNot(droppedGame => deleteDroppedGameStorage(droppedGame.gameId))
-        val failedGameIds = failedDeletes.map(_.gameId).toSet
-        val stillRetained = retained ++ expired.filter(droppedGame => failedGameIds.contains(droppedGame.gameId))
-        saveDroppedGames(stillRetained)
-
-        if failedDeletes.nonEmpty then
-          SimpleTimedLogger.printLogger.logLine(
-            s"Failed to delete dropped game storage for: ${failedDeletes.map(_.gameId.toHexString).mkString(", ")}"
-          )
+      catch {
+        case NonFatal(exception) =>
+          droppedGameCleanupRunning.set(false)
+          SimpleTimedLogger.printLogger.logLine(s"Failed to schedule dropped-game cleanup: $exception")
+          Sentry.captureException(exception): Unit
       }
+    end if
+  }
+
+  private def cleanupExpiredDroppedGames(nowMillis: Long): Unit = {
+    val expired = droppedGamesStorageLock.synchronized {
+      readDroppedGamesFromStorage().filter(droppedGame =>
+        nowMillis - droppedGame.droppedTimestampMillis >= GamesManager.DroppedGameRetentionMillis
+      )
     }
 
+    if expired.nonEmpty then {
+      val failedDeletes       = expired.filterNot { droppedGame =>
+        try deleteDroppedGameStorage(droppedGame.gameId)
+        catch {
+          case NonFatal(exception) =>
+            SimpleTimedLogger.printLogger.logLine(
+              s"Failed to delete dropped game ${droppedGame.gameId.toHexString}: $exception"
+            )
+            Sentry.captureException(exception): Unit
+            false
+        }
+      }
+      val successfullyDeleted = expired.toSet -- failedDeletes
+
+      droppedGamesStorageLock.synchronized {
+        val current = readDroppedGamesFromStorage()
+        val updated = current.filterNot(successfullyDeleted.contains)
+        if updated.size != current.size then saveDroppedGames(updated)
+      }
+
+      if failedDeletes.nonEmpty then
+        SimpleTimedLogger.printLogger.logLine(
+          s"Failed to delete dropped game storage for: ${failedDeletes.map(_.gameId.toHexString).mkString(", ")}"
+        )
+    }
+  }
+
   /**
    * Waits for the flush marker file to exist before loading games from disk. This ensures that during blue-green
    * deployments, we don't serve stale game data.
@@ -1676,7 +1721,7 @@ class GamesManager(
 
   private def save(): Unit = this.synchronized {
     val now = System.currentTimeMillis()
-    cleanupExpiredDroppedGames(now)
+    requestExpiredDroppedGameCleanup(now)
 
     // With lazy loading, gameControllerInfos only contains games that have been loaded.
     // We must merge with existing games.e0es to avoid losing unloaded games.

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

@@ -63,6 +63,8 @@ import org.scalatest.BeforeAndAfterEach
 
 class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with BeforeAndAfterEach with ProtoMatchers {
 
+  private val directExecutor: Executor = command => command.run()
+
   private class RecordingExecutor extends Executor {
     private var commands = Vector.empty[Runnable]
 
@@ -320,7 +322,8 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
       hexMaps = Map.empty,
       userService = new NoOpUserService,
       heroNameCache = new HeroNameCache(targetPoolSize = 0),
-      currentSchemaVersion = 3
+      currentSchemaVersion = 3,
+      droppedGameCleanupExecutor = directExecutor
     )
 
     manager.gamesForWithoutBlocking(userId) shouldBe empty
@@ -480,7 +483,8 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
     userService = new NoOpUserService,
     heroNameCache = new HeroNameCache(targetPoolSize = 0),
     currentSchemaVersion = 0,
-    initialTextDispatchExecutor = initialTextDispatchExecutor
+    initialTextDispatchExecutor = initialTextDispatchExecutor,
+    droppedGameCleanupExecutor = directExecutor
   )
 
   // Helper: allow games.e0es reads to return empty (for save() merging)
@@ -549,7 +553,8 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
       hexMaps = Map.empty,
       userService = new NoOpUserService,
       heroNameCache = new HeroNameCache(targetPoolSize = 0),
-      currentSchemaVersion = 0
+      currentSchemaVersion = 0,
+      droppedGameCleanupExecutor = directExecutor
     )
 
     val mergedController = gamesManager.withLiveHumanClientsFromCurrentController(replacementController)
@@ -645,7 +650,8 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
       hexMaps = Map.empty,
       userService = new NoOpUserService,
       heroNameCache = new HeroNameCache(targetPoolSize = 0),
-      currentSchemaVersion = 0
+      currentSchemaVersion = 0,
+      droppedGameCleanupExecutor = directExecutor
     )
 
     gamesManager.receiveStreamingLlmFailure(failedRequest)
@@ -1032,7 +1038,8 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
       hexMaps = Map.empty,
       userService = new NoOpUserService,
       heroNameCache = new HeroNameCache(targetPoolSize = 0),
-      currentSchemaVersion = 0
+      currentSchemaVersion = 0,
+      droppedGameCleanupExecutor = directExecutor
     )
 
     gamesManager.dropGame(userId = UserId("alice"), gameId = gameId) shouldBe
@@ -1072,7 +1079,8 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
       hexMaps = Map.empty,
       userService = new NoOpUserService,
       heroNameCache = new HeroNameCache(targetPoolSize = 0),
-      currentSchemaVersion = 0
+      currentSchemaVersion = 0,
+      droppedGameCleanupExecutor = directExecutor
     )
 
     gamesManager.dropGame(userId = UserId("alice"), gameId = gameId) shouldBe
@@ -1108,7 +1116,8 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
       hexMaps = Map.empty,
       userService = new NoOpUserService,
       heroNameCache = new HeroNameCache(targetPoolSize = 0),
-      currentSchemaVersion = 0
+      currentSchemaVersion = 0,
+      droppedGameCleanupExecutor = directExecutor
     )
 
     gamesManager.dropGame(userId = UserId("charlie"), gameId = gameId) shouldBe
@@ -1131,7 +1140,8 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
       hexMaps = Map.empty,
       userService = new NoOpUserService,
       heroNameCache = new HeroNameCache(targetPoolSize = 0),
-      currentSchemaVersion = 0
+      currentSchemaVersion = 0,
+      droppedGameCleanupExecutor = directExecutor
     )
 
     // Lazy loading will try to read games.e0es from storage - return empty to indicate no games
@@ -1228,7 +1238,8 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
       hexMaps = Map.empty,
       userService = new NoOpUserService,
       heroNameCache = new HeroNameCache(targetPoolSize = 0),
-      currentSchemaVersion = 0
+      currentSchemaVersion = 0,
+      droppedGameCleanupExecutor = directExecutor
     )
 
     gamesManager
@@ -1248,9 +1259,10 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
     droppedGames.head.droppedTimestampMillis.should(be > 0L)
   }
 
-  it should "delete dropped game storage after the retention period" in {
+  it should "delete expired dropped game storage in the background" in {
     val expiredGameId     = gameId.toLong
     val expiredDroppedAt  = System.currentTimeMillis() - GamesManager.DroppedGameRetentionMillis - 1L
+    val cleanupExecutor   = new RecordingExecutor
     val initialDroppedLog = DroppedGames(
       droppedGames = Vector(
         net.eagle0.eagle.internal.running_games.DroppedGame(
@@ -1290,12 +1302,18 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
       hexMaps = Map.empty,
       userService = new NoOpUserService,
       heroNameCache = new HeroNameCache(targetPoolSize = 0),
-      currentSchemaVersion = 0
+      currentSchemaVersion = 0,
+      droppedGameCleanupExecutor = cleanupExecutor
     )
 
     gamesManager.dropGame(userId = UserId("alice"), gameId = 0x1234L) shouldBe
       DropGameResponse(result = SUCCESS_DROP_GAME_RESULT, gameId = 0x1234L)
 
+    cleanupExecutor.pendingCount shouldBe 1
+    sentinel.exists() shouldBe true
+
+    cleanupExecutor.runAll()
+
     saveDir.exists() shouldBe false
     val droppedGames = ProtoParser
       .parse[DroppedGames](new ByteArrayInputStream(persister.entries(GamesManager.DroppedGamesKey)))(using
@@ -1321,7 +1339,8 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
       hexMaps = Map.empty,
       userService = new NoOpUserService,
       heroNameCache = new HeroNameCache(targetPoolSize = 0),
-      currentSchemaVersion = 0
+      currentSchemaVersion = 0,
+      droppedGameCleanupExecutor = directExecutor
     )
 
     // No calls to storage should have been made yet
@@ -1344,7 +1363,8 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
       hexMaps = Map.empty,
       userService = new NoOpUserService,
       heroNameCache = new HeroNameCache(targetPoolSize = 0),
-      currentSchemaVersion = 0
+      currentSchemaVersion = 0,
+      droppedGameCleanupExecutor = directExecutor
     )
 
     // Lazy loading will attempt to read games.e0es from storage
@@ -1370,7 +1390,8 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
       hexMaps = Map.empty,
       userService = new NoOpUserService,
       heroNameCache = new HeroNameCache(targetPoolSize = 0),
-      currentSchemaVersion = 0
+      currentSchemaVersion = 0,
+      droppedGameCleanupExecutor = directExecutor
     )
 
     val loading = Future(gamesManager.loadGame(nonExistentGameId))
@@ -1422,7 +1443,8 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
       hexMaps = Map.empty,
       userService = new NoOpUserService,
       heroNameCache = new HeroNameCache(targetPoolSize = 0),
-      currentSchemaVersion = 0
+      currentSchemaVersion = 0,
+      droppedGameCleanupExecutor = directExecutor
     )
 
     // First call: cache is cold, reads storage exactly once.
@@ -1458,7 +1480,8 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
       hexMaps = Map.empty,
       userService = new NoOpUserService,
       heroNameCache = new HeroNameCache(targetPoolSize = 0),
-      currentSchemaVersion = 0
+      currentSchemaVersion = 0,
+      droppedGameCleanupExecutor = directExecutor
     )
 
     try {
@@ -1529,7 +1552,8 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
         hexMaps = Map.empty,
         userService = new NoOpUserService,
         heroNameCache = new HeroNameCache(targetPoolSize = 0),
-        currentSchemaVersion = 0
+        currentSchemaVersion = 0,
+        droppedGameCleanupExecutor = directExecutor
       )
       val _            = gamesManager.createTutorialGameForUser(UserId("tut"), gameId)