diff --git a/packages/cultnet-py/tests/test_cultnet.py b/packages/cultnet-py/tests/test_cultnet.py index 7e78c48f..ca7e700e 100644 --- a/packages/cultnet-py/tests/test_cultnet.py +++ b/packages/cultnet-py/tests/test_cultnet.py @@ -731,12 +731,7 @@ class CultNetTests(unittest.TestCase): def test_cultnet_rudp_session_suppresses_duplicates_and_delivers_reliable_ordered_payloads(self) -> None: sender = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=123, initial_sequence=1)) receiver = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=123, initial_sequence=100)) - sender.receive( - CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 123, 90, 0, 0, "control") - ) - receiver.receive( - CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 123, 91, 0, 0, "control") - ) + sender.receive(receiver.accept_connect(sender.create_connect(0), 0)) first = sender.send("schema", b"first", CultNetRudpSendOptions(reliable=True, ordered=True)) second = sender.send("schema", b"second", CultNetRudpSendOptions(reliable=True, ordered=True)) diff --git a/packages/cultnet-py/tests/test_rudp_ordered_delivery.py b/packages/cultnet-py/tests/test_rudp_ordered_delivery.py new file mode 100644 index 00000000..f6b398bc --- /dev/null +++ b/packages/cultnet-py/tests/test_rudp_ordered_delivery.py @@ -0,0 +1,109 @@ +"""Ordered delivery has one owner: the contiguous received-through watermark. + +A frame is held for exactly one reason, a reliable sequence below it has not +arrived, and it is delivered in the call that fills the last such gap, +whichever channel the gap belonged to. +""" +from __future__ import annotations + +import unittest + +from cultnet_py.transport import ( + CultNetRudpPacket, + CultNetRudpPacketType, + CultNetRudpSendOptions, + CultNetRudpSession, + CultNetRudpSessionOptions, +) + + +def handshake() -> tuple[CultNetRudpSession, CultNetRudpSession]: + client = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=410, initial_sequence=1)) + server = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=410, initial_sequence=500)) + accept = server.accept_connect(client.create_connect(0), 0) + client.receive(accept, 0) + return client, server + + +def send(session: CultNetRudpSession, channel: str, payload: str, ordered: bool) -> CultNetRudpPacket: + return session.send(channel, payload.encode(), CultNetRudpSendOptions(reliable=True, ordered=ordered)) + + +def names(result) -> list[str]: + return [frame.payload.decode() for frame in result.delivered] + + +class CultNetRudpOrderedDeliveryTests(unittest.TestCase): + def test_ordered_frame_waits_for_a_gap_filled_by_another_channel(self) -> None: + sender, receiver = handshake() + o1 = send(sender, "schema", "o1", True) + u = send(sender, "rel", "u", False) + o2 = send(sender, "schema", "o2", True) + o3 = send(sender, "schema", "o3", True) + + self.assertEqual(names(receiver.receive(o1, 1)), ["o1"]) + self.assertEqual(names(receiver.receive(o2, 2)), [], "u is missing") + self.assertEqual(names(receiver.receive(o3, 3)), [], "u is missing") + self.assertEqual(names(receiver.receive(u, 4)), ["u", "o2", "o3"]) + + def test_two_ordered_channels_release_each_other_in_sequence_order(self) -> None: + sender, receiver = handshake() + a1 = send(sender, "schema", "A1", True) + b1 = send(sender, "other", "B1", True) + a2 = send(sender, "schema", "A2", True) + a3 = send(sender, "schema", "A3", True) + + delivered = names(receiver.receive(a1, 1)) + names(receiver.receive(a2, 2)) + names(receiver.receive(a3, 3)) + self.assertEqual(delivered, ["A1"], "B1 is missing, so A2 and A3 wait") + self.assertEqual(names(receiver.receive(b1, 4)), ["B1", "A2", "A3"]) + + def test_channel_first_used_after_other_traffic_loses_nothing(self) -> None: + sender, receiver = handshake() + c1 = send(sender, "late", "C1", True) + c2 = send(sender, "late", "C2", True) + x = send(sender, "schema", "X", True) + + self.assertEqual(names(receiver.receive(x, 1)), [], "C1 is missing") + self.assertEqual(names(receiver.receive(c2, 2)), [], "C1 is missing") + self.assertEqual(names(receiver.receive(c1, 3)), ["C1", "C2", "X"]) + + def test_accept_seeds_the_watermark_of_the_connecting_side(self) -> None: + client, server = handshake() + s1 = send(server, "schema", "s1", True) + s2 = send(server, "schema", "s2", True) + + self.assertEqual(names(client.receive(s2, 1)), []) + self.assertEqual(names(client.receive(s1, 2)), ["s1", "s2"]) + + def test_connect_seeds_the_watermark_of_the_accepting_side(self) -> None: + client, server = handshake() + first = send(client, "schema", "first", True) + second = send(client, "schema", "second", True) + + self.assertEqual(names(server.receive(second, 1)), []) + self.assertEqual(names(server.receive(first, 2)), ["first", "second"]) + + def test_duplicate_of_a_held_frame_is_not_delivered_twice(self) -> None: + sender, receiver = handshake() + s1 = send(sender, "schema", "s1", True) + s2 = send(sender, "schema", "s2", True) + + self.assertEqual(names(receiver.receive(s2, 1)), []) + self.assertEqual(names(receiver.receive(s2, 2)), []) + self.assertEqual(names(receiver.receive(s1, 3)), ["s1", "s2"]) + self.assertEqual(names(receiver.receive(s2, 4)), []) + + def test_reset_peer_state_forgets_held_frames_and_the_watermark(self) -> None: + sender, receiver = handshake() + s1 = send(sender, "schema", "s1", True) + s2 = send(sender, "schema", "s2", True) + self.assertEqual(names(receiver.receive(s2, 1)), []) + + receiver.reset_peer_state() + receiver.receive(CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 410, 1, 0, 0, "control"), 2) + + self.assertEqual(names(receiver.receive(s1, 3)), ["s1"], "the held s2 died with the reset") + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index 426e2ed9..ca760ebe 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -1009,42 +1009,9 @@ fn rudp_session_suppresses_duplicates_and_delivers_reliable_ordered_payloads_in_ initial_sequence: 100, ..CultNetRudpSessionOptions::default() }); - sender.receive( - &CultNetRudpPacket { - packet_type: CultNetRudpPacketType::Accept, - connection_id: 123, - sequence: 90, - ack: 0, - ack_mask: 0, - channel_id: "control".to_string(), - reliable: false, - ordered: false, - sequenced: false, - fragment_id: 0, - fragment_index: 0, - fragment_count: 0, - payload: Vec::new(), - }, - 0, - )?; - receiver.receive( - &CultNetRudpPacket { - packet_type: CultNetRudpPacketType::Accept, - connection_id: 123, - sequence: 91, - ack: 0, - ack_mask: 0, - channel_id: "control".to_string(), - reliable: false, - ordered: false, - sequenced: false, - fragment_id: 0, - fragment_index: 0, - fragment_count: 0, - payload: Vec::new(), - }, - 0, - )?; + let connect = sender.create_connect(0, Vec::new())?; + let accept = receiver.accept_connect(&connect, 0, Vec::new())?; + sender.receive(&accept, 0)?; let options = CultNetRudpSendOptions { reliable: true, @@ -2667,6 +2634,18 @@ fn rudp_ordered_channel_tracking_is_bounded() -> Result<()> { let accept = receiver.accept_connect(&connect, 1, Vec::new())?; sender.receive(&accept, 2)?; + // Channels are tracked only while they hold a frame, so the bound is on + // channels held behind one missing sequence. + let gap = sender.send( + "gap", + b"lost".to_vec(), + CultNetRudpSendOptions { + reliable: true, + ordered: false, + ..CultNetRudpSendOptions::default() + }, + )?; + sender.receive(&receiver.create_ack_for(gap.sequence), 3)?; let mut error = None; for channel in 0..256u32 { let packet = sender.send( @@ -2679,7 +2658,8 @@ fn rudp_ordered_channel_tracking_is_bounded() -> Result<()> { }, )?; match receiver.receive(&packet, 3) { - Ok(_) => { + Ok(result) => { + assert!(result.delivered.is_empty(), "the gap holds every frame"); sender.receive(&receiver.create_ack_for(packet.sequence), 3)?; } Err(refused) => { @@ -2743,6 +2723,134 @@ fn rudp_ordered_hold_buffer_is_bounded() -> Result<()> { Ok(()) } +fn rudp_ordered_pair() -> Result<(CultNetRudpSession, CultNetRudpSession)> { + let mut client = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 410, + initial_sequence: 1, + ..CultNetRudpSessionOptions::default() + }); + let mut server = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 410, + initial_sequence: 500, + ..CultNetRudpSessionOptions::default() + }); + let connect = client.create_connect(0, Vec::new())?; + let accept = server.accept_connect(&connect, 0, Vec::new())?; + client.receive(&accept, 0)?; + Ok((client, server)) +} + +fn rudp_send_options(ordered: bool) -> CultNetRudpSendOptions { + CultNetRudpSendOptions { + reliable: true, + ordered, + ..CultNetRudpSendOptions::default() + } +} + +fn rudp_delivered_names(delivered: &[cultnet_rs::CultNetRudpDeliveredFrame]) -> Vec { + delivered + .iter() + .map(|frame| String::from_utf8(frame.payload.clone()).unwrap()) + .collect() +} + +#[test] +fn rudp_ordered_frame_waits_for_a_gap_filled_by_another_channel() -> Result<()> { + let (mut sender, mut receiver) = rudp_ordered_pair()?; + let o1 = sender.send("schema", b"o1".to_vec(), rudp_send_options(true))?; + let u = sender.send("media", b"u".to_vec(), rudp_send_options(false))?; + let o2 = sender.send("schema", b"o2".to_vec(), rudp_send_options(true))?; + let o3 = sender.send("schema", b"o3".to_vec(), rudp_send_options(true))?; + + assert_eq!(rudp_delivered_names(&receiver.receive(&o1, 1)?.delivered), ["o1"]); + // `u` is lost, so nothing behind it may be delivered. + assert!(receiver.receive(&o2, 2)?.delivered.is_empty()); + assert!(receiver.receive(&o3, 3)?.delivered.is_empty()); + // The other channel's packet fills the gap and releases the ordered frames + // in the same call. + assert_eq!( + rudp_delivered_names(&receiver.receive(&u, 4)?.delivered), + ["u", "o2", "o3"] + ); + Ok(()) +} + +#[test] +fn rudp_two_ordered_channels_release_each_other_in_sequence_order() -> Result<()> { + let (mut sender, mut receiver) = rudp_ordered_pair()?; + let a1 = sender.send("schema", b"A1".to_vec(), rudp_send_options(true))?; + let b1 = sender.send("other", b"B1".to_vec(), rudp_send_options(true))?; + let a2 = sender.send("schema", b"A2".to_vec(), rudp_send_options(true))?; + let a3 = sender.send("schema", b"A3".to_vec(), rudp_send_options(true))?; + + let mut delivered = rudp_delivered_names(&receiver.receive(&a1, 1)?.delivered); + delivered.extend(rudp_delivered_names(&receiver.receive(&a2, 2)?.delivered)); + delivered.extend(rudp_delivered_names(&receiver.receive(&a3, 3)?.delivered)); + assert_eq!(delivered, ["A1"], "B1 is missing, so A2 and A3 wait"); + assert_eq!( + rudp_delivered_names(&receiver.receive(&b1, 4)?.delivered), + ["B1", "A2", "A3"] + ); + Ok(()) +} + +#[test] +fn rudp_channel_first_used_after_other_traffic_loses_nothing() -> Result<()> { + let (mut sender, mut receiver) = rudp_ordered_pair()?; + let c1 = sender.send("late", b"C1".to_vec(), rudp_send_options(true))?; + let c2 = sender.send("late", b"C2".to_vec(), rudp_send_options(true))?; + let x = sender.send("schema", b"X".to_vec(), rudp_send_options(true))?; + + // X, then C2, arrive while C1 is still in flight. + let mut delivered = rudp_delivered_names(&receiver.receive(&x, 1)?.delivered); + delivered.extend(rudp_delivered_names(&receiver.receive(&c2, 2)?.delivered)); + assert!(delivered.is_empty(), "C1 is missing, so nothing is delivered"); + assert_eq!( + rudp_delivered_names(&receiver.receive(&c1, 3)?.delivered), + ["C1", "C2", "X"] + ); + Ok(()) +} + +#[test] +fn rudp_accept_seeds_the_watermark_of_the_connecting_side() -> Result<()> { + let (mut client, mut server) = rudp_ordered_pair()?; + let s1 = server.send("schema", b"s1".to_vec(), rudp_send_options(true))?; + let s2 = server.send("schema", b"s2".to_vec(), rudp_send_options(true))?; + assert!(client.receive(&s2, 1)?.delivered.is_empty()); + assert_eq!( + rudp_delivered_names(&client.receive(&s1, 2)?.delivered), + ["s1", "s2"] + ); + Ok(()) +} + +#[test] +fn rudp_large_ordered_frame_without_a_gap_is_never_held_or_refused() -> Result<()> { + let (mut sender, mut receiver) = rudp_ordered_pair()?; + let packets = sender.send_many( + "schema", + vec![7u8; 5 * 1024 * 1024], + rudp_send_options(true), + Some(60_000), + )?; + let mut queue = packets; + let mut delivered = Vec::new(); + while !queue.is_empty() { + let mut next = Vec::new(); + for packet in &queue { + delivered.extend(receiver.receive(packet, 1)?.delivered); + let ack = receiver.create_ack_for_received(packet.sequence); + next.extend(sender.receive(&ack, 1)?.ready_to_send); + } + queue = next; + } + assert_eq!(delivered.len(), 1); + assert_eq!(delivered[0].payload.len(), 5 * 1024 * 1024); + Ok(()) +} + #[test] fn rudp_far_ahead_sequence_cannot_starve_the_session() -> Result<()> { let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions { diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index 35cba767..f67311c9 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -628,8 +628,7 @@ test("rudp session bounds pending reliable packets before enqueue", () => { test("rudp session suppresses duplicates and delivers reliable ordered payloads in sequence", () => { const sender = new CultNetRudpSession({ connectionId: 123, initialSequence: 1 }); const receiver = new CultNetRudpSession({ connectionId: 123, initialSequence: 100 }); - sender.receive({ packetType: "accept", connectionId: 123, sequence: 90, ack: 0, ackMask: 0, channelId: "control" }); - receiver.receive({ packetType: "accept", connectionId: 123, sequence: 91, ack: 0, ackMask: 0, channelId: "control" }); + sender.receive(receiver.acceptConnect(sender.createConnect(0), 0)); const first = sender.send("schema", Buffer.from("first"), { reliable: true, ordered: true }); const second = sender.send("schema", Buffer.from("second"), { reliable: true, ordered: true }); diff --git a/packages/cultnet-ts/test/rudp-ordered-delivery.test.ts b/packages/cultnet-ts/test/rudp-ordered-delivery.test.ts new file mode 100644 index 00000000..4e6bc05a --- /dev/null +++ b/packages/cultnet-ts/test/rudp-ordered-delivery.test.ts @@ -0,0 +1,103 @@ +// Ordered delivery has one owner: the contiguous received-through watermark. A +// frame is held for exactly one reason, a reliable sequence below it has not +// arrived, and it is delivered in the call that fills the last such gap, +// whichever channel the gap belonged to. +import assert from "node:assert/strict"; +import test from "node:test"; +import { CultNetRudpSession } from "../src"; + +const enc = (text: string) => new TextEncoder().encode(text); +const dec = (bytes: Uint8Array) => new TextDecoder().decode(bytes); + +function handshake(): { client: CultNetRudpSession; server: CultNetRudpSession } { + const client = new CultNetRudpSession({ connectionId: 410, initialSequence: 1 }); + const server = new CultNetRudpSession({ connectionId: 410, initialSequence: 500 }); + const accept = server.acceptConnect(client.createConnect(0), 0); + client.receive(accept, 0); + return { client, server }; +} + +const names = (result: { delivered: { payload: Uint8Array }[] }) => result.delivered.map((frame) => dec(frame.payload)); +const send = (session: CultNetRudpSession, channel: string, payload: string, ordered: boolean) => + session.send(channel, enc(payload), { reliable: true, ordered }); + +test("an ordered frame waits for a gap filled by another channel", () => { + const { client: sender, server: receiver } = handshake(); + const o1 = send(sender, "schema", "o1", true); + const u = send(sender, "rel", "u", false); + const o2 = send(sender, "schema", "o2", true); + const o3 = send(sender, "schema", "o3", true); + + assert.deepEqual(names(receiver.receive(o1, 1)), ["o1"]); + assert.deepEqual(names(receiver.receive(o2, 2)), [], "u is missing"); + assert.deepEqual(names(receiver.receive(o3, 3)), [], "u is missing"); + assert.deepEqual(names(receiver.receive(u, 4)), ["u", "o2", "o3"]); +}); + +test("two ordered channels release each other in sequence order", () => { + const { client: sender, server: receiver } = handshake(); + const a1 = send(sender, "schema", "A1", true); + const b1 = send(sender, "other", "B1", true); + const a2 = send(sender, "schema", "A2", true); + const a3 = send(sender, "schema", "A3", true); + + const delivered = [ + ...names(receiver.receive(a1, 1)), + ...names(receiver.receive(a2, 2)), + ...names(receiver.receive(a3, 3)), + ]; + assert.deepEqual(delivered, ["A1"], "B1 is missing, so A2 and A3 wait"); + assert.deepEqual(names(receiver.receive(b1, 4)), ["B1", "A2", "A3"]); +}); + +test("a channel first used after other traffic loses nothing", () => { + const { client: sender, server: receiver } = handshake(); + const c1 = send(sender, "late", "C1", true); + const c2 = send(sender, "late", "C2", true); + const x = send(sender, "schema", "X", true); + + assert.deepEqual(names(receiver.receive(x, 1)), [], "C1 is missing"); + assert.deepEqual(names(receiver.receive(c2, 2)), [], "C1 is missing"); + assert.deepEqual(names(receiver.receive(c1, 3)), ["C1", "C2", "X"]); +}); + +test("Accept seeds the watermark of the connecting side", () => { + const { client, server } = handshake(); + const s1 = send(server, "schema", "s1", true); + const s2 = send(server, "schema", "s2", true); + + assert.deepEqual(names(client.receive(s2, 1)), []); + assert.deepEqual(names(client.receive(s1, 2)), ["s1", "s2"]); +}); + +test("Connect seeds the watermark of the accepting side", () => { + const { client, server } = handshake(); + const first = send(client, "schema", "first", true); + const second = send(client, "schema", "second", true); + + assert.deepEqual(names(server.receive(second, 1)), []); + assert.deepEqual(names(server.receive(first, 2)), ["first", "second"]); +}); + +test("a duplicate of a held frame is not delivered twice", () => { + const { client: sender, server: receiver } = handshake(); + const s1 = send(sender, "schema", "s1", true); + const s2 = send(sender, "schema", "s2", true); + + assert.deepEqual(names(receiver.receive(s2, 1)), []); + assert.deepEqual(names(receiver.receive(s2, 2)), []); + assert.deepEqual(names(receiver.receive(s1, 3)), ["s1", "s2"]); + assert.deepEqual(names(receiver.receive(s2, 4)), []); +}); + +test("resetting peer state forgets held frames and the watermark", () => { + const { client: sender, server: receiver } = handshake(); + const s1 = send(sender, "schema", "s1", true); + const s2 = send(sender, "schema", "s2", true); + assert.deepEqual(names(receiver.receive(s2, 1)), []); + + receiver.resetPeerState(); + receiver.receive({ packetType: "accept", connectionId: 410, sequence: 1, ack: 0, ackMask: 0, channelId: "control" }, 2); + + assert.deepEqual(names(receiver.receive(s1, 3)), ["s1"], "the held s2 died with the reset"); +}); diff --git a/tests/GameCult.Networking.Tests/CultNetRudpOrderedDeliveryTests.cs b/tests/GameCult.Networking.Tests/CultNetRudpOrderedDeliveryTests.cs new file mode 100644 index 00000000..1bad16f2 --- /dev/null +++ b/tests/GameCult.Networking.Tests/CultNetRudpOrderedDeliveryTests.cs @@ -0,0 +1,127 @@ +#nullable enable +using System.Collections.Generic; +using System.Linq; +using System.Text; +using NUnit.Framework; + +namespace GameCult.Networking.Tests +{ + /// + /// Ordered delivery has one owner: the contiguous received-through watermark. A frame is held for + /// exactly one reason, a reliable sequence below it has not arrived, and it is delivered in the call + /// that fills the last such gap, whichever channel the gap belonged to. + /// + public sealed class CultNetRudpOrderedDeliveryTests + { + private static (CultNetRudpSession Client, CultNetRudpSession Server) Handshake() + { + var client = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 410, InitialSequence = 1 }); + var server = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 410, InitialSequence = 500 }); + var accept = server.AcceptConnect(client.CreateConnect(0), 0); + client.Receive(accept, 0); + return (client, server); + } + + private static CultNetRudpSendOptions Reliable(bool ordered) => new CultNetRudpSendOptions { Reliable = true, Ordered = ordered }; + + private static CultNetRudpPacket Send(CultNetRudpSession session, string channel, string payload, bool ordered) + => session.Send(channel, Encoding.UTF8.GetBytes(payload), Reliable(ordered)); + + private static string[] Names(CultNetRudpReceiveResult result) + => result.Delivered.Select(frame => Encoding.UTF8.GetString(frame.Payload)).ToArray(); + + [Test] + public void OrderedFrameWaitsForAGapFilledByAnotherChannel() + { + var (sender, receiver) = Handshake(); + var o1 = Send(sender, "schema", "o1", true); + var u = Send(sender, "rel", "u", false); + var o2 = Send(sender, "schema", "o2", true); + var o3 = Send(sender, "schema", "o3", true); + + Assert.That(Names(receiver.Receive(o1, 1)), Is.EqualTo(new[] { "o1" })); + Assert.That(Names(receiver.Receive(o2, 2)), Is.Empty, "u is missing"); + Assert.That(Names(receiver.Receive(o3, 3)), Is.Empty, "u is missing"); + Assert.That(Names(receiver.Receive(u, 4)), Is.EqualTo(new[] { "u", "o2", "o3" })); + } + + [Test] + public void TwoOrderedChannelsReleaseEachOtherInSequenceOrder() + { + var (sender, receiver) = Handshake(); + var a1 = Send(sender, "schema", "A1", true); + var b1 = Send(sender, "other", "B1", true); + var a2 = Send(sender, "schema", "A2", true); + var a3 = Send(sender, "schema", "A3", true); + + var delivered = new List(); + delivered.AddRange(Names(receiver.Receive(a1, 1))); + delivered.AddRange(Names(receiver.Receive(a2, 2))); + delivered.AddRange(Names(receiver.Receive(a3, 3))); + Assert.That(delivered, Is.EqualTo(new[] { "A1" }), "B1 is missing, so A2 and A3 wait"); + Assert.That(Names(receiver.Receive(b1, 4)), Is.EqualTo(new[] { "B1", "A2", "A3" })); + } + + [Test] + public void ChannelFirstUsedAfterOtherTrafficLosesNothing() + { + var (sender, receiver) = Handshake(); + var c1 = Send(sender, "late", "C1", true); + var c2 = Send(sender, "late", "C2", true); + var x = Send(sender, "schema", "X", true); + + Assert.That(Names(receiver.Receive(x, 1)), Is.Empty, "C1 is missing"); + Assert.That(Names(receiver.Receive(c2, 2)), Is.Empty, "C1 is missing"); + Assert.That(Names(receiver.Receive(c1, 3)), Is.EqualTo(new[] { "C1", "C2", "X" })); + } + + [Test] + public void AcceptSeedsTheWatermarkOfTheConnectingSide() + { + var (client, server) = Handshake(); + var s1 = Send(server, "schema", "s1", true); + var s2 = Send(server, "schema", "s2", true); + + Assert.That(Names(client.Receive(s2, 1)), Is.Empty); + Assert.That(Names(client.Receive(s1, 2)), Is.EqualTo(new[] { "s1", "s2" })); + } + + [Test] + public void ConnectSeedsTheWatermarkOfTheAcceptingSide() + { + var (client, server) = Handshake(); + var first = Send(client, "schema", "first", true); + var second = Send(client, "schema", "second", true); + + Assert.That(Names(server.Receive(second, 1)), Is.Empty); + Assert.That(Names(server.Receive(first, 2)), Is.EqualTo(new[] { "first", "second" })); + } + + [Test] + public void DuplicateOfAHeldFrameIsNotDeliveredTwice() + { + var (sender, receiver) = Handshake(); + var s1 = Send(sender, "schema", "s1", true); + var s2 = Send(sender, "schema", "s2", true); + + Assert.That(Names(receiver.Receive(s2, 1)), Is.Empty); + Assert.That(Names(receiver.Receive(s2, 2)), Is.Empty); + Assert.That(Names(receiver.Receive(s1, 3)), Is.EqualTo(new[] { "s1", "s2" })); + Assert.That(Names(receiver.Receive(s2, 4)), Is.Empty); + } + + [Test] + public void ResetPeerStateForgetsHeldFramesAndTheWatermark() + { + var (sender, receiver) = Handshake(); + var s1 = Send(sender, "schema", "s1", true); + var s2 = Send(sender, "schema", "s2", true); + Assert.That(Names(receiver.Receive(s2, 1)), Is.Empty); + + receiver.ResetPeerState(); + receiver.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 410, Sequence = 1, ChannelId = "control" }, 2); + + Assert.That(Names(receiver.Receive(s1, 3)), Is.EqualTo(new[] { "s1" }), "the held s2 died with the reset"); + } + } +} diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index e32ca91a..63379143 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1032,8 +1032,8 @@ namespace GameCult.Networking.Tests ConnectionId = 123, InitialSequence = 100 }); - sender.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 123, Sequence = 90, ChannelId = "control" }); - receiver.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 123, Sequence = 91, ChannelId = "control" }); + var accept = receiver.AcceptConnect(sender.CreateConnect(0), 0); + sender.Receive(accept); var first = sender.Send("schema", Encoding.UTF8.GetBytes("first"), new CultNetRudpSendOptions { Reliable = true, Ordered = true }); var second = sender.Send("schema", Encoding.UTF8.GetBytes("second"), new CultNetRudpSendOptions { Reliable = true, Ordered = true });