Test sequenced GNS gameplay snapshots

server/tests/test_gns_gameplay_adapter.py

@@ -6,6 +6,7 @@ from collections import deque
 import pytest
 
 from gns_gameplay_adapter import GnsConnectionAdapter, GnsGameplayAdapter
+from gns_snapshot_envelope import decode_snapshot, encode_snapshot
 from gns_transport import EventType, GnsEvent, RemoteEndpoint, SendResult
 from packet_codec import encode_packet
 from server_core import FalloutTogetherServer, PROTOCOL_VERSION
@@ -19,6 +20,7 @@ class FakeGnsTransport:
         self.sent = []
         self.send_results = deque()
         self.disconnected = []
+        self.incoming_sequences = {}
         self.closed = False
 
     def poll(self):
@@ -42,9 +44,15 @@ class FakeGnsTransport:
         self.closed = True
 
 
+def decoded_sent_packet(encoded):
+    envelope = decode_snapshot(encoded.payload)
+    payload = envelope.payload if envelope is not None else encoded.payload
+    return json.loads(payload)
+
+
 def packet_types_sent(transport: FakeGnsTransport, connection_id: int) -> list[str]:
     return [
-        json.loads(encoded.payload)["type"]
+        decoded_sent_packet(encoded)["type"]
         for target, encoded in transport.sent
         if target == connection_id
     ]
