Add GNS gameplay adapter

2026-08-16 00:30:19 -04:00
parent c0caeca254
commit f158619670
1 changed files with 301 additions and 0 deletions
@@ -0,0 +1,301 @@
+from __future__ import annotations
+
+import json
+import threading
+import time
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from client_session import ClientSession
+from gns_transport import EventType, GnsEvent, GnsServerTransport, RemoteEndpoint, SendResult
+from packet_codec import EncodedPacket, PacketCodecError, decode_packet
+from transport_policy import delivery_for_packet_type, is_snapshot_packet
+
+if TYPE_CHECKING:
+    from server_core import FalloutTogetherServer
+
+
+@dataclass(frozen=True)
+class GnsAdapterStats:
+    connections: int
+    thread_running: bool
+
+
+class GnsConnectionAdapter:
+    def __init__(self, transport: GnsServerTransport, connection_id: int) -> None:
+        self.transport = transport
+        self.connection_id = connection_id
+        self._closed = False
+        self._lock = threading.Lock()
+
+    def fileno(self) -> int:
+        with self._lock:
+            return -1 if self._closed else self.connection_id
+
+    def mark_remote_closed(self) -> None:
+        with self._lock:
+            self._closed = True
+
+    def sendall(self, framed_payload: bytes) -> None:
+        raw = bytes(framed_payload)
+        with self._lock:
+            if self._closed:
+                raise OSError("GNS connection is closed")
+        if not raw.endswith(b"\n") or raw.count(b"\n") != 1:
+            raise OSError("GNS compatibility adapter received invalid TCP framing")
+        payload = raw[:-1]
+        try:
+            packet = decode_packet(payload)
+        except PacketCodecError as error:
+            raise OSError(f"invalid outbound GNS packet: {error}") from error
+        packet_type = packet["type"]
+        encoded = EncodedPacket(packet_type, payload, delivery_for_packet_type(packet_type))
+        result = self.transport.send_encoded(self.connection_id, encoded)
+        if result in (SendResult.SENT, SendResult.DROPPED):
+            return
+        if result is SendResult.BACKPRESSURE and is_snapshot_packet(packet_type):
+            return
+        if result is SendResult.BACKPRESSURE:
+            raise OSError("GNS reliable send queue is under backpressure")
+        if result is SendResult.NOT_CONNECTED:
+            self.mark_remote_closed()
+            raise OSError("GNS connection is no longer active")
+        if result is SendResult.TOO_LARGE:
+            raise ValueError("GNS outbound packet exceeds maximum message size")
+        raise OSError("GNS outbound send failed")
+
+    def close(self) -> None:
+        with self._lock:
+            if self._closed:
+                return
+            self._closed = True
+        try:
+            self.transport.disconnect(self.connection_id, debug="Commonwealth Online disconnect")
+        except Exception:
+            pass
+
+
+class GnsGameplayAdapter:
+    def __init__(self, server: "FalloutTogetherServer", transport: GnsServerTransport) -> None:
+        self.server = server
+        self.transport = transport
+        self._clients: dict[int, ClientSession] = {}
+        self._connected_monotonic: dict[int, float] = {}
+        self._lock = threading.RLock()
+        self._thread: threading.Thread | None = None
+        self._thread_running = False
+
+    def get_stats(self) -> GnsAdapterStats:
+        with self._lock:
+            return GnsAdapterStats(len(self._clients), self._thread_running)
+
+    def _send_pre_session_end(self, connection_id: int, code: str, reason: str) -> None:
+        packet = self.server._build_session_ended_packet(code, reason)
+        self.transport.send_packet(connection_id, packet)
+
+    def _welcome_packet(self, client: ClientSession) -> dict:
+        return {
+            "type": "welcome",
+            "playerId": client.player_id,
+            "serverTime": time.time(),
+            "serverName": self.server.server_name,
+            "serverDescription": self.server.server_description,
+            "protocolVersion": self.server.PROTOCOL_VERSION if hasattr(self.server, "PROTOCOL_VERSION") else 2,
+            "capabilities": [
+                "interest-v1",
+                "hello-v2",
+                "bounded-framing",
+                "rate-limit-v1",
+                "movement-correction-v1",
+                "npc-authority-epoch-v1",
+                "player-state-v1",
+                "gns-message-transport-v1",
+            ],
+        }
+
+    def _handle_connected(self, event: GnsEvent) -> None:
+        from server_core import SESSION_ENDED_BANNED, SESSION_ENDED_RATE_LIMITED
+
+        endpoint = self.transport.remote_endpoint(event.connection_id)
+        if endpoint is None:
+            self.transport.disconnect(event.connection_id, debug="Remote endpoint unavailable")
+            return
+
+        ban_entry = self.server._ban_store.get_ban(endpoint.host)
+        if ban_entry is not None:
+            with self.server._lock:
+                self.server._stats["bannedConnectionsRejected"] += 1
+            self._send_pre_session_end(event.connection_id, SESSION_ENDED_BANNED, ban_entry.reason)
+            self.transport.disconnect(event.connection_id, debug="Banned")
+            return
+
+        if not self.server._allow_connect_attempt(endpoint.host):
+            with self.server._lock:
+                self.server._stats["pendingConnectionsRejected"] += 1
+            self._send_pre_session_end(
+                event.connection_id,
+                SESSION_ENDED_RATE_LIMITED,
+                "Too many connection attempts.",
+            )
+            self.transport.disconnect(event.connection_id, debug="Connection attempt rate limited")
+            return
+
+        connection = GnsConnectionAdapter(self.transport, event.connection_id)
+        client = self.server._assign_client(connection, (endpoint.host, endpoint.port))
+        if client is None:
+            self._send_pre_session_end(
+                event.connection_id,
+                SESSION_ENDED_RATE_LIMITED,
+                "Too many pending connections.",
+            )
+            connection.close()
+            return
+
+        connected_mono = time.monotonic()
+        self.server._init_rate_state(client, connected_mono)
+        with self._lock:
+            self._clients[event.connection_id] = client
+            self._connected_monotonic[event.connection_id] = connected_mono
+        try:
+            self.server._send_packet(client, self._welcome_packet(client))
+        except (OSError, ValueError):
+            self.server._disconnect_client(client)
+            self._purge_closed_clients()
+
+    def _handle_message(self, event: GnsEvent) -> None:
+        client = self._client_for_connection(event.connection_id)
+        if client is None:
+            self.transport.disconnect(event.connection_id, debug="Message before GNS admission")
+            return
+        try:
+            line = event.payload.decode("utf-8", errors="strict")
+        except UnicodeDecodeError:
+            self.server._reject_packet(client, "Packet is not valid UTF-8")
+            return
+        self.server._handle_line(client, line)
+        self._purge_closed_clients()
+
+    def _handle_oversize(self, event: GnsEvent) -> None:
+        from server_core import SESSION_ENDED_PACKET_TOO_LARGE
+
+        client = self._client_for_connection(event.connection_id)
+        if client is None:
+            self.transport.disconnect(event.connection_id, debug="Oversized pre-session packet")
+            return
+        self.server._end_client_session(
+            client,
+            code=SESSION_ENDED_PACKET_TOO_LARGE,
+            reason="Packet exceeded maximum message size.",
+        )
+        self._purge_closed_clients()
+
+    def _handle_disconnected(self, event: GnsEvent) -> None:
+        client = self._client_for_connection(event.connection_id)
+        if client is None:
+            return
+        if isinstance(client.connection, GnsConnectionAdapter):
+            client.connection.mark_remote_closed()
+        self.server._disconnect_client(client)
+        self._remove_mapping(event.connection_id)
+
+    def _client_for_connection(self, connection_id: int) -> ClientSession | None:
+        with self._lock:
+            return self._clients.get(connection_id)
+
+    def _remove_mapping(self, connection_id: int) -> None:
+        with self._lock:
+            self._clients.pop(connection_id, None)
+            self._connected_monotonic.pop(connection_id, None)
+
+    def _purge_closed_clients(self) -> None:
+        with self._lock:
+            stale = [
+                connection_id
+                for connection_id, client in self._clients.items()
+                if client.connection.fileno() < 0
+            ]
+        for connection_id in stale:
+            self._remove_mapping(connection_id)
+
+    def _enforce_timeouts(self) -> None:
+        from server_core import CLIENT_HANDSHAKE_TIMEOUT_SECONDS, CLIENT_IDLE_TIMEOUT_SECONDS
+
+        now_mono = time.monotonic()
+        now_wall = time.time()
+        with self._lock:
+            snapshot = [
+                (connection_id, client, self._connected_monotonic.get(connection_id, now_mono))
+                for connection_id, client in self._clients.items()
+            ]
+        for connection_id, client, connected_mono in snapshot:
+            if not client.gameplay_active and now_mono - connected_mono > CLIENT_HANDSHAKE_TIMEOUT_SECONDS:
+                self.transport.disconnect(connection_id, debug="Handshake timeout")
+                if isinstance(client.connection, GnsConnectionAdapter):
+                    client.connection.mark_remote_closed()
+                self.server._disconnect_client(client)
+                self._remove_mapping(connection_id)
+                continue
+            if (
+                client.gameplay_active
+                and client.last_packet_at is not None
+                and now_wall - client.last_packet_at > CLIENT_IDLE_TIMEOUT_SECONDS
+            ):
+                self.transport.disconnect(connection_id, debug="Idle timeout")
+                if isinstance(client.connection, GnsConnectionAdapter):
+                    client.connection.mark_remote_closed()
+                self.server._disconnect_client(client)
+                self._remove_mapping(connection_id)
+
+    def pump_once(self, max_events: int = 128) -> int:
+        processed = 0
+        while processed < max_events:
+            event = self.transport.poll()
+            if event is None:
+                break
+            processed += 1
+            if event.type is EventType.CONNECTED:
+                self._handle_connected(event)
+            elif event.type is EventType.MESSAGE:
+                self._handle_message(event)
+            elif event.type is EventType.OVERSIZE_MESSAGE:
+                self._handle_oversize(event)
+            elif event.type is EventType.DISCONNECTED:
+                self._handle_disconnected(event)
+        self._enforce_timeouts()
+        self._purge_closed_clients()
+        return processed
+
+    def _run(self) -> None:
+        while True:
+            with self._lock:
+                if not self._thread_running:
+                    break
+            processed = self.pump_once()
+            if processed == 0:
+                time.sleep(0.002)
+
+    def start(self) -> None:
+        with self._lock:
+            if self._thread_running:
+                return
+            self._thread_running = True
+            self._thread = threading.Thread(target=self._run, daemon=True, name="CommonwealthOnlineGNS")
+            self._thread.start()
+
+    def stop(self) -> None:
+        with self._lock:
+            self._thread_running = False
+            thread = self._thread
+            self._thread = None
+            clients = list(self._clients.items())
+        if thread is not None and thread is not threading.current_thread():
+            thread.join(timeout=1.0)
+        for connection_id, client in clients:
+            if isinstance(client.connection, GnsConnectionAdapter):
+                client.connection.mark_remote_closed()
+            self.transport.disconnect(connection_id, debug="GNS adapter stopped")
+            self.server._disconnect_client(client)
+        with self._lock:
+            self._clients.clear()
+            self._connected_monotonic.clear()
+        self.transport.close()