GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Ack Cut 1c tests: a reconnect forgets what the old server sent; an Accept for an abandoned Connect does not connect; the cultmesh-py server test observes what the server sent

Browse source 
Two rules the first mutation pass found undefended in every runtime: a new
Connect on a session that had a peer clears the old peer's received window (a
restarted server's sequences overlap the old ones), and ending a session
abandons its pending Connect. The cultmesh-py server test now counts the
server's own data packets by sequence, so a peer replaced on a retransmitted
Connect is seen, and asserts a restarted client inherits no subscription.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 06:34:23 +02:00
commit 86c170fa0b
5 changed files with 201 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

113
packages/cultmesh-py/tests/test_rudp_server_generation.py
View file

@ -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__":

18
packages/cultnet-py/tests/test_rudp_connect_generation.py
View file

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

41
packages/cultnet-rs/tests/rudp_connect_generation.rs
View file

@ -26,6 +26,10 @@ fn session(initial_sequence: u32) -> CultNetRudpSession {
})
}
fn send(from: &mut CultNetRudpSession, text: &str) -> Result<CultNetRudpPacket> {
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]

23
packages/cultnet-ts/test/rudp-connect-generation.test.ts
View file

@ -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);

26
tests/GameCult.Networking.Tests/CultNetRudpConnectGenerationTests.cs
View file

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