diff --git a/packages/cultnet-py/tests/test_cultnet.py b/packages/cultnet-py/tests/test_cultnet.py index ca7e700e..dbe6faa9 100644 --- a/packages/cultnet-py/tests/test_cultnet.py +++ b/packages/cultnet-py/tests/test_cultnet.py @@ -94,6 +94,17 @@ from cultnet_py import ( ) +def acknowledges(packet, sequence: int) -> bool: + """Whether the packet's ack field and mask cover the sequence.""" + if packet.ack == sequence: + return True + return ( + packet.ack > sequence + and packet.ack - sequence <= 32 + and (packet.ack_mask & (1 << (packet.ack - sequence - 1))) != 0 + ) + + def bind_udp_socket() -> socket.socket: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.bind(("127.0.0.1", 0)) @@ -1063,14 +1074,13 @@ class CultNetTests(unittest.TestCase): # The peer is told, and the goodbye does not acknowledge the refused frame. (goodbye,) = [p for p in drain_peer() if p.packet_type == CultNetRudpPacketType.DISCONNECT] self.assertTrue(peer.receive(goodbye, 1).disconnected) - self.assertIn(poison.sequence, peer.pending_reliable_sequences) + self.assertFalse(acknowledges(goodbye, poison.sequence), "the goodbye acknowledged the refused frame") # A retransmit is dropped, not acknowledged. to_server(poison) self.assertIsNone(server.receive_once()) self.assertEqual(server.stats.packets_dropped, 2) self.assertEqual(drain_peer(), []) - self.assertIn(poison.sequence, peer.pending_reliable_sequences) # A Connect claims the endpoint again. again = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id)) diff --git a/packages/cultnet-py/tests/test_rudp_session_ending.py b/packages/cultnet-py/tests/test_rudp_session_ending.py index 27932fdc..4be604ab 100644 --- a/packages/cultnet-py/tests/test_rudp_session_ending.py +++ b/packages/cultnet-py/tests/test_rudp_session_ending.py @@ -37,11 +37,26 @@ def client_with_a_lost_write() -> tuple[CultNetRudpSession, CultNetRudpSession]: server = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=CONNECTION_ID, initial_sequence=500)) client.receive(server.accept_connect(client.create_connect(0), 0), 0) client.send("schema", b"owed to the old session", CultNetRudpSendOptions(reliable=True, ordered=True)) - assert client.outstanding_reliable_packet_count == 1 + # A fragmented write larger than the send window leaves part of it queued. + client.send_many( + "schema", bytes(40 * 8), CultNetRudpSendOptions(reliable=True, ordered=True), max_fragment_bytes=8 + ) + assert client.outstanding_reliable_packet_count == 41 + assert client.queued_reliable_packet_count > 0 return client, server -def connected_client() -> tuple[CultNetRudpSocketTransportConnection, socket.socket]: +def drain(sock: socket.socket) -> list: + packets = [] + while True: + try: + wire, _ = sock.recvfrom(65535) + except TimeoutError: + return packets + packets.append(decode_rudp_packet(wire)) + + +def connected_client() -> tuple[CultNetRudpSocketTransportConnection, socket.socket, tuple]: peer_socket = bind_udp_socket() client = CultNetRudpSocketTransportConnection( CultNetRudpSocketTransportOptions( @@ -59,7 +74,7 @@ def connected_client() -> tuple[CultNetRudpSocketTransportConnection, socket.soc peer_socket.sendto(encode_rudp_packet(accept), client_addr) client.receive_once() assert client.connected - return client, peer_socket + return client, peer_socket, client_addr class CultNetRudpSessionEndingTests(unittest.TestCase): @@ -88,7 +103,7 @@ class CultNetRudpSessionEndingTests(unittest.TestCase): ) def test_a_flush_fails_after_a_local_disconnect_or_a_timeout(self) -> None: - client, peer_socket = connected_client() + client, peer_socket, _ = connected_client() try: client.send("schema", b"owed") client.disconnect(b"bye") @@ -98,7 +113,7 @@ class CultNetRudpSessionEndingTests(unittest.TestCase): client.close() peer_socket.close() - client, peer_socket = connected_client() + client, peer_socket, _ = connected_client() try: client.send("schema", b"owed") time.sleep(0.005) @@ -110,7 +125,7 @@ class CultNetRudpSessionEndingTests(unittest.TestCase): peer_socket.close() def test_a_flush_that_started_before_the_end_fails_even_after_a_later_connect(self) -> None: - client, peer_socket = connected_client() + client, peer_socket, _ = connected_client() outcome: list[BaseException | None] = [] def flush() -> None: @@ -135,5 +150,56 @@ class CultNetRudpSessionEndingTests(unittest.TestCase): peer_socket.close() + def test_a_flush_started_after_a_reconnect_waits_on_the_new_session(self) -> None: + client, peer_socket, client_addr = connected_client() + try: + client.send("schema", b"owed to the old session") + client.disconnect(b"bye") + client.connect(b"hello again") + connect = [p for p in drain(peer_socket) if p.packet_type == CultNetRudpPacketType.CONNECT][-1] + accept = CultNetRudpSession( + CultNetRudpSessionOptions(connection_id=CONNECTION_ID, initial_sequence=700) + ).accept_connect(connect, 0) + peer_socket.sendto(encode_rudp_packet(accept), client_addr) + client.receive_once() + + client.flush_reliable(0.5) + finally: + client.close() + peer_socket.close() + + def test_a_flush_started_after_the_server_accepts_a_new_connect_waits_on_the_new_session(self) -> None: + server_socket = bind_udp_socket() + peer_socket = bind_udp_socket() + server = CultNetRudpSocketTransportConnection( + CultNetRudpSocketTransportOptions( + runtime_id="python-rudp-server", + socket=server_socket, + mode=CultNetRudpSocketMode.SERVER, + connection_id=CONNECTION_ID, + ) + ) + to_server = lambda packet: peer_socket.sendto(encode_rudp_packet(packet), server_socket.getsockname()) + try: + first = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=CONNECTION_ID)) + to_server(first.create_connect(0)) + server.receive_once() + to_server(first.create_disconnect()) + server.receive_once() + + following = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=CONNECTION_ID)) + to_server(following.create_connect(0)) + server.receive_once() + accept = [p for p in drain(peer_socket) if p.packet_type == CultNetRudpPacketType.ACCEPT][-1] + following.receive(accept, 0) + to_server(following.create_ack()) + server.receive_once() + + server.flush_reliable(0.5) + finally: + server.close() + peer_socket.close() + + if __name__ == "__main__": unittest.main() diff --git a/packages/cultnet-rs/tests/rudp_stray_packets.rs b/packages/cultnet-rs/tests/rudp_stray_packets.rs index ea485a4d..c2de98c2 100644 --- a/packages/cultnet-rs/tests/rudp_stray_packets.rs +++ b/packages/cultnet-rs/tests/rudp_stray_packets.rs @@ -766,7 +766,19 @@ fn client_with_a_lost_write() -> Result<(CultNetRudpSession, CultNetRudpSession) ..Default::default() }, )?; - assert_eq!(client.outstanding_reliable_packet_count(), 1); + // A fragmented write larger than the send window leaves part of it queued. + client.send_many( + "schema", + vec![7u8; 40 * 8], + CultNetRudpSendOptions { + reliable: true, + ordered: true, + ..Default::default() + }, + Some(8), + )?; + assert_eq!(client.outstanding_reliable_packet_count(), 41); + assert!(client.queued_reliable_packet_count() > 0); Ok((client, server)) } @@ -873,3 +885,56 @@ fn a_flush_fails_after_a_local_disconnect_or_a_timeout() -> Result<()> { assert!(error.to_string().contains("ended before its reliable writes were acknowledged"), "{error}"); Ok(()) } + +/// A flush that starts after a reconnect waits on the new session: the end of +/// the old one does not fail it. +#[test] +fn a_flush_started_after_a_reconnect_waits_on_the_new_session() -> Result<()> { + let server = socket()?; + let server_addr = server.local_addr()?; + let (mut client, _peer, client_addr) = connected_client(&server, server_addr)?; + client.send_reliable("schema", b"owed to the old session".to_vec())?; + client.disconnect(b"bye".to_vec())?; + client.connect(b"hello again".to_vec())?; + let connect = drain(&server)? + .into_iter() + .rfind(|packet| packet.packet_type == CultNetRudpPacketType::Connect) + .expect("the reconnect reached the peer"); + let accept = raw_session(CONNECTION_ID).accept_connect(&connect, 0, Vec::new())?; + send_to(&server, client_addr, &accept)?; + let _ = client.receive_once()?; + + client.flush_reliable(Duration::from_millis(500))?; + Ok(()) +} + +/// A server-mode transport that accepts a new Connect after its peer left has a +/// live session again, and a flush started in it waits on that session. +#[test] +fn a_flush_started_after_the_server_accepts_a_new_connect_waits_on_the_new_session() -> Result<()> { + let server_socket = socket()?; + let server_addr = server_socket.local_addr()?; + let mut server = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::server("server", server_socket, CONNECTION_ID), + )?; + let peer = socket()?; + let mut first = raw_session(CONNECTION_ID); + send_to(&peer, server_addr, &first.create_connect(0, b"peer".to_vec())?)?; + let _ = server.receive_once()?; + send_to(&peer, server_addr, &first.create_disconnect(b"bye".to_vec()))?; + let _ = server.receive_once()?; + + let mut next = raw_session(CONNECTION_ID); + send_to(&peer, server_addr, &next.create_connect(0, b"peer".to_vec())?)?; + let _ = server.receive_once()?; + let accept = drain(&peer)? + .into_iter() + .rfind(|packet| packet.packet_type == CultNetRudpPacketType::Accept) + .expect("the new Connect was accepted"); + next.receive(&accept, 0)?; + send_to(&peer, server_addr, &next.create_ack_for_received(accept.sequence))?; + let _ = server.receive_once()?; + + server.flush_reliable(Duration::from_millis(500))?; + Ok(()) +} diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index f67311c9..0f4a5f8f 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -166,6 +166,14 @@ function createDuplexPair(): { a: Duplex; b: Duplex } { return { a, b }; } +/// Whether the packet's ack field and mask cover the sequence. +function acknowledges(packet: CultNetRudpPacket, sequence: number): boolean { + if (packet.ack === sequence) return true; + return packet.ack > sequence + && packet.ack - sequence <= 32 + && (packet.ackMask & (1 << (packet.ack - sequence - 1))) !== 0; +} + async function bindUdpSocket(): Promise { const socket = dgram.createSocket("udp4"); await new Promise((resolve) => socket.bind(0, "127.0.0.1", resolve)); @@ -1975,7 +1983,7 @@ test("rudp socket transport ends the session that refuses a reliable frame and n // neither in the goodbye nor after it retransmits. const goodbye = received.find((p) => p.packetType === "disconnect")!; assert.equal(peer.receive(goodbye, 1).disconnected, true); - assert.ok(peer.pendingReliableSequences.includes(poison.sequence), "the refused frame was acknowledged"); + assert.equal(acknowledges(goodbye, poison.sequence), false, "the goodbye acknowledged the refused frame"); const before = received.length; toServer(poison); await waitFor(() => server.stats.packetsDropped === 2, "the retransmit dropped"); @@ -2489,3 +2497,84 @@ test("operation service sweeps a session that ran a handler once it idles out", await server.close(); } }); + +test("a flush started after a reconnect waits on the new session", async () => { + const peerSocket = await bindUdpSocket(); + const clientSocket = await bindUdpSocket(); + const connectionId = 0x1020305b; + const client = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-client", + socket: clientSocket, + mode: "client", + remoteHost: "127.0.0.1", + remotePort: udpPort(peerSocket), + connectionId, + resendDelayMs: 1000, + resendPollMs: 1000, + }); + const received: CultNetRudpPacket[] = []; + let clientPort = 0; + peerSocket.on("message", (wire, remote) => { + clientPort = remote.port; + received.push(decodeRudpPacket(wire)); + }); + try { + client.connect(); + await waitFor(() => received.some((p) => p.packetType === "connect"), "the Connect"); + const first = new CultNetRudpSession({ connectionId }); + peerSocket.send(encodeRudpPacket(first.acceptConnect(received[0]!, 0)), clientPort, "127.0.0.1"); + await waitFor(() => client.connected, "the client connected"); + client.send("schema", Buffer.from("owed to the old session")); + client.checkTimeout(50, Date.now() + 10_000); + assert.equal(client.connected, false); + + client.connect(); + await waitFor(() => received.filter((p) => p.packetType === "connect").length === 2, "the reconnect"); + const second = new CultNetRudpSession({ connectionId }); + const reconnect = received.filter((p) => p.packetType === "connect")[1]!; + peerSocket.send(encodeRudpPacket(second.acceptConnect(reconnect, 0)), clientPort, "127.0.0.1"); + await waitFor(() => client.connected, "the client reconnected"); + await client.flush(500); + } finally { + peerSocket.close(); + client.close(); + } +}); + +test("a flush started after the server accepts a new Connect waits on the new session", async () => { + const serverSocket = await bindUdpSocket(); + const peerSocket = await bindUdpSocket(); + const connectionId = 0x1020305c; + const server = new CultNetRudpSocketTransportConnection({ + runtimeId: "rudp-server", + socket: serverSocket, + mode: "server", + connectionId, + resendDelayMs: 1000, + resendPollMs: 1000, + }); + const received: CultNetRudpPacket[] = []; + peerSocket.on("message", (wire) => received.push(decodeRudpPacket(wire))); + const toServer = (packet: CultNetRudpPacket) => + peerSocket.send(encodeRudpPacket(packet), udpPort(serverSocket), "127.0.0.1"); + try { + const first = new CultNetRudpSession({ connectionId }); + toServer(first.createConnect(0)); + await waitFor(() => received.some((p) => p.packetType === "accept"), "the first Accept"); + toServer(first.createDisconnect(Buffer.from("bye"))); + await waitFor(() => !server.connected, "the first session ended"); + + const next = new CultNetRudpSession({ connectionId }); + toServer(next.createConnect(0)); + await waitFor(() => received.filter((p) => p.packetType === "accept").length === 2, "the second Accept"); + const accept = received.filter((p) => p.packetType === "accept")[1]!; + next.receive(accept, 0); + toServer(next.createAckForReceived(accept.sequence)); + await waitFor(() => server.outstandingReliablePacketCount === 0, "the second Accept acknowledged"); + + await server.flush(500); + } finally { + peerSocket.close(); + server.close(); + } +}); diff --git a/packages/cultnet-ts/test/rudp-session-ending.test.ts b/packages/cultnet-ts/test/rudp-session-ending.test.ts index f0d3fae2..bf69c286 100644 --- a/packages/cultnet-ts/test/rudp-session-ending.test.ts +++ b/packages/cultnet-ts/test/rudp-session-ending.test.ts @@ -13,7 +13,10 @@ function clientWithALostWrite(): { client: CultNetRudpSession; server: CultNetRu const server = new CultNetRudpSession({ connectionId, initialSequence: 500 }); client.receive(server.acceptConnect(client.createConnect(0), 0), 0); client.send("schema", enc("owed to the old session"), { reliable: true, ordered: true }); - assert.equal(client.outstandingReliablePacketCount, 1); + // A fragmented write larger than the send window leaves part of it queued. + client.sendMany("schema", new Uint8Array(40 * 8), { reliable: true, ordered: true, maxFragmentBytes: 8 }); + assert.equal(client.outstandingReliablePacketCount, 41); + assert.ok(client.queuedReliablePacketCount > 0); return { client, server }; } diff --git a/tests/GameCult.Networking.Tests/CultNetRudpSessionEndingTests.cs b/tests/GameCult.Networking.Tests/CultNetRudpSessionEndingTests.cs index 3a35b450..fff7eeeb 100644 --- a/tests/GameCult.Networking.Tests/CultNetRudpSessionEndingTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetRudpSessionEndingTests.cs @@ -33,7 +33,10 @@ namespace GameCult.Networking.Tests var server = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId, InitialSequence = 500 }); client.Receive(server.AcceptConnect(client.CreateConnect(0), 0), 0); client.Send("schema", Encoding.UTF8.GetBytes("owed to the old session"), new CultNetRudpSendOptions { Reliable = true, Ordered = true }); - Assert.That(client.OutstandingReliablePacketCount, Is.EqualTo(1)); + // A fragmented write larger than the send window leaves part of it queued. + client.SendMany("schema", new byte[40 * 8], new CultNetRudpSendOptions { Reliable = true, Ordered = true }, maxFragmentBytes: 8); + Assert.That(client.OutstandingReliablePacketCount, Is.EqualTo(41)); + Assert.That(client.QueuedReliablePacketCount, Is.GreaterThan(0)); return (client, server); } @@ -65,7 +68,7 @@ namespace GameCult.Networking.Tests } } - private static (CultNetRudpSocketTransportConnection Client, Socket Peer, CultNetRudpSession PeerSession) ConnectedClient() + private static (CultNetRudpSocketTransportConnection Client, Socket Peer, EndPoint ClientEndPoint) ConnectedClient() { var peerSocket = Bind(); var client = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions @@ -85,7 +88,7 @@ namespace GameCult.Networking.Tests peerSocket.SendTo(CultNetRudpPacketCodec.Encode(accept), from); client.ReceiveOnce(); Assert.That(client.Connected, Is.True); - return (client, peerSocket, peer); + return (client, peerSocket, from); } [Test] @@ -128,5 +131,66 @@ namespace GameCult.Networking.Tests Assert.That(error.InnerException, Is.InstanceOf()); } } + + 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 AFlushStartedAfterAReconnectWaitsOnTheNewSession() + { + var (client, peerSocket, clientEndPoint) = ConnectedClient(); + using (client) + using (peerSocket) + { + client.Send("schema", Encoding.UTF8.GetBytes("owed to the old session")); + client.Disconnect(); + client.Connect("hello again"); + var connect = Drain(peerSocket).Last(packet => packet.PacketType == CultNetRudpPacketType.Connect); + var accept = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId, InitialSequence = 700 }).AcceptConnect(connect, 0); + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(accept), clientEndPoint); + client.ReceiveOnce(); + + Assert.DoesNotThrow(() => client.FlushReliable(TimeSpan.FromMilliseconds(500))); + } + } + + [Test] + public void AFlushStartedAfterTheServerAcceptsANewConnectWaitsOnTheNewSession() + { + 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 + }); + var first = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId }); + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(first.CreateConnect(0)), serverSocket.LocalEndPoint!); + server.ReceiveOnce(); + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(first.CreateDisconnect()), serverSocket.LocalEndPoint!); + server.ReceiveOnce(); + + var next = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId }); + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(next.CreateConnect(0)), serverSocket.LocalEndPoint!); + server.ReceiveOnce(); + var accept = Drain(peerSocket).Last(packet => packet.PacketType == CultNetRudpPacketType.Accept); + next.Receive(accept, 0); + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(next.CreateAck()), serverSocket.LocalEndPoint!); + server.ReceiveOnce(); + + Assert.DoesNotThrow(() => server.FlushReliable(TimeSpan.FromMilliseconds(500))); + } } } diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 63379143..1ae5d116 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1269,6 +1269,16 @@ namespace GameCult.Networking.Tests Assert.That(Encoding.UTF8.GetString(ReceiveRudpFrame(client).Payload), Is.EqualTo("server still here")); } + /// Whether the packet's ack field and mask cover the sequence. + private static bool Acknowledges(CultNetRudpPacket packet, uint sequence) + { + if (packet.Ack == sequence) + return true; + return packet.Ack > sequence + && packet.Ack - sequence <= 32 + && (packet.AckMask & (1u << (int)(packet.Ack - sequence - 1))) != 0; + } + private static List DrainPackets(Socket socket) { var packets = new List(); @@ -1397,7 +1407,7 @@ namespace GameCult.Networking.Tests var goodbye = DrainPackets(peerSocket).Single(); Assert.That(goodbye.PacketType, Is.EqualTo(CultNetRudpPacketType.Disconnect)); Assert.That(peer.Receive(goodbye, 1).Disconnected, Is.True); - Assert.That(peer.PendingReliableSequences, Does.Contain(poison.Sequence)); + Assert.That(Acknowledges(goodbye, poison.Sequence), Is.False, "the goodbye acknowledged the refused frame"); // A retransmit is dropped, not acknowledged. peerSocket.SendTo(CultNetRudpPacketCodec.Encode(poison), serverEndPoint); @@ -1444,7 +1454,7 @@ namespace GameCult.Networking.Tests var goodbye = DrainPackets(peerSocket).Single(); Assert.That(goodbye.PacketType, Is.EqualTo(CultNetRudpPacketType.Disconnect)); Assert.That(peer.Receive(goodbye, 1).Disconnected, Is.True); - Assert.That(peer.PendingReliableSequences, Does.Contain(poison.Sequence)); + Assert.That(Acknowledges(goodbye, poison.Sequence), Is.False, "the goodbye acknowledged the refused frame"); peerSocket.SendTo(CultNetRudpPacketCodec.Encode(poison), serverEndPoint); Assert.That(server.ReceiveOnce(), Is.Null);