"""Game-server orchestration. On ``POST /api/worlds/{id}/join`` the backend ensures a headless Godot dedicated-server process is running for that world and returns its address. One process per world, keyed by ``world_id``. A background reaper kills a world's server once its status file reports zero players for ``GAME_SERVER_IDLE_TIMEOUT_SECONDS``. This lives in-process (a module-level singleton) rather than as a separate service, so the backend MUST run single-worker -- which it does (``uvicorn app.main:app`` with no ``--workers``). The Godot binary and project are on the host (see ``settings.GODOT_BIN`` / ``GAME_PROJECT_DIR``); the backend spawns them as subprocesses. Design notes / limits (PC3): - Port allocation is a linear scan of the configured range. - No crash-restart: if a server process dies, the next join respawns it. With world content now durable in Postgres (world_persistence_impl.md D2), this is genuinely safe rather than merely tolerated: the sim process holds no unsaved world content, so a reap/crash/kill can never lose world data. Per-spawn auth token (world_persistence_impl.md §7.3): each spawned server gets a random ``secrets.token_urlsafe(32)`` on its command line, which it uses as a Bearer token against ``/api/internal`` to pull its world's content read-only before binding its ENet port. The token grants access to exactly one world and dies with its process (the reaper's ``del self._servers[wid]`` invalidates it; see ``validate_token``). """ from __future__ import annotations import asyncio import hmac import json import secrets import socket import subprocess import tempfile import time from dataclasses import dataclass, field from pathlib import Path from app.config import settings @dataclass class _WorldServer: world_id: str port: int proc: subprocess.Popen status_path: Path token: str log_path: Path | None = None started_at: float = field(default_factory=time.time) last_join: float = field(default_factory=time.time) def status(self) -> dict: try: return json.loads(self.status_path.read_text()) except (OSError, ValueError): return {} def player_count(self) -> int: return int(self.status().get("players", 0)) def is_ready(self) -> bool: return self.status().get("ready") is True def alive(self) -> bool: return self.proc.poll() is None def stop(self) -> None: if self.alive(): self.proc.terminate() try: self.proc.wait(timeout=10) except subprocess.TimeoutExpired: self.proc.kill() self.status_path.unlink(missing_ok=True) class GameServerSupervisor: def __init__(self) -> None: self._servers: dict[str, _WorldServer] = {} self._lock = asyncio.Lock() self._reaper: asyncio.Task | None = None self._runtime_dir = Path(tempfile.gettempdir()) / "agp-game-servers" self._runtime_dir.mkdir(parents=True, exist_ok=True) # ── lifecycle ──────────────────────────────────────────────────────── def start(self) -> None: if not settings.GAME_SERVERS_ENABLED: return if self._reaper is None or self._reaper.done(): self._reaper = asyncio.create_task(self._reap_loop()) async def shutdown(self) -> None: if self._reaper: self._reaper.cancel() async with self._lock: for srv in list(self._servers.values()): srv.stop() self._servers.clear() # ── join ───────────────────────────────────────────────────────────── async def ensure_server(self, world_id: str) -> dict: """Return ``{"host", "port", "world_id"}`` for the world's dedicated server, spawning it if needed. Raises RuntimeError if orchestration is disabled or no port/engine is available.""" if not settings.GAME_SERVERS_ENABLED: raise RuntimeError("game servers are not enabled on this backend") async with self._lock: srv = self._servers.get(world_id) if srv is not None and not srv.alive(): srv.stop() srv = None del self._servers[world_id] if srv is None: srv = self._spawn(world_id) self._servers[world_id] = srv try: await self._await_ready(srv) except TimeoutError: # Real failure now, not a silent "hope it's fine" -- # see _await_ready's docstring. Don't leave a possibly # half-started process registered. srv.stop() del self._servers[world_id] raise RuntimeError( f"dedicated server for world {world_id} did not become ready in time" ) srv.last_join = time.time() return {"host": settings.GAME_SERVER_HOST, "port": srv.port, "world_id": world_id} def validate_token(self, world_id: str, token: str) -> bool: """`require_game_server` dependency support (world_persistence_impl.md §7.3): true iff *token* is the live token for *world_id*'s server. A dict lookup, as the doc specifies, plus a constant-time compare since it's a bearer credential.""" srv = self._servers.get(str(world_id)) if srv is None or not srv.alive(): return False return hmac.compare_digest(srv.token, token) # ── internals ──────────────────────────────────────────────────────── def _spawn(self, world_id: str) -> _WorldServer: port = self._free_port() token = secrets.token_urlsafe(32) status_path = self._runtime_dir / f"world_{world_id}_{port}.json" status_path.unlink(missing_ok=True) log_path = self._runtime_dir / f"world_{world_id}_{port}.log" cmd = [ settings.GODOT_BIN, "--headless", "--path", settings.GAME_PROJECT_DIR, "res://scenes/server.tscn", "--", "--port", str(port), "--world", str(world_id), "--status-file", str(status_path), "--api-url", settings.GAME_SERVER_INTERNAL_API_URL, "--server-token", token, ] log_file = open(log_path, "w") proc = subprocess.Popen(cmd, stdout=log_file, stderr=subprocess.STDOUT) return _WorldServer( world_id=world_id, port=port, proc=proc, status_path=status_path, token=token, log_path=log_path ) def _free_port(self) -> int: used = {s.port for s in self._servers.values()} for p in range(settings.GAME_SERVER_PORT_MIN, settings.GAME_SERVER_PORT_MAX + 1): if p in used: continue with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s: try: s.bind(("0.0.0.0", p)) return p except OSError: continue raise RuntimeError("no free game-server port in the configured range") async def _await_ready(self, srv: _WorldServer) -> None: """Wait for a ``"ready": true`` field in the status file (world_persistence_impl.md §7.4). This replaces the old ``_await_port``, which had two real bugs: it treated a bound UDP port as "ready", but with content loading the port binds *before* the server has fetched its world's rooms and chunks (world_persistence_impl.md §7.2 step 4) -- so a client could get handed an address for a server still mid-fetch. And it *failed open*: past its 8s deadline it simply returned, with no error, so ``ensure_server`` reported success for a server that may never have started at all. This raises ``TimeoutError`` instead, which ``ensure_server`` turns into a real 503 for the joining client, and also bails out early (same exception) if the process has already exited. """ deadline = time.time() + settings.GAME_SERVER_READY_TIMEOUT_SECONDS while time.time() < deadline: if srv.is_ready(): return if not srv.alive(): log_snippet = "" if srv.log_path and srv.log_path.exists(): try: log_snippet = srv.log_path.read_text()[-500:] except OSError: pass raise TimeoutError( f"server process for world {srv.world_id} exited before becoming ready" + (f": {log_snippet}" if log_snippet else "") ) await asyncio.sleep(0.25) log_snippet = "" if srv.log_path and srv.log_path.exists(): try: log_snippet = srv.log_path.read_text()[-500:] except OSError: pass raise TimeoutError( f"server for world {srv.world_id} did not report ready within " f"{settings.GAME_SERVER_READY_TIMEOUT_SECONDS}s" + (f": {log_snippet}" if log_snippet else "") ) async def _reap_loop(self) -> None: try: while True: await asyncio.sleep(30) async with self._lock: now = time.time() for wid, srv in list(self._servers.items()): if not srv.alive(): srv.stop() del self._servers[wid] continue idle_for = now - srv.last_join if idle_for > settings.GAME_SERVER_IDLE_TIMEOUT_SECONDS and srv.player_count() == 0: srv.stop() del self._servers[wid] except asyncio.CancelledError: pass def debug_state(self) -> list[dict]: return [ { "world_id": s.world_id, "port": s.port, "alive": s.alive(), "players": s.player_count(), "uptime_s": round(time.time() - s.started_at), } for s in self._servers.values() ] supervisor = GameServerSupervisor()