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()