From f158619670eb918a786964019ea0283566940a77 Mon Sep 17 00:00:00 2001 From: Nomads_Reach <144523850+NomadsReach@users.noreply.github.com> Date: Sun, 16 Aug 2026 00:30:19 -0400 Subject: [PATCH] Add GNS gameplay adapter --- server/gns_gameplay_adapter.py | 301 +++++++++++++++++++++++++++++++++ 1 file changed, 301 insertions(+) create mode 100644 server/gns_gameplay_adapter.py diff --git a/server/gns_gameplay_adapter.py b/server/gns_gameplay_adapter.py new file mode 100644 index 0000000..ed9fd62 --- /dev/null +++ b/server/gns_gameplay_adapter.py @@ -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()