From 0736ae4824e7effab73cdd3619c0bcea29caad5d Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 26 Aug 2026 17:16:57 +0200 Subject: [PATCH] Separate RUDP reliable and lossy sequence domains --- packages/cultnet-rs/src/rudp.rs | 65 ++++++--- packages/cultnet-rs/tests/cultnet.rs | 127 ++++++++++++++++++ packages/cultnet-ts/src/rudp.ts | 55 +++++--- packages/cultnet-ts/test/cultnet.test.ts | 47 +++++++ src/GameCult.Networking/CultNetTransport.cs | 68 ++++++---- .../NetworkingTests.cs | 118 +++++++++------- 6 files changed, 372 insertions(+), 108 deletions(-) diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 674b2996..ab16bf0d 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -159,11 +159,13 @@ pub struct CultNetRudpSession { max_pending_fragment_sets: usize, initial_sequence: u32, next_sequence: u32, + next_sequenced_by_channel: BTreeMap, next_fragment_id: u16, connected: bool, last_received_at_ms: Option, highest_received_sequence: Option, received_sequences: BTreeSet, + latest_sequenced_by_channel: BTreeMap, pending_reliable: BTreeMap, queued_reliable: VecDeque, ordered_next_sequence_by_channel: BTreeMap, @@ -182,11 +184,13 @@ impl CultNetRudpSession { max_pending_fragment_sets: 64, initial_sequence: options.initial_sequence, next_sequence: options.initial_sequence, + next_sequenced_by_channel: BTreeMap::new(), next_fragment_id: 1, connected: false, last_received_at_ms: None, highest_received_sequence: None, received_sequences: BTreeSet::new(), + latest_sequenced_by_channel: BTreeMap::new(), pending_reliable: BTreeMap::new(), queued_reliable: VecDeque::new(), ordered_next_sequence_by_channel: BTreeMap::new(), @@ -290,11 +294,13 @@ impl CultNetRudpSession { pub fn reset_peer_state(&mut self) { self.session_scope = Uuid::new_v4(); self.next_sequence = self.initial_sequence; + self.next_sequenced_by_channel.clear(); self.next_fragment_id = 1; self.connected = false; self.last_received_at_ms = None; self.highest_received_sequence = None; self.received_sequences.clear(); + self.latest_sequenced_by_channel.clear(); self.pending_reliable.clear(); self.queued_reliable.clear(); self.ordered_next_sequence_by_channel.clear(); @@ -377,6 +383,9 @@ impl CultNetRudpSession { "Cannot send RUDP data before the session is connected" )); } + if options.ordered && !options.reliable { + return Err(anyhow!("RUDP ordered delivery requires reliability")); + } self.require_payload_size(payload.len())?; if let Some(max_fragment_bytes) = max_fragment_bytes { @@ -459,7 +468,6 @@ impl CultNetRudpSession { } if packet.packet_type == CultNetRudpPacketType::Ping { - self.remember_received(packet.sequence); return Ok(CultNetRudpReceiveResult { delivered: Vec::new(), ready_to_send, @@ -481,9 +489,6 @@ impl CultNetRudpSession { if packet.packet_type == CultNetRudpPacketType::Ack || packet.packet_type == CultNetRudpPacketType::Pong { - if packet.packet_type == CultNetRudpPacketType::Pong { - self.remember_received(packet.sequence); - } return Ok(CultNetRudpReceiveResult { delivered: Vec::new(), ready_to_send, @@ -500,7 +505,6 @@ impl CultNetRudpSession { } if packet.packet_type == CultNetRudpPacketType::Disconnect { - self.remember_received(packet.sequence); self.connected = false; return Ok(CultNetRudpReceiveResult { delivered: Vec::new(), @@ -525,12 +529,15 @@ impl CultNetRudpSession { }); } - let duplicate = self.received_sequences.contains(&packet.sequence) - || self.highest_received_sequence.is_some_and(|highest| { - packet.sequence < highest - && highest - packet.sequence >= RUDP_RECEIVED_SEQUENCE_WINDOW as u32 - }); - self.remember_received(packet.sequence); + let duplicate = packet.reliable + && (self.received_sequences.contains(&packet.sequence) + || self.highest_received_sequence.is_some_and(|highest| { + packet.sequence < highest + && highest - packet.sequence >= RUDP_RECEIVED_SEQUENCE_WINDOW as u32 + })); + if packet.reliable { + self.remember_received(packet.sequence); + } if duplicate { return Ok(CultNetRudpReceiveResult { delivered: Vec::new(), @@ -556,6 +563,19 @@ impl CultNetRudpSession { }; let delivered = if ordered { self.deliver_ordered(frame, next_sequence, expected_sequence_if_uninitialized) + } else if packet.sequenced { + let latest = self + .latest_sequenced_by_channel + .get(&frame.channel_id) + .copied(); + let newest_sequence = next_sequence.saturating_sub(1); + if latest.is_some_and(|latest| newest_sequence <= latest) { + Vec::new() + } else { + self.latest_sequenced_by_channel + .insert(frame.channel_id.clone(), newest_sequence); + vec![frame] + } } else { vec![frame] }; @@ -700,11 +720,24 @@ impl CultNetRudpSession { fragment_index: u16, fragment_count: u16, ) -> CultNetRudpPacket { - let sequence = self.next_sequence; - self.next_sequence = self - .next_sequence - .checked_add(1) - .expect("sequence overflow"); + let sequence = if reliable { + let sequence = self.next_sequence; + self.next_sequence = self + .next_sequence + .checked_add(1) + .expect("reliable sequence overflow"); + sequence + } else if sequenced { + let next = self + .next_sequenced_by_channel + .entry(channel_id.to_string()) + .or_insert(1); + let sequence = *next; + *next = next.checked_add(1).expect("sequenced channel overflow"); + sequence + } else { + 0 + }; let (ack, ack_mask) = self.ack_state(); CultNetRudpPacket { packet_type, diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index a66411f0..293b63a6 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -757,6 +757,133 @@ fn rudp_session_pings_and_detects_receive_timeout() -> Result<()> { Ok(()) } +#[test] +fn rudp_lossy_packets_cannot_create_reliable_ordered_gaps() -> Result<()> { + let mut client = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 198, + initial_sequence: 1, + ..CultNetRudpSessionOptions::default() + }); + let mut server = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 198, + initial_sequence: 100, + ..CultNetRudpSessionOptions::default() + }); + let connect = client.create_connect(0, Vec::new())?; + let accept = server.accept_connect(&connect, 1, Vec::new())?; + client.receive(&accept, 2)?; + + let ping = client.create_ping(b"pulse".to_vec()); + let pong = server + .receive(&ping, 3)? + .reply + .expect("ping should produce pong"); + let realtime = server.send( + "realtime", + b"discarded realtime".to_vec(), + CultNetRudpSendOptions::default(), + )?; + let latest = server.send( + "latest", + b"discarded latest state".to_vec(), + CultNetRudpSendOptions { + sequenced: true, + ..CultNetRudpSendOptions::default() + }, + )?; + let schema = server.send( + "schema", + b"committed response".to_vec(), + CultNetRudpSendOptions { + reliable: true, + ordered: true, + ..CultNetRudpSendOptions::default() + }, + )?; + + assert_eq!(ping.sequence, 0); + assert_eq!(pong.sequence, 0); + assert_eq!(realtime.sequence, 0); + assert_eq!(latest.sequence, 1); + assert_eq!(schema.sequence, accept.sequence + 1); + assert_eq!( + client.receive(&schema, 4)?.delivered[0].payload, + b"committed response" + ); + Ok(()) +} + +#[test] +fn rudp_unreliable_sequenced_delivery_is_scoped_to_its_channel() -> Result<()> { + let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 197, + initial_sequence: 50, + ..CultNetRudpSessionOptions::default() + }); + let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 197, + initial_sequence: 100, + ..CultNetRudpSessionOptions::default() + }); + let connect = sender.create_connect(0, Vec::new())?; + let accept = receiver.accept_connect(&connect, 1, Vec::new())?; + sender.receive(&accept, 2)?; + let options = CultNetRudpSendOptions { + sequenced: true, + ..CultNetRudpSendOptions::default() + }; + let older = sender.send("latest", b"older".to_vec(), options.clone())?; + let newer = sender.send("latest", b"newer".to_vec(), options)?; + + assert_eq!(older.sequence, 1); + assert_eq!(newer.sequence, 2); + assert_eq!(receiver.receive(&newer, 3)?.delivered[0].payload, b"newer"); + assert!(receiver.receive(&older, 4)?.delivered.is_empty()); + Ok(()) +} + +#[test] +fn rudp_rejects_unreliable_ordered_delivery() -> Result<()> { + let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 196, + ..CultNetRudpSessionOptions::default() + }); + session.receive( + &CultNetRudpPacket { + packet_type: CultNetRudpPacketType::Accept, + connection_id: 196, + sequence: 50, + 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 error = session + .send( + "schema", + b"cannot order what will not retransmit".to_vec(), + CultNetRudpSendOptions { + ordered: true, + ..CultNetRudpSendOptions::default() + }, + ) + .unwrap_err(); + assert!( + error + .to_string() + .contains("ordered delivery requires reliability") + ); + Ok(()) +} + #[test] fn rudp_session_bounds_pending_reliable_packets_before_enqueue() -> Result<()> { let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions { diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index 51b5cc5b..b6cae87d 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -147,12 +147,14 @@ export class CultNetRudpSession { readonly connectionId: number; readonly resendDelayMs: number; #nextSequence: number; + readonly #nextSequencedByChannel = new Map(); #nextFragmentId = 1; #connected = false; readonly #maxPendingReliablePackets: number | undefined; #lastReceivedAtMs: number | undefined; #highestReceivedSequence: number | undefined; readonly #receivedSequences = new Set(); + readonly #latestSequencedByChannel = new Map(); readonly #pendingReliable = new Map(); readonly #queuedReliable: CultNetRudpPacket[] = []; readonly #orderedNextSequenceByChannel = new Map(); @@ -194,6 +196,8 @@ export class CultNetRudpSession { this.#lastReceivedAtMs = undefined; this.#highestReceivedSequence = undefined; this.#receivedSequences.clear(); + this.#nextSequencedByChannel.clear(); + this.#latestSequencedByChannel.clear(); this.#pendingReliable.clear(); this.#queuedReliable.splice(0); this.#orderedNextSequenceByChannel.clear(); @@ -271,6 +275,9 @@ export class CultNetRudpSession { if (!this.#connected) { throw new Error("Cannot send RUDP data before the session is connected."); } + if (options.ordered && !options.reliable) { + throw new Error("RUDP ordered delivery requires reliability."); + } const maxFragmentBytes = options.maxFragmentBytes; if (maxFragmentBytes !== undefined && maxFragmentBytes <= 0) { @@ -336,7 +343,6 @@ export class CultNetRudpSession { } if (packet.packetType === "ping") { - this.#rememberReceived(packet.sequence); return { delivered: [], readyToSend, @@ -349,9 +355,8 @@ export class CultNetRudpSession { } if (packet.packetType === "ack" || packet.packetType === "pong") { - if (packet.packetType === "pong") this.#rememberReceived(packet.sequence); return { - delivered: this.#drainUnblockedOrdered(), + delivered: [], readyToSend, pong: packet.packetType === "pong", pongPayload: packet.packetType === "pong" ? packet.payload ?? new Uint8Array() : undefined, @@ -359,7 +364,6 @@ export class CultNetRudpSession { } if (packet.packetType === "disconnect") { - this.#rememberReceived(packet.sequence); this.#connected = false; return { delivered: [], @@ -373,11 +377,12 @@ export class CultNetRudpSession { return { delivered: [], readyToSend }; } - const isDuplicate = this.#receivedSequences.has(packet.sequence) - || (this.#highestReceivedSequence !== undefined - && packet.sequence < this.#highestReceivedSequence - && this.#highestReceivedSequence - packet.sequence >= RUDP_RECEIVED_SEQUENCE_WINDOW); - this.#rememberReceived(packet.sequence); + const isDuplicate = (packet.reliable ?? false) + && (this.#receivedSequences.has(packet.sequence) + || (this.#highestReceivedSequence !== undefined + && packet.sequence < this.#highestReceivedSequence + && this.#highestReceivedSequence - packet.sequence >= RUDP_RECEIVED_SEQUENCE_WINDOW)); + if (packet.reliable ?? false) this.#rememberReceived(packet.sequence); if (isDuplicate) { return { delivered: [], readyToSend }; } @@ -387,6 +392,16 @@ export class CultNetRudpSession { return { delivered: [], readyToSend }; } + if (!reassembled.ordered && (packet.sequenced ?? false)) { + const newestSequence = reassembled.nextSequence - 1; + const latestSequence = this.#latestSequencedByChannel.get(reassembled.frame.channelId); + if (latestSequence !== undefined && newestSequence <= latestSequence) { + return { delivered: [], readyToSend }; + } + this.#latestSequencedByChannel.set(reassembled.frame.channelId, newestSequence); + return { delivered: [reassembled.frame], readyToSend }; + } + if (!reassembled.ordered) { return { delivered: [reassembled.frame], readyToSend }; } @@ -492,8 +507,17 @@ export class CultNetRudpSession { fragmentIndex?: number; fragmentCount?: number; }): CultNetRudpPacket { - const sequence = this.#nextSequence; - this.#nextSequence = toUint32(this.#nextSequence + 1, "sequence"); + let sequence = 0; + if (packet.reliable ?? false) { + sequence = this.#nextSequence; + this.#nextSequence = toUint32(this.#nextSequence + 1, "reliable sequence"); + } else if (packet.sequenced ?? false) { + sequence = this.#nextSequencedByChannel.get(packet.channelId) ?? 1; + this.#nextSequencedByChannel.set( + packet.channelId, + toUint32(sequence + 1, "sequenced channel sequence"), + ); + } const { ack, ackMask } = this.#ackState(); return { packetType: packet.packetType, @@ -712,15 +736,6 @@ export class CultNetRudpSession { return delivered; } - #drainUnblockedOrdered(): CultNetRudpDeliveredFrame[] { - const delivered: CultNetRudpDeliveredFrame[] = []; - for (const channelId of this.#orderedBuffers.keys()) { - this.#skipReceivedNonChannelSequences(channelId); - delivered.push(...this.#drainOrdered(channelId)); - } - return delivered; - } - #skipReceivedNonChannelSequences(channelId: string): void { let next = this.#orderedNextSequenceByChannel.get(channelId); while (next !== undefined && this.#receivedSequences.has(next) && !this.#orderedBuffers.get(channelId)?.has(next)) { diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index 111b7361..7ffbafb9 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -553,6 +553,53 @@ test("rudp session pings and detects receive timeout", () => { assert.equal(client.connected, false); }); +test("rudp lossy packets cannot create reliable ordered gaps", () => { + const client = new CultNetRudpSession({ connectionId: 198, initialSequence: 1 }); + const server = new CultNetRudpSession({ connectionId: 198, initialSequence: 100 }); + const connect = client.createConnect(0); + const accept = server.acceptConnect(connect, 1); + client.receive(accept, 2); + + const ping = client.createPing(Buffer.from("pulse")); + const pong = server.receive(ping, 3).reply!; + const realtime = server.send("realtime", Buffer.from("discarded realtime")); + const latest = server.send("latest", Buffer.from("discarded latest state"), { sequenced: true }); + const schema = server.send("schema", Buffer.from("committed response"), { reliable: true, ordered: true }); + + assert.equal(ping.sequence, 0); + assert.equal(pong.sequence, 0); + assert.equal(realtime.sequence, 0); + assert.equal(latest.sequence, 1); + assert.equal(schema.sequence, accept.sequence + 1); + assert.deepEqual(client.receive(schema, 4).delivered.map(frame => Buffer.from(frame.payload).toString("utf8")), [ + "committed response", + ]); +}); + +test("rudp unreliable sequenced delivery is scoped to its channel", () => { + const sender = new CultNetRudpSession({ connectionId: 197, initialSequence: 50 }); + const receiver = new CultNetRudpSession({ connectionId: 197, initialSequence: 100 }); + const connect = sender.createConnect(0); + const accept = receiver.acceptConnect(connect, 1); + sender.receive(accept, 2); + const older = sender.send("latest", Buffer.from("older"), { sequenced: true }); + const newer = sender.send("latest", Buffer.from("newer"), { sequenced: true }); + + assert.equal(older.sequence, 1); + assert.equal(newer.sequence, 2); + assert.deepEqual(receiver.receive(newer, 3).delivered.map(frame => Buffer.from(frame.payload).toString("utf8")), ["newer"]); + assert.deepEqual(receiver.receive(older, 4).delivered, []); +}); + +test("rudp rejects unreliable ordered delivery", () => { + const session = new CultNetRudpSession({ connectionId: 196 }); + session.receive({ packetType: "accept", connectionId: 196, sequence: 50, ack: 0, ackMask: 0, channelId: "control" }); + assert.throws( + () => session.send("schema", Buffer.from("cannot order what will not retransmit"), { ordered: true }), + /ordered delivery requires reliability/, + ); +}); + test("rudp session bounds pending reliable packets before enqueue", () => { const session = new CultNetRudpSession({ connectionId: 102, initialSequence: 1, maxPendingReliablePackets: 2 }); session.receive({ packetType: "accept", connectionId: 102, sequence: 50, ack: 0, ackMask: 0, channelId: "control" }); diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index 6e706fbc..f549f735 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -894,12 +894,14 @@ namespace GameCult.Networking } private uint _nextSequence; + private readonly Dictionary _nextSequencedByChannel = new Dictionary(StringComparer.Ordinal); private ushort _nextFragmentId = 1; private readonly int? _maxPendingReliablePackets; private volatile bool _connected; private long? _lastReceivedAtMs; private uint? _highestReceivedSequence; private readonly HashSet _receivedSequences = new HashSet(); + private readonly Dictionary _latestSequencedByChannel = new Dictionary(StringComparer.Ordinal); private readonly object _pendingReliableGate = new object(); private readonly Dictionary _pendingReliable = new Dictionary(); private readonly Queue _queuedReliable = new Queue(); @@ -1022,6 +1024,10 @@ namespace GameCult.Networking } options ??= new CultNetRudpSendOptions(); + if (options.Ordered && !options.Reliable) + { + throw new InvalidOperationException("RUDP ordered delivery requires reliability."); + } payload ??= Array.Empty(); if (maxFragmentBytes.HasValue && maxFragmentBytes.Value <= 0) { @@ -1099,7 +1105,6 @@ namespace GameCult.Networking if (packet.PacketType == CultNetRudpPacketType.Ping) { - RememberReceived(packet.Sequence); return new CultNetRudpReceiveResult { ReadyToSend = readyToSend, @@ -1115,11 +1120,9 @@ namespace GameCult.Networking if (packet.PacketType == CultNetRudpPacketType.Ack || packet.PacketType == CultNetRudpPacketType.Pong) { - if (packet.PacketType == CultNetRudpPacketType.Pong) - RememberReceived(packet.Sequence); return new CultNetRudpReceiveResult { - Delivered = DrainUnblockedOrdered(), + Delivered = Array.Empty(), ReadyToSend = readyToSend, Pong = packet.PacketType == CultNetRudpPacketType.Pong, PongPayload = packet.PacketType == CultNetRudpPacketType.Pong @@ -1130,7 +1133,6 @@ namespace GameCult.Networking if (packet.PacketType == CultNetRudpPacketType.Disconnect) { - RememberReceived(packet.Sequence); _connected = false; return new CultNetRudpReceiveResult { @@ -1145,11 +1147,15 @@ namespace GameCult.Networking return new CultNetRudpReceiveResult { ReadyToSend = readyToSend }; } - var duplicate = _receivedSequences.Contains(packet.Sequence) - || (_highestReceivedSequence.HasValue - && packet.Sequence < _highestReceivedSequence.Value - && _highestReceivedSequence.Value - packet.Sequence >= ReceivedSequenceWindow); - RememberReceived(packet.Sequence); + var duplicate = packet.Reliable + && (_receivedSequences.Contains(packet.Sequence) + || (_highestReceivedSequence.HasValue + && packet.Sequence < _highestReceivedSequence.Value + && _highestReceivedSequence.Value - packet.Sequence >= ReceivedSequenceWindow)); + if (packet.Reliable) + { + RememberReceived(packet.Sequence); + } if (duplicate) { return new CultNetRudpReceiveResult { ReadyToSend = readyToSend }; @@ -1161,6 +1167,17 @@ namespace GameCult.Networking return new CultNetRudpReceiveResult { ReadyToSend = readyToSend }; } + if (!reassembled.Ordered && packet.Sequenced) + { + var newestSequence = reassembled.NextSequence - 1; + if (_latestSequencedByChannel.TryGetValue(reassembled.Frame.ChannelId, out var latestSequence) + && newestSequence <= latestSequence) + { + return new CultNetRudpReceiveResult { ReadyToSend = readyToSend }; + } + _latestSequencedByChannel[reassembled.Frame.ChannelId] = newestSequence; + } + return new CultNetRudpReceiveResult { ReadyToSend = readyToSend, @@ -1282,8 +1299,24 @@ namespace GameCult.Networking ushort fragmentIndex, ushort fragmentCount) { - var sequence = _nextSequence; - _nextSequence = checked(_nextSequence + 1); + uint sequence; + if (reliable) + { + sequence = _nextSequence; + _nextSequence = checked(_nextSequence + 1); + } + else if (sequenced) + { + if (!_nextSequencedByChannel.TryGetValue(channelId, out sequence)) + { + sequence = 1; + } + _nextSequencedByChannel[channelId] = checked(sequence + 1); + } + else + { + sequence = 0; + } var (ack, ackMask) = AckState(); return new CultNetRudpPacket { @@ -1563,17 +1596,6 @@ namespace GameCult.Networking return delivered; } - private IReadOnlyList DrainUnblockedOrdered() - { - var delivered = new List(); - foreach (var channelId in _orderedBuffers.Keys.ToArray()) - { - SkipReceivedNonChannelSequences(channelId); - delivered.AddRange(DrainOrdered(channelId)); - } - return delivered; - } - private void SkipReceivedNonChannelSequences(string channelId) { if (!_orderedNextSequenceByChannel.TryGetValue(channelId, out var next)) diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 9ff8953e..c065d0af 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -889,6 +889,75 @@ namespace GameCult.Networking.Tests Assert.That(client.Connected, Is.False); } + [Test] + public void RudpSession_LossyPacketsCannotCreateReliableOrderedGaps() + { + var client = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 198, InitialSequence = 1 }); + var server = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 198, InitialSequence = 100 }); + var connect = client.CreateConnect(0); + var accept = server.AcceptConnect(connect, 1); + client.Receive(accept, 2); + + var ping = client.CreatePing(Encoding.UTF8.GetBytes("pulse")); + var pong = server.Receive(ping, 3).Reply!; + var realtime = server.Send("realtime", Encoding.UTF8.GetBytes("discarded realtime")); + var latest = server.Send( + "latest", + Encoding.UTF8.GetBytes("discarded latest state"), + new CultNetRudpSendOptions { Sequenced = true }); + var schema = server.Send( + "schema", + Encoding.UTF8.GetBytes("committed response"), + new CultNetRudpSendOptions { Reliable = true, Ordered = true }); + + Assert.That(ping.Sequence, Is.Zero); + Assert.That(pong.Sequence, Is.Zero); + Assert.That(realtime.Sequence, Is.Zero); + Assert.That(latest.Sequence, Is.EqualTo(1)); + Assert.That(schema.Sequence, Is.EqualTo(accept.Sequence + 1)); + Assert.That( + client.Receive(schema, 4).Delivered.Select(frame => Encoding.UTF8.GetString(frame.Payload)).ToArray(), + Is.EqualTo(new[] { "committed response" })); + } + + [Test] + public void RudpSession_UnreliableSequencedDeliveryIsScopedToItsChannel() + { + var sender = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 197, InitialSequence = 50 }); + var receiver = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 197, InitialSequence = 100 }); + var connect = sender.CreateConnect(0); + var accept = receiver.AcceptConnect(connect, 1); + sender.Receive(accept, 2); + var older = sender.Send("latest", Encoding.UTF8.GetBytes("older"), new CultNetRudpSendOptions { Sequenced = true }); + var newer = sender.Send("latest", Encoding.UTF8.GetBytes("newer"), new CultNetRudpSendOptions { Sequenced = true }); + + Assert.That(older.Sequence, Is.EqualTo(1)); + Assert.That(newer.Sequence, Is.EqualTo(2)); + Assert.That( + receiver.Receive(newer, 3).Delivered.Select(frame => Encoding.UTF8.GetString(frame.Payload)).ToArray(), + Is.EqualTo(new[] { "newer" })); + Assert.That(receiver.Receive(older, 4).Delivered, Is.Empty); + } + + [Test] + public void RudpSession_RejectsUnreliableOrderedDelivery() + { + var session = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 196 }); + session.Receive(new CultNetRudpPacket + { + PacketType = CultNetRudpPacketType.Accept, + ConnectionId = 196, + Sequence = 50, + ChannelId = "control" + }); + var error = Assert.Throws(() => + session.Send( + "schema", + Encoding.UTF8.GetBytes("cannot order what will not retransmit"), + new CultNetRudpSendOptions { Ordered = true })); + Assert.That(error!.Message, Does.Contain("ordered delivery requires reliability")); + } + [Test] public void RudpSession_BoundsPendingReliablePacketsBeforeEnqueue() { @@ -955,55 +1024,6 @@ namespace GameCult.Networking.Tests Is.EqualTo(new[] { "second", "third" })); } - [Test] - public void RudpSession_SequencedControlPacketReleasesBufferedOrderedFrame() - { - var receiver = new CultNetRudpSession(new CultNetRudpSessionOptions - { - ConnectionId = 199, - InitialSequence = 1 - }); - receiver.Receive(new CultNetRudpPacket - { - PacketType = CultNetRudpPacketType.Accept, - ConnectionId = 199, - Sequence = 100, - ChannelId = "control" - }); - var first = receiver.Receive(new CultNetRudpPacket - { - PacketType = CultNetRudpPacketType.Data, - ConnectionId = 199, - Sequence = 101, - ChannelId = "schema", - Reliable = true, - Ordered = true, - Payload = Encoding.UTF8.GetBytes("first") - }); - var buffered = receiver.Receive(new CultNetRudpPacket - { - PacketType = CultNetRudpPacketType.Data, - ConnectionId = 199, - Sequence = 103, - ChannelId = "schema", - Reliable = true, - Ordered = true, - Payload = Encoding.UTF8.GetBytes("second") - }); - var released = receiver.Receive(new CultNetRudpPacket - { - PacketType = CultNetRudpPacketType.Pong, - ConnectionId = 199, - Sequence = 102, - ChannelId = "control" - }); - - Assert.That(first.Delivered, Has.Count.EqualTo(1)); - Assert.That(buffered.Delivered, Is.Empty); - Assert.That(released.Delivered, Has.Count.EqualTo(1)); - Assert.That(Encoding.UTF8.GetString(released.Delivered[0].Payload), Is.EqualTo("second")); - } - [Test] public void RudpSession_SkipsControlPacketsWhileOrderingSchemaPayloads() {