diff --git a/server/gns_gameplay_adapter.py b/server/gns_gameplay_adapter.py index db2eeca..e5f8d21 100644 --- a/server/gns_gameplay_adapter.py +++ b/server/gns_gameplay_adapter.py @@ -6,8 +6,10 @@ from dataclasses import dataclass from typing import TYPE_CHECKING from client_session import ClientSession +from gns_snapshot_envelope import SnapshotEnvelopeError, decode_snapshot, encode_snapshot from gns_transport import EventType, GnsEvent, GnsServerTransport, GnsTransportError, SendResult from packet_codec import EncodedPacket, PacketCodecError, decode_packet +from snapshot_sequence import SequenceCounter, SequenceWindow from transport_policy import delivery_for_packet_type, is_snapshot_packet if TYPE_CHECKING: @@ -26,6 +28,10 @@ class GnsConnectionAdapter: self.connection_id = connection_id self._closed = False self._lock = threading.Lock() + self._snapshot_counters = { + "transform": SequenceCounter(), + "npcState": SequenceCounter(), + } def fileno(self) -> int: with self._lock: @@ -48,7 +54,15 @@ class GnsConnectionAdapter: 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)) + wire_payload = payload + if is_snapshot_packet(packet_type): + with self._lock: + sequence = self._snapshot_counters[packet_type].advance() + try: + wire_payload = encode_snapshot(packet_type, payload, sequence) + except SnapshotEnvelopeError as error: + raise ValueError(str(error)) from error + encoded = EncodedPacket(packet_type, wire_payload, delivery_for_packet_type(packet_type)) result = self.transport.send_encoded(self.connection_id, encoded) if result is SendResult.SENT: return @@ -84,6 +98,7 @@ class GnsGameplayAdapter: self.transport = transport self._clients: dict[int, ClientSession] = {} self._connected_monotonic: dict[int, float] = {} + self._incoming_snapshot_windows: dict[tuple[int, str], SequenceWindow] = {} self._lock = threading.RLock() self._thread: threading.Thread | None = None self._thread_running = False @@ -115,6 +130,7 @@ class GnsGameplayAdapter: "npc-authority-epoch-v1", "player-state-v1", "gns-message-transport-v1", + "gns-snapshot-sequence-v1", ], } @@ -167,15 +183,57 @@ class GnsGameplayAdapter: self.server._disconnect_client(client) self._purge_closed_clients() + def _account_transport_reject(self, client: ClientSession, reason: str, *, warning: bool = False) -> None: + if not self.server._allow_packet(client): + return + client.record_received(time.time()) + with self.server._lock: + self.server._stats["packetsReceived"] += 1 + self.server._reject_packet(client, reason, warning=warning) + 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 + + payload = event.payload try: - line = event.payload.decode("utf-8", errors="strict") + envelope = decode_snapshot(payload) + except SnapshotEnvelopeError as error: + self._account_transport_reject(client, f"Malformed GNS snapshot envelope: {error}") + return + + if envelope is not None: + window_key = (event.connection_id, envelope.packet_type) + with self._lock: + window = self._incoming_snapshot_windows.setdefault(window_key, SequenceWindow()) + accepted = window.accept(envelope.sequence) + if not accepted: + self._account_transport_reject(client, "Stale or duplicate GNS snapshot sequence", warning=False) + return + try: + packet = decode_packet(envelope.payload) + except PacketCodecError as error: + self._account_transport_reject(client, f"Invalid GNS snapshot payload: {error}") + return + if packet["type"] != envelope.packet_type: + self._account_transport_reject(client, "GNS snapshot envelope family does not match packet type") + return + payload = envelope.payload + else: + try: + packet = decode_packet(payload) + except PacketCodecError: + packet = None + if packet is not None and is_snapshot_packet(packet["type"]): + self._account_transport_reject(client, "GNS snapshot missing required sequence envelope") + return + + try: + line = payload.decode("utf-8", errors="strict") except UnicodeDecodeError: - self.server._reject_packet(client, "Packet is not valid UTF-8") + self._account_transport_reject(client, "Packet is not valid UTF-8") return self.server._handle_line(client, line) self._purge_closed_clients() @@ -211,6 +269,9 @@ class GnsGameplayAdapter: with self._lock: self._clients.pop(connection_id, None) self._connected_monotonic.pop(connection_id, None) + stale_windows = [key for key in self._incoming_snapshot_windows if key[0] == connection_id] + for key in stale_windows: + self._incoming_snapshot_windows.pop(key, None) def _purge_closed_clients(self) -> None: with self._lock: @@ -312,4 +373,5 @@ class GnsGameplayAdapter: with self._lock: self._clients.clear() self._connected_monotonic.clear() + self._incoming_snapshot_windows.clear() self.transport.close()