"""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): - All worlds currently load the same test level (per-world content is M2/M7). ``world_id`` still scopes a distinct server process + session. - Port allocation is a linear scan of the configured range. - No crash-restart: if a server process dies, the next join respawns it. """ from __future__ import annotations import asyncio import json 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 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 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 # Give the engine a moment to bind the ENet port before we # hand the address to a client. await self._await_port(srv.port) srv.last_join = time.time() return {"host": settings.GAME_SERVER_HOST, "port": srv.port, "world_id": world_id} # ── internals ──────────────────────────────────────────────────────── def _spawn(self, world_id: str) -> _WorldServer: port = self._free_port() status_path = self._runtime_dir / f"world_{world_id}_{port}.json" status_path.unlink(missing_ok=True) 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), ] proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return _WorldServer(world_id=world_id, port=port, proc=proc, status_path=status_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_port(self, port: int, timeout: float = 8.0) -> None: deadline = time.time() + timeout while time.time() < deadline: # ENet is UDP; a bind failure means something (our server) has it. with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s: try: s.bind(("0.0.0.0", port)) except OSError: return await asyncio.sleep(0.25) 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()