diff --git a/packages/cultnet-py/tests/test_cultnet.py b/packages/cultnet-py/tests/test_cultnet.py
index dbe6faa9..aebcfa10 100644
--- a/packages/cultnet-py/tests/test_cultnet.py
+++ b/packages/cultnet-py/tests/test_cultnet.py
@@ -711,7 +711,7 @@ class CultNetTests(unittest.TestCase):
server = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=103, initial_sequence=100))
connect = client.create_connect(0, b"join")
accept = server.accept_connect(connect, 0)
- reply = server.answer_repeated_connect(replace(connect, ack=accept.sequence), 1)
+ reply = server.accept_connect(replace(connect, ack=accept.sequence), 1)
self.assertEqual(reply.packet_type, CultNetRudpPacketType.ACK)
self.assertEqual(server.outstanding_reliable_packet_count, 0)
@@ -1150,6 +1150,78 @@ class CultNetTests(unittest.TestCase):
peer_socket.close()
server.close()
+ def test_cultnet_rudp_server_mode_admits_a_restarted_client_and_keeps_its_session_on_a_retransmit(self) -> None:
+ server, _, peer_socket, _, to_server, drain_peer = self._server_mode_pair(0x10203058)
+ connection_id = 0x10203058
+ reliable = CultNetRudpSendOptions(reliable=True, ordered=True)
+ try:
+ first = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id, initial_sequence=50))
+ connect = first.create_connect(0, b"same")
+ to_server(connect)
+ server.receive_once()
+ first.receive([p for p in drain_peer() if p.packet_type == CultNetRudpPacketType.ACCEPT][0], 0)
+ sent = [first.send("schema", f"c{index}".encode(), reliable) for index in range(3)]
+ delivered = []
+ for packet in sent:
+ to_server(packet)
+ delivered.append(server.receive_once().payload)
+ self.assertEqual(delivered, [b"c0", b"c1", b"c2"])
+
+ # A retransmitted Connect is a repeat: what was delivered stays delivered once.
+ to_server(connect)
+ server.receive_once()
+ to_server(sent[0])
+ self.assertIsNone(server.receive_once(), "the retransmitted Connect reset the session")
+
+ # The same client restarted on the same address, connection id and payload.
+ drain_peer()
+ second = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id, initial_sequence=7))
+ to_server(second.create_connect(0, b"same"))
+ server.receive_once()
+ second.receive([p for p in drain_peer() if p.packet_type == CultNetRudpPacketType.ACCEPT][0], 0)
+ to_server(second.send("schema", b"after restart", reliable))
+ self.assertEqual(server.receive_once().payload, b"after restart")
+ finally:
+ peer_socket.close()
+ server.close()
+
+ def test_cultnet_rudp_server_mode_admits_a_connect_from_a_new_endpoint(self) -> None:
+ server, server_socket, peer_socket, peer, to_server, drain_peer = self._server_mode_pair(0x10203059)
+ other_socket = bind_udp_socket()
+ try:
+ to_server(peer.create_connect(0, b"a"))
+ server.receive_once()
+ drain_peer()
+
+ other = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=0x10203059, initial_sequence=9))
+ other_socket.sendto(encode_rudp_packet(other.create_connect(0, b"b")), server_socket.getsockname())
+ server.receive_once()
+ wire, _ = other_socket.recvfrom(65535)
+ other.receive(decode_rudp_packet(wire), 0)
+ self.assertTrue(other.connected, "the new endpoint was not admitted")
+ other_socket.sendto(
+ encode_rudp_packet(other.send("schema", b"from b", CultNetRudpSendOptions(reliable=True, ordered=True))),
+ server_socket.getsockname(),
+ )
+ self.assertEqual(server.receive_once().payload, b"from b")
+ finally:
+ other_socket.close()
+ peer_socket.close()
+ server.close()
+
+ def test_cultnet_rudp_flush_after_a_timeout_names_the_timeout(self) -> None:
+ server, _, peer_socket, peer, to_server, drain_peer = self._server_mode_pair(0x1020305A)
+ try:
+ to_server(peer.create_connect(0, b"join"))
+ server.receive_once()
+ server.send("schema", b"unanswered")
+ self.assertTrue(server.check_timeout(0, now_ms=10**12))
+ with self.assertRaisesRegex(ConnectionError, "session timed out"):
+ server.flush_reliable(0.1)
+ finally:
+ peer_socket.close()
+ server.close()
+
def test_cultnet_rudp_session_rejects_limits_that_cannot_admit_a_connect(self) -> None:
with self.assertRaises(ValueError):
CultNetRudpSession(CultNetRudpSessionOptions(connection_id=1, initial_sequence=0xFFFFFFFF))
diff --git a/packages/cultnet-py/tests/test_rudp_connect_generation.py b/packages/cultnet-py/tests/test_rudp_connect_generation.py
new file mode 100644
index 00000000..b1e68f3e
--- /dev/null
+++ b/packages/cultnet-py/tests/test_rudp_connect_generation.py
@@ -0,0 +1,178 @@
+"""A Connect's sequence says whether it repeats the one a session accepted, and
+the handshake alone seeds the watermark that orders delivery."""
+from __future__ import annotations
+
+import unittest
+from dataclasses import replace
+
+from cultnet_py import random_initial_sequence
+from cultnet_py.transport import (
+ CultNetRudpPacketType,
+ CultNetRudpSendOptions,
+ CultNetRudpSession,
+ CultNetRudpSessionOptions,
+ CultNetRudpSocketMode,
+ CultNetRudpSocketTransportOptions,
+)
+
+CONNECTION_ID = 0x10203090
+
+
+def session(initial_sequence: int) -> CultNetRudpSession:
+ return CultNetRudpSession(CultNetRudpSessionOptions(connection_id=CONNECTION_ID, initial_sequence=initial_sequence))
+
+
+def send(from_session: CultNetRudpSession, text: str, *, ordered: bool = True, channel: str = "schema"):
+ return from_session.send(channel, text.encode(), CultNetRudpSendOptions(reliable=True, ordered=ordered))
+
+
+def names(result) -> list[str]:
+ return [frame.payload.decode() for frame in result.delivered]
+
+
+def handshake(client: CultNetRudpSession, server: CultNetRudpSession):
+ connect = client.create_connect(0)
+ accept = server.accept_connect(connect, 0)
+ client.receive(accept, 0)
+ assert client.connected and server.connected
+ return connect, accept
+
+
+class CultNetRudpConnectGenerationTests(unittest.TestCase):
+ def test_a_reconnect_on_the_same_session_is_not_stranded_by_a_write_the_ended_session_owed(self) -> None:
+ client, server = session(1), session(500)
+ handshake(client, server)
+ send(server, "lost")
+ server.create_disconnect(b"gone")
+
+ client.receive(server.accept_connect(client.create_connect(1), 1), 1)
+ self.assertEqual(names(client.receive(send(server, "after"), 2)), ["after"])
+
+ def test_a_server_that_accepts_a_new_client_forgets_the_old_clients_sequences(self) -> None:
+ old_client, server = session(50), session(500)
+ handshake(old_client, server)
+ for name in ("c1", "c2", "c3"):
+ self.assertEqual(names(server.receive(send(old_client, name), 1)), [name])
+ server.create_disconnect(b"restart")
+
+ new_client = session(2)
+ new_client.receive(server.accept_connect(new_client.create_connect(2), 2), 2)
+ self.assertEqual(names(server.receive(send(new_client, "fresh"), 3)), ["fresh"])
+
+ def test_a_retransmitted_connect_after_data_has_flowed_does_not_reset_the_session(self) -> None:
+ client, server = session(1), session(500)
+ connect, _ = handshake(client, server)
+ frame = send(client, "once")
+ self.assertEqual(names(server.receive(frame, 1)), ["once"])
+
+ self.assertTrue(server.connect_repeats(connect))
+ self.assertEqual(server.accept_connect(connect, 2).packet_type, CultNetRudpPacketType.ACK)
+ self.assertEqual(server.outstanding_reliable_packet_count, 0, "a repeat queued something")
+ self.assertEqual(
+ names(server.receive(frame, 3)), [], "a retransmitted Connect made the server forget what it had delivered"
+ )
+
+ def test_a_connect_with_another_sequence_is_not_a_repeat_whatever_it_carries(self) -> None:
+ server = session(500)
+ connect = session(1).create_connect(0, b"same payload")
+ server.accept_connect(connect, 0)
+ self.assertTrue(server.connect_repeats(connect))
+
+ self.assertFalse(server.connect_repeats(session(9).create_connect(0, b"same payload")))
+ self.assertFalse(server.connect_repeats(replace(connect, packet_type=CultNetRudpPacketType.PING)))
+ server.create_disconnect()
+ self.assertFalse(server.connect_repeats(connect), "an ended session repeats nothing")
+
+ def test_an_accept_that_does_not_name_the_pending_connect_is_ignored(self) -> None:
+ client, server = session(10), session(500)
+ connect = client.create_connect(0)
+ accept = server.accept_connect(connect, 0)
+
+ client.receive(replace(accept, ack=connect.sequence + 50, ack_mask=0), 1)
+ self.assertFalse(client.connected, "a stale Accept connected the session")
+ self.assertEqual(client.pending_reliable_sequences, (connect.sequence,))
+
+ client.receive(replace(accept, ack=connect.sequence + 3, ack_mask=1 << 2), 2)
+ self.assertTrue(client.connected, "the mask names the Connect")
+
+ def test_an_accept_after_a_local_disconnect_does_not_reconnect_the_session(self) -> None:
+ client, server = session(1), session(500)
+ _, accept = handshake(client, server)
+ client.create_disconnect(b"bye")
+ self.assertFalse(client.connected)
+
+ client.receive(accept, 5)
+ self.assertFalse(client.connected, "a late Accept revived an ended session")
+
+ def test_a_duplicate_accept_does_not_reseed_the_watermark(self) -> None:
+ client, server = session(1), session(500)
+ _, accept = handshake(client, server)
+ s1, s2 = send(server, "s1"), send(server, "s2")
+ client.receive(replace(accept, sequence=s2.sequence), 1)
+ self.assertEqual(names(client.receive(s2, 2)), [], "s1 is still missing")
+ self.assertEqual(names(client.receive(s1, 3)), ["s1", "s2"])
+
+ def test_reliable_data_before_the_accept_is_neither_delivered_nor_acknowledged(self) -> None:
+ client, server = session(1), session(500)
+ accept = server.accept_connect(client.create_connect(0), 0)
+ a, b = send(server, "A"), send(server, "B")
+
+ for early in (b, a):
+ self.assertEqual(names(client.receive(early, 1)), [])
+ ack = client.create_ack_for_received(early.sequence)
+ self.assertEqual((ack.ack, ack.ack_mask), (0, 0), "an unhandled packet was acknowledged")
+ client.receive(accept, 2)
+ self.assertTrue(client.connected)
+ self.assertEqual(server.outstanding_reliable_packet_count, 3, "nothing was acknowledged")
+
+ self.assertEqual(names(client.receive(a, 3)), ["A"])
+ self.assertEqual(names(client.receive(b, 3)), ["B"])
+
+ def test_ordered_frames_after_the_accept_are_delivered_in_order_whatever_arrives_first(self) -> None:
+ client, server = session(1), session(500)
+ handshake(client, server)
+ a, b = send(server, "A"), send(server, "B")
+ self.assertEqual(names(client.receive(b, 1)), [])
+ self.assertEqual(names(client.receive(a, 2)), ["A", "B"])
+
+ def test_an_unordered_reliable_frame_is_delivered_on_receipt_never_held_behind_a_gap(self) -> None:
+ client, server = session(1), session(500)
+ handshake(client, server)
+ send(server, "lost")
+ unordered = send(server, "now", ordered=False, channel="media")
+ ordered = send(server, "later")
+ self.assertEqual(names(client.receive(unordered, 1)), ["now"])
+ self.assertEqual(names(client.receive(ordered, 2)), [], "the ordered frame waits for the gap")
+
+ def test_a_peer_disconnect_does_not_forget_what_the_peer_sent(self) -> None:
+ client, server = session(1), session(500)
+ handshake(client, server)
+ s1, s2 = send(client, "s1"), send(client, "s2")
+ self.assertEqual(names(server.receive(s2, 1)), [], "s1 is missing, so s2 is held")
+
+ server.receive(client.create_disconnect(b"bye"), 2)
+ self.assertEqual(names(server.receive(s1, 3)), ["s1", "s2"], "the Disconnect forgot the held frame")
+ self.assertEqual(names(server.receive(s2, 4)), [], "the Disconnect forgot what was received")
+
+ def test_default_initial_sequences_are_drawn_at_random_from_one_to_two_to_the_thirty_first(self) -> None:
+ def assert_random(label: str, draw) -> None:
+ draws = {draw() for _ in range(64)}
+ self.assertGreater(len(draws), 32, f"{label} does not draw at random")
+ self.assertTrue(all(1 <= value < 2**31 for value in draws), f"{label} left [1, 2^31)")
+
+ assert_random("random_initial_sequence", random_initial_sequence)
+ assert_random("session options", lambda: CultNetRudpSessionOptions(connection_id=1).initial_sequence)
+ assert_random(
+ "transport options",
+ lambda: CultNetRudpSocketTransportOptions(
+ runtime_id="r", socket=None, mode=CultNetRudpSocketMode.CLIENT, connection_id=1
+ ).initial_sequence,
+ )
+ assert_random(
+ "a default session's first Connect",
+ lambda: CultNetRudpSession(CultNetRudpSessionOptions(connection_id=1)).create_connect(0).sequence,
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/GameCult.Networking.Tests/CultNetRudpConnectGenerationTests.cs b/tests/GameCult.Networking.Tests/CultNetRudpConnectGenerationTests.cs
new file mode 100644
index 00000000..98ae028d
--- /dev/null
+++ b/tests/GameCult.Networking.Tests/CultNetRudpConnectGenerationTests.cs
@@ -0,0 +1,375 @@
+#nullable enable
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using GameCult.Mesh;
+using NUnit.Framework;
+
+namespace GameCult.Networking.Tests
+{
+ ///
+ /// A Connect's sequence says whether it repeats the one a session accepted, and the handshake
+ /// alone seeds the watermark that orders delivery.
+ ///
+ public sealed class CultNetRudpConnectGenerationTests
+ {
+ private const uint ConnectionId = 0x10203090;
+ private static readonly CultNetRudpSendOptions Ordered = new CultNetRudpSendOptions { Reliable = true, Ordered = true };
+
+ private static CultNetRudpSession Session(uint initialSequence) =>
+ new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId, InitialSequence = initialSequence });
+
+ private static CultNetRudpPacket Send(CultNetRudpSession session, string text, bool ordered = true, string channel = "schema") =>
+ session.Send(channel, Encoding.UTF8.GetBytes(text), new CultNetRudpSendOptions { Reliable = true, Ordered = ordered });
+
+ private static string[] Names(CultNetRudpReceiveResult result) =>
+ result.Delivered.Select(frame => Encoding.UTF8.GetString(frame.Payload)).ToArray();
+
+ private static (CultNetRudpPacket Connect, CultNetRudpPacket Accept) Handshake(CultNetRudpSession client, CultNetRudpSession server)
+ {
+ var connect = client.CreateConnect(0);
+ var accept = server.AcceptConnect(connect, 0);
+ client.Receive(accept, 0);
+ Assert.That(client.Connected && server.Connected, Is.True);
+ return (connect, accept);
+ }
+
+ private static Socket Bind()
+ {
+ var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
+ socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
+ socket.ReceiveTimeout = 20;
+ return socket;
+ }
+
+ private static List Drain(Socket socket)
+ {
+ var packets = new List();
+ var buffer = new byte[65535];
+ EndPoint from = new IPEndPoint(IPAddress.Any, 0);
+ while (socket.Poll(20_000, SelectMode.SelectRead))
+ {
+ var received = socket.ReceiveFrom(buffer, ref from);
+ packets.Add(CultNetRudpPacketCodec.Decode(buffer.AsSpan(0, received).ToArray()));
+ }
+ return packets;
+ }
+
+ [Test]
+ public void AReconnectOnTheSameSessionIsNotStrandedByAWriteTheEndedSessionOwed()
+ {
+ var client = Session(1);
+ var server = Session(500);
+ Handshake(client, server);
+ Send(server, "lost");
+ server.CreateDisconnect(Encoding.UTF8.GetBytes("gone"));
+
+ client.Receive(server.AcceptConnect(client.CreateConnect(1), 1), 1);
+ Assert.That(Names(client.Receive(Send(server, "after"), 2)), Is.EqualTo(new[] { "after" }));
+ }
+
+ [Test]
+ public void AServerThatAcceptsANewClientForgetsTheOldClientsSequences()
+ {
+ var oldClient = Session(50);
+ var server = Session(500);
+ Handshake(oldClient, server);
+ foreach (var name in new[] { "c1", "c2", "c3" })
+ Assert.That(Names(server.Receive(Send(oldClient, name), 1)), Is.EqualTo(new[] { name }));
+ server.CreateDisconnect(Encoding.UTF8.GetBytes("restart"));
+
+ var newClient = Session(2);
+ newClient.Receive(server.AcceptConnect(newClient.CreateConnect(2), 2), 2);
+ Assert.That(Names(server.Receive(Send(newClient, "fresh"), 3)), Is.EqualTo(new[] { "fresh" }));
+ }
+
+ [Test]
+ public void ARetransmittedConnectAfterDataHasFlowedDoesNotResetTheSession()
+ {
+ var client = Session(1);
+ var server = Session(500);
+ var (connect, _) = Handshake(client, server);
+ var frame = Send(client, "once");
+ Assert.That(Names(server.Receive(frame, 1)), Is.EqualTo(new[] { "once" }));
+
+ Assert.That(server.ConnectRepeats(connect), Is.True);
+ Assert.That(server.AcceptConnect(connect, 2).PacketType, Is.EqualTo(CultNetRudpPacketType.Ack), "the Accept was already acknowledged");
+ Assert.That(server.OutstandingReliablePacketCount, Is.EqualTo(0), "a repeat queued something");
+ Assert.That(server.Receive(frame, 3).Delivered, Is.Empty, "a retransmitted Connect made the server forget what it had delivered");
+ }
+
+ [Test]
+ public void AConnectWithAnotherSequenceIsNotARepeatWhateverItCarries()
+ {
+ var server = Session(500);
+ var connect = Session(1).CreateConnect(0, Encoding.UTF8.GetBytes("same payload"));
+ server.AcceptConnect(connect, 0);
+ Assert.That(server.ConnectRepeats(connect), Is.True);
+
+ Assert.That(server.ConnectRepeats(Session(9).CreateConnect(0, Encoding.UTF8.GetBytes("same payload"))), Is.False);
+ var notAConnect = CultNetRudpPacketCodec.Decode(CultNetRudpPacketCodec.Encode(connect));
+ notAConnect.PacketType = CultNetRudpPacketType.Ping;
+ Assert.That(server.ConnectRepeats(notAConnect), Is.False);
+ server.CreateDisconnect();
+ Assert.That(server.ConnectRepeats(connect), Is.False, "an ended session repeats nothing");
+ }
+
+ [Test]
+ public void AnAcceptThatDoesNotNameThePendingConnectIsIgnored()
+ {
+ var client = Session(10);
+ var server = Session(500);
+ var connect = client.CreateConnect(0);
+ var accept = server.AcceptConnect(connect, 0);
+
+ var stale = CultNetRudpPacketCodec.Decode(CultNetRudpPacketCodec.Encode(accept));
+ stale.Ack = connect.Sequence + 50;
+ stale.AckMask = 0;
+ client.Receive(stale, 1);
+ Assert.That(client.Connected, Is.False, "a stale Accept connected the session");
+ Assert.That(client.PendingReliableSequences, Is.EqualTo(new[] { connect.Sequence }));
+
+ var named = CultNetRudpPacketCodec.Decode(CultNetRudpPacketCodec.Encode(accept));
+ named.Ack = connect.Sequence + 3;
+ named.AckMask = 1u << 2;
+ client.Receive(named, 2);
+ Assert.That(client.Connected, Is.True, "the mask names the Connect");
+ }
+
+ [Test]
+ public void AnAcceptAfterALocalDisconnectDoesNotReconnectTheSession()
+ {
+ var client = Session(1);
+ var server = Session(500);
+ var (_, accept) = Handshake(client, server);
+ client.CreateDisconnect(Encoding.UTF8.GetBytes("bye"));
+ Assert.That(client.Connected, Is.False);
+
+ client.Receive(accept, 5);
+ Assert.That(client.Connected, Is.False, "a late Accept revived an ended session");
+ }
+
+ [Test]
+ public void ADuplicateAcceptDoesNotReseedTheWatermark()
+ {
+ var client = Session(1);
+ var server = Session(500);
+ var (_, accept) = Handshake(client, server);
+ var s1 = Send(server, "s1");
+ var s2 = Send(server, "s2");
+ var forged = CultNetRudpPacketCodec.Decode(CultNetRudpPacketCodec.Encode(accept));
+ forged.Sequence = s2.Sequence;
+ client.Receive(forged, 1);
+ Assert.That(client.Receive(s2, 2).Delivered, Is.Empty, "s1 is still missing");
+ Assert.That(Names(client.Receive(s1, 3)), Is.EqualTo(new[] { "s1", "s2" }));
+ }
+
+ [Test]
+ public void ReliableDataBeforeTheAcceptIsNeitherDeliveredNorAcknowledged()
+ {
+ var client = Session(1);
+ var server = Session(500);
+ var accept = server.AcceptConnect(client.CreateConnect(0), 0);
+ var a = Send(server, "A");
+ var b = Send(server, "B");
+
+ foreach (var early in new[] { b, a })
+ {
+ Assert.That(client.Receive(early, 1).Delivered, Is.Empty);
+ var ack = client.CreateAckForReceived(early.Sequence);
+ Assert.That((ack.Ack, ack.AckMask), Is.EqualTo((0u, 0u)), "an unhandled packet was acknowledged");
+ }
+ client.Receive(accept, 2);
+ Assert.That(client.Connected, Is.True);
+ Assert.That(server.OutstandingReliablePacketCount, Is.EqualTo(3), "nothing was acknowledged");
+
+ Assert.That(Names(client.Receive(a, 3)), Is.EqualTo(new[] { "A" }));
+ Assert.That(Names(client.Receive(b, 3)), Is.EqualTo(new[] { "B" }));
+ }
+
+ [Test]
+ public void OrderedFramesAfterTheAcceptAreDeliveredInOrderWhateverArrivesFirst()
+ {
+ var client = Session(1);
+ var server = Session(500);
+ Handshake(client, server);
+ var a = Send(server, "A");
+ var b = Send(server, "B");
+ Assert.That(client.Receive(b, 1).Delivered, Is.Empty);
+ Assert.That(Names(client.Receive(a, 2)), Is.EqualTo(new[] { "A", "B" }));
+ }
+
+ [Test]
+ public void AnUnorderedReliableFrameIsDeliveredOnReceiptNeverHeldBehindAGap()
+ {
+ var client = Session(1);
+ var server = Session(500);
+ Handshake(client, server);
+ Send(server, "lost");
+ var unordered = Send(server, "now", ordered: false, channel: "media");
+ var ordered = Send(server, "later");
+ Assert.That(Names(client.Receive(unordered, 1)), Is.EqualTo(new[] { "now" }));
+ Assert.That(client.Receive(ordered, 2).Delivered, Is.Empty, "the ordered frame waits for the gap");
+ }
+
+ [Test]
+ public void APeerDisconnectDoesNotForgetWhatThePeerSent()
+ {
+ var client = Session(1);
+ var server = Session(500);
+ Handshake(client, server);
+ var s1 = Send(client, "s1");
+ var s2 = Send(client, "s2");
+ Assert.That(server.Receive(s2, 1).Delivered, Is.Empty, "s1 is missing, so s2 is held");
+
+ server.Receive(client.CreateDisconnect(Encoding.UTF8.GetBytes("bye")), 2);
+ Assert.That(Names(server.Receive(s1, 3)), Is.EqualTo(new[] { "s1", "s2" }), "the Disconnect forgot the held frame");
+ Assert.That(server.Receive(s2, 4).Delivered, Is.Empty, "the Disconnect forgot what was received");
+ }
+
+ [Test]
+ public void DefaultInitialSequencesAreDrawnAtRandomFromOneToTwoToTheThirtyFirst()
+ {
+ static void AssertRandom(string label, Func draw)
+ {
+ var draws = Enumerable.Range(0, 64).Select(_ => draw()).ToArray();
+ Assert.That(draws.Distinct().Count(), Is.GreaterThan(32), $"{label} does not draw at random");
+ Assert.That(draws.All(value => value >= 1 && value < (1u << 31)), Is.True, $"{label} left [1, 2^31)");
+ }
+
+ AssertRandom("RandomInitialSequence", CultNetRudpSession.RandomInitialSequence);
+ AssertRandom("session options", () => new CultNetRudpSessionOptions().InitialSequence);
+ AssertRandom("socket transport options", () => new CultNetRudpSocketTransportOptions().InitialSequence);
+ AssertRandom("listener options", () => new CultNetRudpSocketTransportServerOptions().InitialSequence);
+ AssertRandom("schema server options", () => new RudpCultNetSchemaServerOptions().InitialSequence);
+ AssertRandom("cultmesh options", () => new CultMeshRudpSocketOptions().InitialSequence);
+ }
+
+ // Transports.
+
+ [Test]
+ public void ServerModeAdmitsARestartedClientKeepsItsSessionOnARetransmitAndAdmitsANewEndpoint()
+ {
+ using var serverSocket = Bind();
+ using var peerSocket = Bind();
+ using var otherSocket = Bind();
+ using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
+ {
+ RuntimeId = "csharp-rudp-server",
+ Socket = serverSocket,
+ Mode = CultNetRudpSocketMode.Server,
+ ConnectionId = ConnectionId
+ });
+ var serverEndPoint = serverSocket.LocalEndPoint!;
+ void ToServer(Socket from, CultNetRudpPacket packet) => from.SendTo(CultNetRudpPacketCodec.Encode(packet), serverEndPoint);
+
+ var first = Session(50);
+ var connect = first.CreateConnect(0, Encoding.UTF8.GetBytes("same"));
+ ToServer(peerSocket, connect);
+ server.ReceiveOnce();
+ first.Receive(Drain(peerSocket).First(p => p.PacketType == CultNetRudpPacketType.Accept), 0);
+ var sent = new[] { "c0", "c1", "c2" }.Select(name => Send(first, name)).ToArray();
+ foreach (var packet in sent)
+ {
+ ToServer(peerSocket, packet);
+ Assert.That(Encoding.UTF8.GetString(server.ReceiveOnce()!.Payload), Is.EqualTo(Encoding.UTF8.GetString(packet.Payload)));
+ }
+
+ // A retransmitted Connect is a repeat: what was delivered stays delivered once.
+ ToServer(peerSocket, connect);
+ server.ReceiveOnce();
+ ToServer(peerSocket, sent[0]);
+ Assert.That(server.ReceiveOnce(), Is.Null, "the retransmitted Connect reset the session");
+
+ // The same client restarted on the same address, connection id and payload.
+ Drain(peerSocket);
+ var second = Session(7);
+ ToServer(peerSocket, second.CreateConnect(0, Encoding.UTF8.GetBytes("same")));
+ server.ReceiveOnce();
+ second.Receive(Drain(peerSocket).First(p => p.PacketType == CultNetRudpPacketType.Accept), 0);
+ ToServer(peerSocket, Send(second, "after restart"));
+ Assert.That(Encoding.UTF8.GetString(server.ReceiveOnce()!.Payload), Is.EqualTo("after restart"));
+
+ // A Connect from a new endpoint claims the transport.
+ var third = Session(9);
+ ToServer(otherSocket, third.CreateConnect(0, Encoding.UTF8.GetBytes("elsewhere")));
+ server.ReceiveOnce();
+ third.Receive(Drain(otherSocket).First(p => p.PacketType == CultNetRudpPacketType.Accept), 0);
+ ToServer(otherSocket, Send(third, "from elsewhere"));
+ Assert.That(Encoding.UTF8.GetString(server.ReceiveOnce()!.Payload), Is.EqualTo("from elsewhere"));
+ }
+
+ [Test]
+ public void TheListenerAdmitsARestartedClientWithTheSameAddressIdAndPayload()
+ {
+ using var serverSocket = Bind();
+ using var peerSocket = Bind();
+ using var server = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions
+ {
+ RuntimeId = "csharp-rudp-listener",
+ Socket = serverSocket,
+ ConnectionId = ConnectionId
+ });
+ var ended = new List();
+ server.PeerDisconnected += ended.Add;
+ var serverEndPoint = serverSocket.LocalEndPoint!;
+ void ToServer(CultNetRudpPacket packet) => peerSocket.SendTo(CultNetRudpPacketCodec.Encode(packet), serverEndPoint);
+
+ var first = Session(50);
+ ToServer(first.CreateConnect(0, Encoding.UTF8.GetBytes("same")));
+ server.ReceiveOnce();
+ first.Receive(Drain(peerSocket).First(p => p.PacketType == CultNetRudpPacketType.Accept), 0);
+ for (var index = 0; index < 3; index++)
+ {
+ ToServer(Send(first, $"c{index}"));
+ Assert.That(server.ReceiveOnce(), Is.Not.Null);
+ }
+ Assert.That(ended, Is.Empty);
+
+ var second = Session(7);
+ ToServer(second.CreateConnect(0, Encoding.UTF8.GetBytes("same")));
+ server.ReceiveOnce();
+ Assert.That(ended, Has.Count.EqualTo(1), "the restarted client was taken for a retransmit");
+ second.Receive(Drain(peerSocket).First(p => p.PacketType == CultNetRudpPacketType.Accept), 0);
+ ToServer(Send(second, "after restart"));
+ var frame = server.ReceiveOnce();
+ Assert.That(frame, Is.Not.Null);
+ Assert.That(Encoding.UTF8.GetString(frame!.Frame.Payload), Is.EqualTo("after restart"));
+ }
+
+ ///
+ /// Windows reports the ICMP port-unreachable for a datagram sent to a closed port as a connection
+ /// reset on the next receive. Nothing ended, so nothing may be reported. Other platforms do not
+ /// raise it, so the test is inconclusive there.
+ ///
+ [Test]
+ public void AnIcmpConnectionResetDoesNotEndTheSession()
+ {
+ Assume.That(OperatingSystem.IsWindows(), "only Windows raises an ICMP connection reset on UDP");
+ using var serverSocket = Bind();
+ using var peerSocket = Bind();
+ using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
+ {
+ RuntimeId = "csharp-rudp-server",
+ Socket = serverSocket,
+ Mode = CultNetRudpSocketMode.Server,
+ ConnectionId = ConnectionId
+ });
+ peerSocket.SendTo(CultNetRudpPacketCodec.Encode(Session(1).CreateConnect(0)), serverSocket.LocalEndPoint!);
+ server.ReceiveOnce();
+ Assert.That(server.Connected, Is.True);
+
+ var closed = Bind();
+ var closedEndPoint = closed.LocalEndPoint!;
+ closed.Dispose();
+ serverSocket.SendTo(new byte[] { 1 }, closedEndPoint);
+ System.Threading.Thread.Sleep(50);
+ server.ReceiveOnce();
+ Assert.That(server.DisconnectReason, Is.Null, "an ICMP reset claimed an ending that did not happen");
+ Assert.That(server.Connected, Is.True);
+ }
+ }
+}