@@ -55,14 +63,23 @@ def queue_connect(transport: FakeGnsTransport, connection_id: int, port: int = 5
     transport.events.append(GnsEvent(EventType.CONNECTED, connection_id))
 
 
-def queue_packet(transport: FakeGnsTransport, connection_id: int, packet: dict):
+def queue_packet(
-    transport.events.append(
+    transport: FakeGnsTransport,
-        GnsEvent(
+    connection_id: int,
-            EventType.MESSAGE,
+    packet: dict,
-            connection_id,
+    *,
-            json.dumps(packet, separators=(",", ":")).encode(),
+    sequence: int | None = None,
-        )
+    raw_snapshot: bool = False,
-    )
+):
+    payload = json.dumps(packet, separators=(",", ":")).encode()
+    packet_type = packet.get("type")
+    if packet_type in {"transform", "npcState"} and not raw_snapshot:
+        key = (connection_id, packet_type)
+        if sequence is None:
+            sequence = transport.incoming_sequences.get(key, 0) + 1
+            transport.incoming_sequences[key] = sequence
+        payload = encode_snapshot(packet_type, payload, sequence)
+    transport.events.append(GnsEvent(EventType.MESSAGE, connection_id, payload))
 
 
 def complete_handshake(adapter: GnsGameplayAdapter, transport: FakeGnsTransport, connection_id: int):
@@ -74,6 +91,19 @@ def complete_handshake(adapter: GnsGameplayAdapter, transport: FakeGnsTransport,
     assert "sessionReady" in packet_types_sent(transport, connection_id)
 
 
+def transform_packet(x: float) -> dict:
+    return {
+        "type": "transform",
+        "x": x,
+        "y": 0.0,
+        "z": 0.0,
+        "angleZ": 0.0,
+        "cellId": "00000010",
+        "worldspaceId": "0000003C",
+        "movementType": "normal",
+    }
+
+
 def test_gns_adapter_reuses_v2_admission_and_server_owned_identity():
     server = FalloutTogetherServer(host="127.0.0.1", port=0)
     transport = FakeGnsTransport()
@@ -85,21 +115,9 @@ def test_gns_adapter_reuses_v2_admission_and_server_owned_identity():
     assert client.gameplay_active
     assert client.protocol_version == PROTOCOL_VERSION
 
-    queue_packet(
+    packet = transform_packet(1.0)
-        transport,
+    packet["playerId"] = 999999
-        101,
+    queue_packet(transport, 101, packet)
-        {
-            "type": "transform",
-            "playerId": 999999,
-            "x": 1.0,
-            "y": 2.0,
-            "z": 3.0,
-            "angleZ": 4.0,
-            "cellId": "00000010",
-            "worldspaceId": "0000003C",
-            "movementType": "normal",
-        },
-    )
     adapter.pump_once()
     assert client.last_transform is not None
     assert client.last_transform["playerId"] == client.player_id
@@ -114,44 +132,23 @@ def test_gns_adapter_preserves_reliable_and_snapshot_delivery_policy():
     complete_handshake(adapter, transport, 202)
 
     transport.sent.clear()
-    queue_packet(
+    queue_packet(transport, 202, transform_packet(10.0))
-        transport,
-        202,
-        {
-            "type": "transform",
-            "x": 10.0,
-            "y": 0.0,
-            "z": 0.0,
-            "angleZ": 0.0,
-            "cellId": "00000010",
-            "worldspaceId": "0000003C",
-            "movementType": "normal",
-        },
-    )
     adapter.pump_once()
-    queue_packet(
+    queue_packet(transport, 201, transform_packet(0.0))
-        transport,
-        201,
-        {
-            "type": "transform",
-            "x": 0.0,
-            "y": 0.0,
-            "z": 0.0,
-            "angleZ": 0.0,
-            "cellId": "00000010",
-            "worldspaceId": "0000003C",
-            "movementType": "normal",
-        },
-    )
     adapter.pump_once()
 
     transform_relays = [
         encoded
         for target, encoded in transport.sent
-        if target == 202 and json.loads(encoded.payload).get("type") == "transform"
+        if target == 202 and decoded_sent_packet(encoded).get("type") == "transform"
     ]
     assert transform_relays
-    assert transform_relays[-1].delivery is Delivery.UNRELIABLE_SEQUENCED
+    transform_relay = transform_relays[-1]
+    assert transform_relay.delivery is Delivery.UNRELIABLE_SEQUENCED
+    transform_envelope = decode_snapshot(transform_relay.payload)
+    assert transform_envelope is not None
+    assert transform_envelope.packet_type == "transform"
+    assert transform_envelope.sequence > 0
 
     queue_packet(
         transport,
@@ -162,10 +159,57 @@ def test_gns_adapter_preserves_reliable_and_snapshot_delivery_policy():
     player_state_relays = [
         encoded
         for target, encoded in transport.sent
-        if target == 202 and json.loads(encoded.payload).get("type") == "playerState"
+        if target == 202 and decoded_sent_packet(encoded).get("type") == "playerState"
     ]
     assert player_state_relays
-    assert player_state_relays[-1].delivery is Delivery.RELIABLE_ORDERED
+    player_state_relay = player_state_relays[-1]
+    assert player_state_relay.delivery is Delivery.RELIABLE_ORDERED
+    assert decode_snapshot(player_state_relay.payload) is None
+
+
+def test_gns_snapshot_reordering_and_duplicates_never_roll_state_backward():
+    server = FalloutTogetherServer(host="127.0.0.1", port=0)
+    transport = FakeGnsTransport()
+    adapter = GnsGameplayAdapter(server, transport)
+    complete_handshake(adapter, transport, 501)
+    client = adapter._client_for_connection(501)
+    assert client is not None
+
+    queue_packet(transport, 501, transform_packet(1.0), sequence=1)
+    adapter.pump_once()
+    assert client.last_transform is not None
+    assert client.last_transform["x"] == 1.0
+
+    queue_packet(transport, 501, transform_packet(3.0), sequence=3)
+    adapter.pump_once()
+    assert client.last_transform["x"] == 3.0
+
+    rejected_before = server.get_stats()["packetsRejected"]
+    queue_packet(transport, 501, transform_packet(2.0), sequence=2)
+    adapter.pump_once()
+    assert client.last_transform["x"] == 3.0
+    assert server.get_stats()["packetsRejected"] == rejected_before + 1
+
+    queue_packet(transport, 501, transform_packet(9.0), sequence=3)
+    adapter.pump_once()
+    assert client.last_transform["x"] == 3.0
+    assert server.get_stats()["packetsRejected"] == rejected_before + 2
+
+
+def test_gns_raw_unsequenced_snapshot_is_rejected_before_state_mutation():
+    server = FalloutTogetherServer(host="127.0.0.1", port=0)
+    transport = FakeGnsTransport()
+    adapter = GnsGameplayAdapter(server, transport)
+    complete_handshake(adapter, transport, 601)
+    client = adapter._client_for_connection(601)
+    assert client is not None
+    assert client.last_transform is None
+
+    rejected_before = server.get_stats()["packetsRejected"]
+    queue_packet(transport, 601, transform_packet(1.0), raw_snapshot=True)
+    adapter.pump_once()
+    assert client.last_transform is None
+    assert server.get_stats()["packetsRejected"] == rejected_before + 1
 
 
 def test_connection_adapter_may_drop_snapshots_but_never_reliable_messages_silently():
@@ -174,9 +218,13 @@ def test_connection_adapter_may_drop_snapshots_but_never_reliable_messages_silen
 
     transport.send_results.append(SendResult.DROPPED)
     connection.sendall(b'{"type":"transform","x":1}\n')
+    transform_wire = transport.sent[-1][1]
+    assert decode_snapshot(transform_wire.payload) is not None
 
     transport.send_results.append(SendResult.BACKPRESSURE)
     connection.sendall(b'{"type":"npcState","npcs":[]}\n')
+    npc_wire = transport.sent[-1][1]
+    assert decode_snapshot(npc_wire.payload) is not None
 
     transport.send_results.append(SendResult.DROPPED)
     with pytest.raises(OSError):
@@ -198,9 +246,9 @@ def test_gns_adapter_enforces_existing_ban_policy(tmp_path):
     assert adapter._client_for_connection(301) is None
     assert transport.disconnected
     session_ended = [
-        json.loads(encoded.payload)
+        decoded_sent_packet(encoded)
         for target, encoded in transport.sent
-        if target == 301 and json.loads(encoded.payload).get("type") == "sessionEnded"
+        if target == 301 and decoded_sent_packet(encoded).get("type") == "sessionEnded"
     ]
     assert session_ended
     assert session_ended[-1]["code"] == "banned"
@@ -220,9 +268,9 @@ def test_gns_adapter_ends_session_on_native_oversize_event():
     assert adapter._client_for_connection(401) is None
     assert transport.disconnected
     ended_packets = [
-        json.loads(encoded.payload)
+        decoded_sent_packet(encoded)
         for target, encoded in transport.sent
-        if target == 401 and json.loads(encoded.payload).get("type") == "sessionEnded"
+        if target == 401 and decoded_sent_packet(encoded).get("type") == "sessionEnded"
     ]
     assert ended_packets
     assert ended_packets[-1]["code"] == "packet_too_large"