diff --git a/packages/cultmesh-py/tests/test_rudp_server_generation.py b/packages/cultmesh-py/tests/test_rudp_server_generation.py index 4f8e11b2..259e648c 100644 --- a/packages/cultmesh-py/tests/test_rudp_server_generation.py +++ b/packages/cultmesh-py/tests/test_rudp_server_generation.py @@ -1,35 +1,62 @@ """The CultMesh local server's RUDP loop asks each peer's session whether a Connect repeats. A retransmitted Connect keeps the peer; a restarted client on the same -address is a new one.""" +address is a new one, with none of the old one's subscriptions.""" from __future__ import annotations import socket import time import unittest +from dataclasses import dataclass import msgpack +from cultcache_py import define_database_entry_type from cultmesh_py import create_node from cultmesh_py.server import CultMeshLocalServer from cultnet_py import ( CultNetRudpSendOptions, CultNetRudpSession, CultNetRudpSessionOptions, + document_put_raw, ) -from cultnet_py.transport import decode_rudp_packet, encode_rudp_packet +from cultnet_py.transport import CultNetRudpPacketType, decode_rudp_packet, encode_rudp_packet CONNECTION_ID = 0x10203091 -HELLO = msgpack.packb({"schemaVersion": "cultnet.hello.v0", "runtimeId": "probe", "runtimeKind": "test"}, use_bin_type=True) +SCHEMA_ID = "mesh.rudp_generation_note.v1" +RELIABLE = CultNetRudpSendOptions(reliable=True, ordered=True) + + +@dataclass +class Note: + body: str + + +NOTE = define_database_entry_type( + "mesh.rudp_generation_note", + [("body", 0)], + cls=Note, + schema_id=SCHEMA_ID, + schema_name="mesh.rudp_generation_note", + schema_version=SCHEMA_ID, +) + + +def wire(message: dict) -> bytes: + return msgpack.packb(message, use_bin_type=True) class RudpServerGenerationTests(unittest.TestCase): def setUp(self) -> None: - self.server = CultMeshLocalServer(node=create_node(runtime_id="rudp-generation"), rudp_connection_id=CONNECTION_ID) + node = create_node(runtime_id="rudp-generation") + node.database.register_document(NOTE) + self.server = CultMeshLocalServer(node=node, rudp_connection_id=CONNECTION_ID) self.server.start() self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock.bind(("127.0.0.1", 0)) self.sock.settimeout(0.02) self.target = ("127.0.0.1", self.server.port) - self.responses = 0 + # Every schema data packet the server sent, by sequence, whether or not the receiving + # session delivered it: a new server-side session numbers its packets afresh. + self.sent: dict[int, str] = {} def tearDown(self) -> None: self.sock.close() @@ -44,42 +71,88 @@ class RudpServerGenerationTests(unittest.TestCase): if until(): return True try: - wire, _ = self.sock.recvfrom(65535) + data, _ = self.sock.recvfrom(65535) except TimeoutError: continue - packet = decode_rudp_packet(wire) - result = session.receive(packet, int(time.monotonic() * 1000)) - if packet.reliable or packet.packet_type.value == "accept": + packet = decode_rudp_packet(data) + session.receive(packet, int(time.monotonic() * 1000)) + if packet.reliable or packet.packet_type == CultNetRudpPacketType.ACCEPT: self.to_server(session.create_ack_for_received(packet.sequence)) - for frame in result.delivered: - if msgpack.unpackb(frame.payload, raw=False).get("schemaVersion") == "cultnet.hello.v0": - self.responses += 1 + if packet.packet_type == CultNetRudpPacketType.DATA and packet.channel_id == "schema": + self.sent[packet.sequence] = msgpack.unpackb(packet.payload, raw=False).get("schemaVersion") return until() + def count(self, schema_version: str) -> int: + return sum(1 for value in self.sent.values() if value == schema_version) + def new_session(self, initial_sequence: int) -> CultNetRudpSession: return CultNetRudpSession(CultNetRudpSessionOptions(connection_id=CONNECTION_ID, initial_sequence=initial_sequence)) - def test_a_retransmitted_connect_keeps_the_peer_and_a_restarted_client_is_admitted(self) -> None: + def test_a_retransmitted_connect_keeps_the_peer_and_a_restarted_client_is_a_new_one(self) -> None: first = self.new_session(50) connect = first.create_connect(0, b"same") self.to_server(connect) self.assertTrue(self.pump(first, lambda: first.connected), "the first Accept") - hello = first.send("schema", HELLO, CultNetRudpSendOptions(reliable=True, ordered=True)) + hello = first.send( + "schema", wire({"schemaVersion": "cultnet.hello.v0", "runtimeId": "probe", "runtimeKind": "test"}), RELIABLE + ) self.to_server(hello) - self.assertTrue(self.pump(first, lambda: self.responses == 1), "the first answer") + self.assertTrue(self.pump(first, lambda: self.count("cultnet.hello.v0") == 1), "the first answer") + self.to_server( + first.send( + "schema", + wire({ + "schemaVersion": "cultnet.database_subscribe.v0", + "messageId": "subscribe", + "subscriptionId": "old-incarnation", + "schemaIds": [SCHEMA_ID], + "includeSnapshot": False, + }), + RELIABLE, + ) + ) + self.pump(first, lambda: False, 0.2) + first_put = document_put_raw( + message_id="first-put", + key="note:a", + schema_id=SCHEMA_ID, + stored_at="2026-09-30T00:00:00Z", + payload=NOTE.encode_payload(Note("first")), + ) + self.to_server(first.send("schema", first_put.to_bytes(), RELIABLE)) + self.assertTrue( + self.pump(first, lambda: self.count("cultnet.database_change_raw.v0") == 1), + "the subscription did not notify its own connection", + ) - # A retransmitted Connect is a repeat: the data already delivered is not delivered, or answered, again. + # A retransmitted Connect repeats: the data already delivered is not delivered, or answered, again. self.to_server(connect) self.to_server(hello) self.pump(first, lambda: False, 0.4) - self.assertEqual(self.responses, 1, "the retransmitted Connect reset the peer") + self.assertEqual(self.count("cultnet.hello.v0"), 1, "the retransmitted Connect reset the peer") - # The same client restarted on the same address, connection id and payload. + # The same client restarted on the same address, connection id and payload: a new peer. second = self.new_session(7) self.to_server(second.create_connect(0, b"same")) self.assertTrue(self.pump(second, lambda: second.connected), "the restarted client's Accept") - self.to_server(second.send("schema", HELLO, CultNetRudpSendOptions(reliable=True, ordered=True))) - self.assertTrue(self.pump(second, lambda: self.responses == 2), "the restarted client's answer") + self.to_server( + second.send( + "schema", wire({"schemaVersion": "cultnet.hello.v0", "runtimeId": "probe", "runtimeKind": "test"}), RELIABLE + ) + ) + self.assertTrue(self.pump(second, lambda: self.count("cultnet.hello.v0") == 2), "the restarted client's answer") + + # It inherited none of the old incarnation's subscriptions. + put = document_put_raw( + message_id="second-put", + key="note:a", + schema_id=SCHEMA_ID, + stored_at="2026-09-30T00:00:01Z", + payload=NOTE.encode_payload(Note("second")), + ) + self.to_server(second.send("schema", put.to_bytes(), RELIABLE)) + self.pump(second, lambda: False, 0.4) + self.assertEqual(self.count("cultnet.database_change_raw.v0"), 1, "the new peer inherited a subscription") if __name__ == "__main__": diff --git a/packages/cultnet-py/tests/test_rudp_connect_generation.py b/packages/cultnet-py/tests/test_rudp_connect_generation.py index b1e68f3e..77c3556f 100644 --- a/packages/cultnet-py/tests/test_rudp_connect_generation.py +++ b/packages/cultnet-py/tests/test_rudp_connect_generation.py @@ -104,6 +104,24 @@ class CultNetRudpConnectGenerationTests(unittest.TestCase): client.receive(accept, 5) self.assertFalse(client.connected, "a late Accept revived an ended session") + def test_a_reconnect_forgets_what_the_old_server_sent(self) -> None: + client, old_server = session(1), session(500) + handshake(client, old_server) + for name in ("d1", "d2", "d3"): + self.assertEqual(names(client.receive(send(old_server, name), 1)), [name]) + + new_server = session(501) + client.receive(new_server.accept_connect(client.create_connect(2), 2), 2) + self.assertEqual(names(client.receive(send(new_server, "fresh"), 3)), ["fresh"]) + + def test_an_accept_for_an_abandoned_connect_does_not_connect_the_session(self) -> None: + client, server = session(1), session(500) + accept = server.accept_connect(client.create_connect(0), 0) + client.create_disconnect(b"never mind") + + client.receive(accept, 1) + self.assertFalse(client.connected, "an Accept revived an abandoned Connect") + def test_a_duplicate_accept_does_not_reseed_the_watermark(self) -> None: client, server = session(1), session(500) _, accept = handshake(client, server) diff --git a/packages/cultnet-rs/tests/rudp_connect_generation.rs b/packages/cultnet-rs/tests/rudp_connect_generation.rs index 83795d67..21a81d48 100644 --- a/packages/cultnet-rs/tests/rudp_connect_generation.rs +++ b/packages/cultnet-rs/tests/rudp_connect_generation.rs @@ -26,6 +26,10 @@ fn session(initial_sequence: u32) -> CultNetRudpSession { }) } +fn send(from: &mut CultNetRudpSession, text: &str) -> Result { + from.send("schema", text.as_bytes().to_vec(), ordered()) +} + fn ordered() -> CultNetRudpSendOptions { CultNetRudpSendOptions { reliable: true, @@ -174,6 +178,43 @@ fn an_accept_after_a_local_disconnect_does_not_reconnect_the_session() -> Result Ok(()) } +/// A client that reconnects to a restarted server is not silenced by what it +/// received from the old one: the new server's sequences may overlap the old +/// ones, and none of them is a duplicate. +#[test] +fn a_reconnect_forgets_what_the_old_server_sent() -> Result<()> { + let mut client = session(1); + let mut old_server = session(500); + handshake(&mut client, &mut old_server)?; + for name in ["d1", "d2", "d3"] { + let packet = send(&mut old_server, name)?; + assert_eq!(names(&client.receive(&packet, 1)?.delivered), [name]); + } + + let mut new_server = session(501); + let connect = client.create_connect(2, Vec::new())?; + let accept = new_server.accept_connect(&connect, 2, Vec::new())?; + client.receive(&accept, 2)?; + let fresh = send(&mut new_server, "fresh")?; + assert_eq!(names(&client.receive(&fresh, 3)?.delivered), ["fresh"]); + Ok(()) +} + +/// An Accept for a Connect that was abandoned before it was answered does not +/// connect the session. +#[test] +fn an_accept_for_an_abandoned_connect_does_not_connect_the_session() -> Result<()> { + let mut client = session(1); + let mut server = session(500); + let connect = client.create_connect(0, Vec::new())?; + let accept = server.accept_connect(&connect, 0, Vec::new())?; + client.create_disconnect(b"never mind".to_vec()); + + client.receive(&accept, 1)?; + assert!(!client.connected(), "an Accept revived an abandoned Connect"); + Ok(()) +} + /// An Accept counts once: a duplicate of the one honoured does not seed the /// watermark again. #[test] diff --git a/packages/cultnet-ts/test/rudp-connect-generation.test.ts b/packages/cultnet-ts/test/rudp-connect-generation.test.ts index 6bea717a..8f72beca 100644 --- a/packages/cultnet-ts/test/rudp-connect-generation.test.ts +++ b/packages/cultnet-ts/test/rudp-connect-generation.test.ts @@ -94,6 +94,29 @@ test("an Accept after a local disconnect does not reconnect the session", () => assert.equal(client.connected, false, "a late Accept revived an ended session"); }); +test("a reconnect forgets what the old server sent", () => { + const client = session(1); + const oldServer = session(500); + handshake(client, oldServer); + for (const name of ["d1", "d2", "d3"]) { + assert.deepEqual(names(client.receive(send(oldServer, name), 1)), [name]); + } + + const newServer = session(501); + client.receive(newServer.acceptConnect(client.createConnect(2), 2), 2); + assert.deepEqual(names(client.receive(send(newServer, "fresh"), 3)), ["fresh"]); +}); + +test("an Accept for an abandoned Connect does not connect the session", () => { + const client = session(1); + const server = session(500); + const accept = server.acceptConnect(client.createConnect(0), 0); + client.createDisconnect(enc("never mind")); + + client.receive(accept, 1); + assert.equal(client.connected, false, "an Accept revived an abandoned Connect"); +}); + test("a duplicate Accept does not reseed the watermark", () => { const client = session(1); const server = session(500); diff --git a/tests/GameCult.Networking.Tests/CultNetRudpConnectGenerationTests.cs b/tests/GameCult.Networking.Tests/CultNetRudpConnectGenerationTests.cs index 98ae028d..fb436af4 100644 --- a/tests/GameCult.Networking.Tests/CultNetRudpConnectGenerationTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetRudpConnectGenerationTests.cs @@ -152,6 +152,32 @@ namespace GameCult.Networking.Tests Assert.That(client.Connected, Is.False, "a late Accept revived an ended session"); } + [Test] + public void AReconnectForgetsWhatTheOldServerSent() + { + var client = Session(1); + var oldServer = Session(500); + Handshake(client, oldServer); + foreach (var name in new[] { "d1", "d2", "d3" }) + Assert.That(Names(client.Receive(Send(oldServer, name), 1)), Is.EqualTo(new[] { name })); + + var newServer = Session(501); + client.Receive(newServer.AcceptConnect(client.CreateConnect(2), 2), 2); + Assert.That(Names(client.Receive(Send(newServer, "fresh"), 3)), Is.EqualTo(new[] { "fresh" })); + } + + [Test] + public void AnAcceptForAnAbandonedConnectDoesNotConnectTheSession() + { + var client = Session(1); + var server = Session(500); + var accept = server.AcceptConnect(client.CreateConnect(0), 0); + client.CreateDisconnect(Encoding.UTF8.GetBytes("never mind")); + + client.Receive(accept, 1); + Assert.That(client.Connected, Is.False, "an Accept revived an abandoned Connect"); + } + [Test] public void ADuplicateAcceptDoesNotReseedTheWatermark() {