From 536e650d82df6a9a25163ea49a4062f190e91906 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 05:52:37 +0200 Subject: [PATCH] Ack Cut 1c tests (C#, Python): Connect generation, handshake-seeded watermark, restarts Co-Authored-By: Claude Sonnet 5.5 --- packages/cultnet-py/tests/test_cultnet.py | 74 +++- .../tests/test_rudp_connect_generation.py | 178 +++++++++ .../CultNetRudpConnectGenerationTests.cs | 375 ++++++++++++++++++ 3 files changed, 626 insertions(+), 1 deletion(-) create mode 100644 packages/cultnet-py/tests/test_rudp_connect_generation.py create mode 100644 tests/GameCult.Networking.Tests/CultNetRudpConnectGenerationTests.cs 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); + } + } +}