From 4403978537d856d75247930413e6a721827d2968 Mon Sep 17 00:00:00 2001 From: Daniel Dolezal Date: Tue, 3 Nov 2026 10:38:20 +0100 Subject: [PATCH] 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 | 3 + server/blueprints/battlesnake.py | 47 ++++----- server/services/game_lifecycle.py | 75 ++++++++++++++ server/services/game_runtime.py | 4 +- server/services/gameplay_tracking.py | 5 +- tests/test_BattlesnakeLifecycle.py | 141 +++++++++++++++++++++++++++ 6 files changed, 242 insertions(+), 33 deletions(-) create mode 100644 server/services/game_lifecycle.py create mode 100644 tests/test_BattlesnakeLifecycle.py diff --git a/server/Server.py b/server/Server.py index 84a2ce0..bb9a9cb 100644 --- a/server/Server.py +++ b/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: diff --git a/server/blueprints/battlesnake.py b/server/blueprints/battlesnake.py index a29a89a..5775ded 100644 --- a/server/blueprints/battlesnake.py +++ b/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') diff --git a/server/services/game_lifecycle.py b/server/services/game_lifecycle.py new file mode 100644 index 0000000..fd9167f --- /dev/null +++ b/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']) diff --git a/server/services/game_runtime.py b/server/services/game_runtime.py index c704976..4bf667f 100644 --- a/server/services/game_runtime.py +++ b/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 diff --git a/server/services/gameplay_tracking.py b/server/services/gameplay_tracking.py index d312dc6..7212da1 100644 --- a/server/services/gameplay_tracking.py +++ b/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}")) diff --git a/tests/test_BattlesnakeLifecycle.py b/tests/test_BattlesnakeLifecycle.py new file mode 100644 index 0000000..4d76299 --- /dev/null +++ b/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, {})