perf(battlesnake): defer ordered lifecycle bookkeeping

- Acknowledge start and end without waiting for persistence or dashboards.
- Keep runtime creation and finished-game removal on the response path.
- Serialize start, turn, and end bookkeeping per game in Quart tasks.
- Capture turn thinking before scheduling deferred persistence.
- Defer start metrics, logging, and stale-game pruning.
- Run legacy history exports in a worker thread.
- Record results and end metrics even when history export fails.
- Drain pending bookkeeping before closing storage during shutdown.
- Test sub-500 ms responses with slow backends and real SQLite.
- Cover event ordering, export failures, and rejected user agents.

server/Server.py

@@ -21,6 +21,7 @@ from server.blueprints import (
   create_metrics_blueprint,
   create_dashboard_blueprint,
 )
+from server.services.game_lifecycle import GameLifecycleService
 from server.services import (
   DashboardEventsService,
   DashboardWebSocketHub,
@@ -106,6 +107,7 @@ class Server:
 
     self.app = Quart('Battlesnake', template_folder=os.path.join(data_path, 'templates', 'side'), static_folder=os.path.join(data_path, 'templates', 'files'))
 
+    self.game_lifecycle = GameLifecycleService(self)
     self.app.register_blueprint(create_battlesnake_blueprint(self))
     self.app.register_blueprint(create_metrics_blueprint(self))
     self.app.register_blueprint(create_dashboard_blueprint(self))
@@ -143,6 +145,7 @@ class Server:
 
     @self.app.after_serving
     async def shutdown_state_storage():
+      await self.game_lifecycle.drain()
       await self.dashboard_events_service.stop_listener()
       await self.metrics_collector.close()
       if self.gameplay_database is not None:

server/blueprints/battlesnake.py

@@ -5,7 +5,6 @@ from quart import Blueprint, request, jsonify
 
 from quart_common.web.decorators import require_user_agent
 from quart_common.web.logger import await_log
-from server.database import StorageLoader
 from snakes import DEFAULT_SNAKE_CONFIG
 from server.GameBoard import GameBoard
 from server.Files import read_file
@@ -47,14 +46,13 @@ def create_battlesnake_blueprint(server:'Server') -> Blueprint:
   @require_user_agent(("BattlesnakeEngine", "BattlesnakeArena"), abort_code=404)
   async def on_start():
     server.metrics_collector.record_http_request('start')
-    await server.game_runtime.prune_stale_games()
     game_state = await request.get_json()
-    game_board = await server.game_runtime.create_game_board(game_state)
-    await server.gameplay_tracking.record_gameplay_start(game_state, game_board)
-    await server.dashboard_query.push_dashboard_games_update(
-      game_state, trigger='game_started',
+    game_board = await server.game_runtime.create_game_board(game_state, publish_metrics=False)
+    active_count = len(server.game_runtime.running_games)
+    server.game_lifecycle.submit(
+      game_state['game']['id'],
+      lambda: server.game_lifecycle.started(game_state, game_board, active_count),
     )
-    await await_log(server.logger.info(f'GAME START: {game_state['game']}'))
     return 'ok'
 
   @blueprint.post('/move')
@@ -79,8 +77,10 @@ def create_battlesnake_blueprint(server:'Server') -> Blueprint:
     except TimeoutError:
       await await_log(server.logger.warning(f'MOVE TIMEOUT: turn={game_state.get("turn")}, game={game_id}, returning fallback {next_move!r}'))
 
-    await server.gameplay_tracking.record_gameplay_turn(game_state, next_move, game_board)
-    await server.dashboard_query.push_dashboard_game_replay_update(game_id)
+    thinking = server.gameplay_tracking._extract_latest_snake_thinking(game_board)
+    server.game_lifecycle.submit(
+      game_id, lambda: server.game_lifecycle.turn(game_state, next_move, thinking),
+    )
     elapsed_ms = (time.perf_counter() - move_started) * 1000.0
     await server.metrics_collector.record_move(next_move, elapsed_ms)
 
@@ -93,29 +93,16 @@ def create_battlesnake_blueprint(server:'Server') -> Blueprint:
   @require_user_agent(("BattlesnakeEngine", "BattlesnakeArena"), abort_code=404)
   async def on_end():
     server.metrics_collector.record_http_request('end')
-    await server.game_runtime.prune_stale_games()
     game_state = await request.get_json()
-    if server.store_game_state:
-      game_board = cast(GameBoard, await server.game_runtime.get_game_board(game_state, end=True))
-      if server.check_tls_security:
-        await game_board.save(
-          StorageLoader.build(server.storage_type),
-          file_path=os.path.join(server.data_path, 'data'),
-          database=os.getenv('EDGEDB_DATABASE', None),
-          tls_security=None,
-        )
-      else:
-        await game_board.save(
-          StorageLoader.build(server.storage_type),
-          file_path=os.path.join(server.data_path, 'data'),
-          database=os.getenv('EDGEDB_DATABASE', None),
-        )
-
-    await server.gameplay_tracking.record_gameplay_end(game_state)
-    await server.dashboard_query.push_dashboard_games_update(game_state)
-    await await_log(server.logger.info(f'GAME ENDED: Winner is {[x['name'] for x in game_state['board']['snakes']]}'))
+    game_id = game_state['game']['id']
+    game_board = server.game_runtime.running_games.get(game_id)
+    if game_board is not None:
+      game_board.read_game_data(game_state)
+      game_board.end_game(game_state)
     await server.game_runtime.delete_game_board(game_state)
-    await server.metrics_collector.record_game_end(game_state)
+    server.game_lifecycle.submit(
+      game_id, lambda: server.game_lifecycle.ended(game_state, game_board),
+    )
     return 'ok'
 
   # @blueprint.get('/cleanup')

server/services/game_lifecycle.py

@@ -0,0 +1,75 @@
+"""Ordered, application-managed game bookkeeping outside HTTP response budgets."""
+
+import asyncio
+from collections.abc import Awaitable, Callable
+import os
+
+from quart_common.web.logger import await_log
+from server.database import StorageLoader
+
+class GameLifecycleService:
+  def __init__(self, server):
+    self.server = server
+    self.pending: dict[str, asyncio.Future] = {}
+
+  def submit(self, game_id:str, operation:Callable[[], Awaitable[None]]) -> None:
+    previous = self.pending.get(game_id)
+    completed = asyncio.get_running_loop().create_future()
+    self.pending[game_id] = completed
+
+    async def run():
+      try:
+        if previous is not None:
+          await asyncio.shield(previous)
+        await operation()
+      except Exception:
+        self.server.app.logger.exception('Game bookkeeping failed for %s', game_id)
+      finally:
+        if not completed.done():
+          completed.set_result(None)
+        if self.pending.get(game_id) is completed:
+          self.pending.pop(game_id, None)
+
+    self.server.app.add_background_task(run)
+
+  async def drain(self) -> None:
+    while self.pending:
+      await asyncio.gather(*(asyncio.shield(task) for task in list(self.pending.values())))
+
+  async def started(self, game_state:dict, game_board, active_count:int) -> None:
+    server = self.server
+    await server.metrics_collector.record_game_started(active_count)
+    await server.gameplay_tracking.record_gameplay_start(game_state, game_board)
+    await server.dashboard_query.push_dashboard_games_update(game_state, trigger='game_started')
+    await await_log(server.logger.info(f'GAME START: {game_state["game"]}'))
+    await server.game_runtime.prune_stale_games()
+
+  async def ended(self, game_state:dict, game_board) -> None:
+    server = self.server
+    try:
+      if server.store_game_state and game_board is not None:
+        kwargs = {
+          'file_path': os.path.join(server.data_path, 'data'),
+          'database': os.getenv('EDGEDB_DATABASE'),
+        }
+        if server.check_tls_security:
+          kwargs['tls_security'] = None
+        # Legacy exporters include synchronous serialization and database calls.
+        # Keep those off the request event loop as well as off the response path.
+        def export():
+          asyncio.run(game_board.save(StorageLoader.build(server.storage_type), **kwargs))
+
+        await asyncio.to_thread(export)
+    finally:
+      # A history export failure must not suppress the game result or metrics.
+      await server.gameplay_tracking.record_gameplay_end(game_state)
+      await server.metrics_collector.record_game_end(game_state)
+      await server.dashboard_query.push_dashboard_games_update(game_state)
+      await await_log(server.logger.info(
+        f'GAME ENDED: Winner is {[snake["name"] for snake in game_state["board"]["snakes"]]}'
+      ))
+
+  async def turn(self, game_state:dict, move:str|None, thinking:dict|None) -> None:
+    server = self.server
+    await server.gameplay_tracking.record_gameplay_turn_snapshot(game_state, move, thinking)
+    await server.dashboard_query.push_dashboard_game_replay_update(game_state['game']['id'])

server/services/game_runtime.py

@@ -18,7 +18,7 @@ class GameRuntimeService:
   def attach_metrics_collector(self, metrics_collector:MetricsCollector) -> None:
     self.metrics_collector = metrics_collector
 
-  async def create_game_board(self, game_state:dict) -> GameBoard:
+  async def create_game_board(self, game_state:dict, publish_metrics:bool=True) -> GameBoard:
     game_id = game_state['game']['id']
     new_game_board = GameBoard(
       game_id=game_id,
@@ -34,7 +34,7 @@ class GameRuntimeService:
     self.running_games[game_id] = new_game_board
     self.game_move_counts[game_id] = 0
     self.game_last_seen_unix[game_id] = int(time.time())
-    if self.metrics_collector is not None:
+    if publish_metrics and self.metrics_collector is not None:
       await self.metrics_collector.record_game_started(len(self.game_last_seen_unix))
     return new_game_board
 

server/services/gameplay_tracking.py

@@ -23,10 +23,13 @@ class GameplayTrackingService:
       await await_log(self.logger.warning(f"Gameplay DB start record failed:{error}"))
 
   async def record_gameplay_turn(self, game_state:dict, my_move:str, game_board:GameBoard) -> None:
+    thinking = self._extract_latest_snake_thinking(game_board)
+    await self.record_gameplay_turn_snapshot(game_state, my_move, thinking)
+
+  async def record_gameplay_turn_snapshot(self, game_state:dict, my_move:str|None, thinking:dict|None) -> None:
     if self.gameplay_database is None:
       return
     try:
-      thinking = self._extract_latest_snake_thinking(game_board)
       await self.gameplay_database.record_turn(game_state, my_move, thinking)
     except Exception as error:
       await await_log(self.logger.warning(f"Gameplay DB turn record failed:{error}"))

tests/test_BattlesnakeLifecycle.py

@@ -0,0 +1,141 @@
+import asyncio
+from copy import deepcopy
+import os
+import tempfile
+from time import perf_counter
+import unittest
+from unittest.mock import AsyncMock, Mock, patch
+
+from server.Server import Server
+
+class TestBattlesnakeLifecycle(unittest.IsolatedAsyncioTestCase):
+  def setUp(self):
+    self.server = Server(
+      data_path=os.path.dirname(os.path.dirname(__file__)),
+      snake_type='TemplateSnake', storage_type='LocalStorage',
+      gameplay_db_enabled=False,
+    )
+    snake = {
+      'id': 'snake-1', 'name': 'test', 'health': 100, 'length': 3,
+      'head': {'x': 1, 'y': 1},
+      'body': [{'x': 1, 'y': 1}] * 3,
+    }
+    self.state = {
+      'game': {
+        'id': 'game-1', 'source': 'test', 'map': 'standard', 'timeout': 500,
+        'ruleset': {'name': 'standard', 'version': 'v1'},
+      },
+      'turn': 0, 'you': snake,
+      'board': {'width': 11, 'height': 11, 'food': [], 'hazards': [], 'snakes': [snake]},
+    }
+    self.headers = {'User-Agent': 'BattlesnakeArena'}
+
+  async def test_slow_io_does_not_delay_acknowledgements_or_reorder_events(self):
+    events = []
+    gate = asyncio.Event()
+
+    async def slow_start(*args):
+      await gate.wait()
+      await asyncio.sleep(0.55)
+      events.append('start')
+
+    async def slow_end(*args):
+      await asyncio.sleep(0.55)
+      events.append('end')
+
+    async def record_turn(*args):
+      events.append('turn')
+
+    server = self.server
+    server.logger = Mock()
+    server.logger.info.return_value = None
+    server.gameplay_tracking.record_gameplay_start = slow_start
+    server.gameplay_tracking.record_gameplay_end = slow_end
+    server.gameplay_tracking.record_gameplay_turn_snapshot = record_turn
+    server.dashboard_query.push_dashboard_games_update = AsyncMock()
+    server.dashboard_query.push_dashboard_game_replay_update = AsyncMock()
+    server.metrics_collector.record_game_started = AsyncMock()
+    server.metrics_collector.record_game_end = AsyncMock()
+    client = server.app.test_client()
+
+    started = perf_counter()
+    response = await client.post('/start', json=self.state, headers=self.headers)
+    start_ms = (perf_counter() - started) * 1000
+    self.assertEqual(response.status_code, 200)
+    self.assertEqual(await response.get_data(), b'ok')
+    self.assertLess(start_ms, 500)
+    self.assertIn('game-1', server.game_runtime.running_games)
+    self.assertEqual(events, [])
+
+    # A real move request must not overtake the pending start DB write.
+    response = await client.post('/move', json=self.state, headers=self.headers)
+    self.assertEqual(response.status_code, 200)
+    self.assertIn((await response.get_json())['move'], ('up', 'down', 'left', 'right'))
+    end_state = deepcopy(self.state)
+    end_state['turn'] = 1
+    started = perf_counter()
+    response = await client.post('/end', json=end_state, headers=self.headers)
+    end_ms = (perf_counter() - started) * 1000
+    self.assertEqual(response.status_code, 200)
+    self.assertLess(end_ms, 500)
+    self.assertNotIn('game-1', server.game_runtime.running_games)
+    self.assertEqual(events, [])
+
+    gate.set()
+    await server.game_lifecycle.drain()
+    self.assertEqual(events, ['start', 'turn', 'end'])
+    self.assertEqual(server.game_lifecycle.pending, {})
+    server.metrics_collector.record_game_started.assert_awaited_once_with(1)
+    server.metrics_collector.record_game_end.assert_awaited_once_with(end_state)
+    print(f'Lifecycle HTTP timings with >500 ms I/O: start={start_ms:.1f} ms, end={end_ms:.1f} ms')
+
+  async def test_export_failure_still_records_result_and_cleans_runtime(self):
+    server = self.server
+    server.logger = Mock()
+    server.logger.info.return_value = None
+    server.gameplay_tracking.record_gameplay_start = AsyncMock()
+    server.gameplay_tracking.record_gameplay_end = AsyncMock()
+    server.dashboard_query.push_dashboard_games_update = AsyncMock()
+    server.metrics_collector.record_game_end = AsyncMock()
+    client = server.app.test_client()
+    await client.post('/start', json=self.state, headers=self.headers)
+    await server.game_lifecycle.drain()
+    board = server.game_runtime.running_games['game-1']
+    board.save = AsyncMock(side_effect=OSError('export failed'))
+    server.store_game_state = True
+    with patch('server.services.game_lifecycle.StorageLoader.build', return_value=object):
+      response = await client.post('/end', json=self.state, headers=self.headers)
+      await server.game_lifecycle.drain()
+    self.assertEqual(response.status_code, 200)
+    self.assertNotIn('game-1', server.game_runtime.running_games)
+    board.save.assert_awaited_once()
+    server.gameplay_tracking.record_gameplay_end.assert_awaited_once()
+    server.metrics_collector.record_game_end.assert_awaited_once()
+
+  async def test_real_sqlite_lifecycle_persists_after_acknowledgements(self):
+    with tempfile.TemporaryDirectory() as folder:
+      server = Server(
+        data_path=folder, snake_type='TemplateSnake', storage_type='LocalStorage',
+        gameplay_db_path=os.path.join(folder, 'gameplay.sqlite3'),
+      )
+      server.logger = Mock()
+      server.logger.info.return_value = None
+      async with server.app.test_app():
+        client = server.app.test_client()
+        timings = {}
+        for endpoint in ('start', 'end'):
+          started = perf_counter()
+          response = await client.post('/' + endpoint, json=self.state, headers=self.headers)
+          timings[endpoint] = (perf_counter() - started) * 1000
+          self.assertEqual(response.status_code, 200)
+          self.assertLess(timings[endpoint], 500)
+        await server.game_lifecycle.drain()
+        replay = await server.gameplay_database.get_game_replay('game-1')
+        self.assertIsNotNone(replay)
+        self.assertNotEqual(replay['game']['status'], 'running')
+        print(f'Real SQLite HTTP timings: {timings}')
+
+  async def test_invalid_agent_does_not_schedule_work(self):
+    response = await self.server.app.test_client().post('/start', json=self.state)
+    self.assertEqual(response.status_code, 404)
+    self.assertEqual(self.server.game_lifecycle.pending, {})