diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index d0779196..17fc6964 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -271,6 +271,15 @@ Current progress: flush where the runtime supports one, so callers that require confirmed transport delivery do not substitute a fixed sleep. TCP-framed profiles do not publish the field because they do not own the RUDP pending-reliable queue. +- The receive window is a contract, not a discovery. A receiver must hold 1,023 + sequences and 4 MiB of payload beyond its contiguous received watermark, and + a conforming sender never exceeds either. The sender's flow window enforces + it without a wire field: a reliable packet is admitted only while its + sequence is at most 1,023 above the lowest unacked sequence and the payload + above that sequence stays within 4 MiB. The first packet is always + admissible, so a payload larger than 4 MiB goes out alone, and packets wait + in the FIFO in order, so a small packet never overtakes a large one. C#, + Rust, TypeScript, and Python enforce it; Kotlin does not yet. - TypeScript, C#, Rust, Python, and Kotlin now share a portable `cultnet.reconnect_policy.v0` document and deterministic exponential delay helper, and RUDP transport profiles advertise that policy under diff --git a/packages/cultnet-py/src/cultnet_py/transport.py b/packages/cultnet-py/src/cultnet_py/transport.py index 355faa51..1e718aab 100644 --- a/packages/cultnet-py/src/cultnet_py/transport.py +++ b/packages/cultnet-py/src/cultnet_py/transport.py @@ -438,6 +438,10 @@ def create_rudp_transport_profile( class CultNetRudpSession: RELIABLE_SEND_WINDOW_PACKETS = 32 RECEIVED_SEQUENCE_WINDOW = 4_096 + # Flow window: a reliable packet is admitted only while its sequence is at most FLOW_WINDOW_SEQUENCES above the + # lowest unacked one and the payload above that sequence stays within FLOW_WINDOW_BYTES. + FLOW_WINDOW_SEQUENCES = 1_023 + FLOW_WINDOW_BYTES = 4 * 1024 * 1024 def __init__(self, options: CultNetRudpSessionOptions) -> None: self.connection_id = _uint32(options.connection_id, "connection_id") @@ -478,6 +482,9 @@ class CultNetRudpSession: self._received_sequences: set[int] = set() self._latest_sequenced_by_channel: dict[str, int] = {} self._pending_reliable: dict[int, _PendingReliablePacket] = {} + # Payload sizes of acknowledged reliable sequences above the lowest unacknowledged one. The receiver still + # holds those bytes behind the gap, so the flow window keeps counting them until the lowest sequence passes them. + self._acked_above_lowest: dict[int, int] = {} self._queued_reliable: deque[CultNetRudpPacket] = deque() # Every reliable sequence up to and including this one has been received # since the peer state was last reset. Only the handshake seeds it: the @@ -540,6 +547,7 @@ class CultNetRudpSession: self._awaiting_accept = False self._generation += 1 self._pending_reliable.clear() + self._acked_above_lowest.clear() self._queued_reliable.clear() def reset_peer_state(self) -> None: @@ -656,7 +664,7 @@ class CultNetRudpSession: options: CultNetRudpSendOptions | None = None, ) -> CultNetRudpPacket: resolved = options or CultNetRudpSendOptions() - if resolved.reliable and len(self._pending_reliable) >= self.RELIABLE_SEND_WINDOW_PACKETS: + if resolved.reliable and (self._queued_reliable or not self._window_admits(self._next_sequence, len(payload))): raise ValueError("RUDP reliable send window is full; receive acknowledgements before sending") return self.send_many(channel_id, payload, resolved)[0] @@ -935,22 +943,39 @@ class CultNetRudpSession: last_sent_at_ms=now_ms, ) + def _window_admits(self, sequence: int, payload_length: int) -> bool: + """Whether a reliable packet may go on the wire now: the window has a slot, and its sequence and the + payload above the lowest unacked sequence stay inside the flow window. With nothing pending, any packet + is admissible.""" + if len(self._pending_reliable) >= self.RELIABLE_SEND_WINDOW_PACKETS: + return False + if not self._pending_reliable: + return True + lowest = min(self._pending_reliable) + bytes_above = sum( + len(pending.packet.payload) for pending_sequence, pending in self._pending_reliable.items() if pending_sequence > lowest + ) + sum(self._acked_above_lowest.values()) + return sequence - lowest <= self.FLOW_WINDOW_SEQUENCES and bytes_above + payload_length <= self.FLOW_WINDOW_BYTES + def _admit_reliable_packets( self, packets: tuple[CultNetRudpPacket, ...], now_ms: int, ) -> tuple[CultNetRudpPacket, ...]: - available = max(0, self.RELIABLE_SEND_WINDOW_PACKETS - len(self._pending_reliable)) - ready = packets[:available] - for packet in ready: - self._track_reliable(packet, now_ms) - self._queued_reliable.extend(packets[available:]) - return ready + ready: list[CultNetRudpPacket] = [] + for packet in packets: + if not self._queued_reliable and self._window_admits(packet.sequence, len(packet.payload)): + self._track_reliable(packet, now_ms) + ready.append(packet) + else: + self._queued_reliable.append(packet) + return tuple(ready) def _promote_queued_reliable(self, now_ms: int) -> tuple[CultNetRudpPacket, ...]: - available = max(0, self.RELIABLE_SEND_WINDOW_PACKETS - len(self._pending_reliable)) ready: list[CultNetRudpPacket] = [] - while len(ready) < available and self._queued_reliable: + while self._queued_reliable and self._window_admits( + self._queued_reliable[0].sequence, len(self._queued_reliable[0].payload) + ): packet = self._queued_reliable.popleft() self._track_reliable(packet, now_ms) ready.append(packet) @@ -963,10 +988,21 @@ class CultNetRudpSession: raise ValueError("RUDP reliable send queue is full") def _apply_acknowledgements(self, packet: CultNetRudpPacket) -> None: - self._pending_reliable.pop(packet.ack, None) + self._acknowledge(packet.ack) for bit in range(32): if packet.ack_mask & (1 << bit): - self._pending_reliable.pop(packet.ack - bit - 1, None) + self._acknowledge(packet.ack - bit - 1) + # Acknowledged sizes stop counting once the lowest unacknowledged sequence passes them, and all of them stop + # when nothing is unacknowledged. + lowest = min(self._pending_reliable, default=None) + self._acked_above_lowest = ( + {} if lowest is None else {sequence: length for sequence, length in self._acked_above_lowest.items() if sequence > lowest} + ) + + def _acknowledge(self, sequence: int) -> None: + pending = self._pending_reliable.pop(sequence, None) + if pending is not None: + self._acked_above_lowest[sequence] = len(pending.packet.payload) def _seed_received(self, sequence: int) -> None: """The handshake's one act on the watermark: the peer's Connect or Accept diff --git a/packages/cultnet-py/tests/test_cultnet.py b/packages/cultnet-py/tests/test_cultnet.py index 0d707730..4c740ac9 100644 --- a/packages/cultnet-py/tests/test_cultnet.py +++ b/packages/cultnet-py/tests/test_cultnet.py @@ -879,6 +879,163 @@ class CultNetTests(unittest.TestCase): self.assertEqual(old_ack.ack, oldest_sequence) self.assertEqual(old_ack.ack_mask, 0) + # The sender's flow window: a reliable packet goes on the wire only while its sequence is at most 1,023 above + # the lowest unacked one and the payload above that sequence stays within 4 MiB. + FLOW_CONNECTION_ID = 0x464C4F57 + MIB = 1024 * 1024 + + def _flow_session(self, initial_sequence: int) -> CultNetRudpSession: + return CultNetRudpSession( + CultNetRudpSessionOptions(connection_id=self.FLOW_CONNECTION_ID, initial_sequence=initial_sequence) + ) + + def _connected_flow_pair( + self, sender_initial_sequence: int, receiver_initial_sequence: int + ) -> tuple[CultNetRudpSession, CultNetRudpSession]: + """A sender and a receiver that have shaken hands: the receiver's watermark is seeded by the sender's + Connect, and the sender's Connect is acknowledged, so nothing is pending on the sender.""" + sender = self._flow_session(sender_initial_sequence) + receiver = self._flow_session(receiver_initial_sequence) + sender.receive(receiver.accept_connect(sender.create_connect(0), 0), 0) + return sender, receiver + + def _connected_flow_session(self, initial_sequence: int) -> CultNetRudpSession: + return self._connected_flow_pair(initial_sequence, 900)[0] + + @staticmethod + def _flow_send(session: CultNetRudpSession, payload_bytes: int) -> tuple[CultNetRudpPacket, ...]: + return session.send_many("state", bytes(payload_bytes), CultNetRudpSendOptions(reliable=True)) + + def _flow_ack(self, sequence: int) -> CultNetRudpPacket: + return CultNetRudpPacket(CultNetRudpPacketType.ACK, self.FLOW_CONNECTION_ID, 0, sequence, 0, "control") + + def _lose_g_and_fill_the_span(self, session: CultNetRudpSession) -> CultNetRudpPacket: + g = self._flow_send(session, 1)[0] + for offset in range(1, 1024): + self.assertEqual([p.sequence for p in self._flow_send(session, 1)], [g.sequence + offset]) + session.receive(self._flow_ack(g.sequence + offset), 1) + return g + + def test_cultnet_rudp_a_lost_packet_holds_the_sender_1023_sequences_ahead_and_is_still_delivered(self) -> None: + sender, receiver = self._connected_flow_pair(1, 100) + g = self._flow_send(sender, 1)[0] + admitted_after_g = 0 + for _ in range(4200): + for packet in self._flow_send(sender, 1): + admitted_after_g += 1 + receiver.receive(packet, 1) + sender.receive(receiver.create_ack(), 1) + self.assertEqual(admitted_after_g, 1023) + self.assertEqual(sender.queued_reliable_packet_count, 4200 - 1023) + + retransmit = next(p for p in sender.due_resends(1000) if p.sequence == g.sequence) + delivered = receiver.receive(retransmit, 1000).delivered + self.assertEqual([frame.sequence for frame in delivered], [g.sequence]) + + promoted = sender.receive(receiver.create_ack_for_received(g.sequence), 1001).ready_to_send + self.assertEqual(promoted[0].sequence, g.sequence + 1024) + + def test_cultnet_rudp_a_direct_send_past_the_span_is_refused_without_consuming_a_sequence(self) -> None: + sender = self._connected_flow_session(1) + g = self._lose_g_and_fill_the_span(sender) + with self.assertRaises(ValueError): + sender.send("state", bytes(1), CultNetRudpSendOptions(reliable=True)) + self.assertEqual(sender.queued_reliable_packet_count, 0) + sender.receive(self._flow_ack(g.sequence), 2) + following = sender.send("state", bytes(1), CultNetRudpSendOptions(reliable=True)) + self.assertEqual(following.sequence, g.sequence + 1024) + + def test_cultnet_rudp_payload_above_the_lowest_unacked_sequence_is_bounded_by_4_mib(self) -> None: + sender = self._connected_flow_session(1) + g = self._flow_send(sender, self.MIB)[0] + # The lowest unacked packet's own bytes do not count: four more MiB fit above it. + for _ in range(4): + self.assertEqual(len(self._flow_send(sender, self.MIB)), 1) + self.assertEqual(len(self._flow_send(sender, self.MIB)), 0) + self.assertEqual(len(self._flow_send(sender, self.MIB)), 0) + self.assertEqual(sender.queued_reliable_packet_count, 2) + + # Acking g moves the floor up one packet: exactly one MiB more fits. + promoted = sender.receive(self._flow_ack(g.sequence), 1).ready_to_send + self.assertEqual([p.sequence for p in promoted], [g.sequence + 5]) + self.assertEqual(sender.queued_reliable_packet_count, 1) + + def test_cultnet_rudp_a_packet_larger_than_the_window_goes_out_alone_and_waits_behind_anything_unacked(self) -> None: + self.assertEqual(len(self._flow_send(self._connected_flow_session(1), 5 * self.MIB)), 1) + sender = self._connected_flow_session(1) + g = self._flow_send(sender, 1)[0] + self.assertEqual(len(self._flow_send(sender, 5 * self.MIB)), 0) + promoted = sender.receive(self._flow_ack(g.sequence), 1).ready_to_send + self.assertEqual([p.sequence for p in promoted], [g.sequence + 1]) + + def test_cultnet_rudp_a_small_packet_does_not_overtake_a_queued_large_one(self) -> None: + sender = self._connected_flow_session(1) + g = self._flow_send(sender, 1)[0] + self.assertEqual(len(self._flow_send(sender, 3 * self.MIB)), 1) + self.assertEqual(len(self._flow_send(sender, 2 * self.MIB)), 0) + # A direct send refuses instead of queueing behind it, and takes no sequence. + with self.assertRaises(ValueError): + sender.send("state", bytes(1), CultNetRudpSendOptions(reliable=True)) + self.assertEqual(sender.queued_reliable_packet_count, 1) + # One byte would fit above g, but the 2 MiB packet is ahead of it. + self.assertEqual(len(self._flow_send(sender, 1)), 0) + promoted = sender.receive(self._flow_ack(g.sequence), 1).ready_to_send + self.assertEqual([p.sequence for p in promoted], [g.sequence + 2, g.sequence + 3]) + + def test_cultnet_rudp_acknowledged_bytes_above_a_lost_packet_still_count_against_4_mib(self) -> None: + sender = self._connected_flow_session(1) + g = self._flow_send(sender, 1)[0] + # g is lost. Each 1 MiB frame above it is delivered and acknowledged, and the receiver still holds it behind + # the gap, so it keeps counting. + for offset in range(1, 5): + self.assertEqual([p.sequence for p in self._flow_send(sender, self.MIB)], [g.sequence + offset]) + sender.receive(self._flow_ack(g.sequence + offset), 1) + self.assertEqual(len(self._flow_send(sender, self.MIB)), 0) + self.assertEqual(sender.queued_reliable_packet_count, 1) + with self.assertRaises(ValueError): + sender.send("state", bytes(self.MIB), CultNetRudpSendOptions(reliable=True)) + + # g arrives: nothing is held behind a gap any more, and the queue drains. + promoted = sender.receive(self._flow_ack(g.sequence), 2).ready_to_send + self.assertEqual([p.sequence for p in promoted], [g.sequence + 5]) + # The bytes above the old gap no longer count: another MiB fits. + self.assertEqual([p.sequence for p in self._flow_send(sender, self.MIB)], [g.sequence + 6]) + + def test_cultnet_rudp_acknowledged_bytes_count_against_the_lowest_unacked_sequence_as_it_advances(self) -> None: + sender = self._connected_flow_session(1) + g = self._flow_send(sender, 1)[0] + h = self._flow_send(sender, 1)[0] + for _ in range(3): + self._flow_send(sender, self.MIB) + # g and the 3 MiB above h are acknowledged; h is lost. + for offset in (2, 3, 4, 0): + sender.receive(self._flow_ack(g.sequence + offset), 1) + self.assertEqual([p.sequence for p in self._flow_send(sender, self.MIB)], [g.sequence + 5]) + # 4 MiB sit above h, acknowledged or not: one more byte does not fit. + self.assertEqual(len(self._flow_send(sender, 1)), 0) + # h arrives, the floor moves to g + 5, and the byte fits. + promoted = sender.receive(self._flow_ack(h.sequence), 2).ready_to_send + self.assertEqual([p.sequence for p in promoted], [g.sequence + 6]) + + def test_cultnet_rudp_a_new_generation_on_the_server_does_not_inherit_acknowledged_bytes(self) -> None: + # A server that accepts a fresh Connect may send before its Accept is acknowledged, so no acknowledgement of + # the new generation has moved the floor yet. Bytes the ended generation had acknowledged above its own lost + # packet belong to no receiver's hold any more and must not count. + server = self._flow_session(500) + accept = server.accept_connect(self._flow_session(1).create_connect(0), 0) + server.receive(self._flow_ack(accept.sequence), 0) + g = self._flow_send(server, 1)[0] + for offset in range(1, 5): + self.assertEqual([p.sequence for p in self._flow_send(server, self.MIB)], [g.sequence + offset]) + server.receive(self._flow_ack(g.sequence + offset), 1) + self.assertEqual(len(self._flow_send(server, self.MIB)), 0) + + next_accept = server.accept_connect(self._flow_session(200_000).create_connect(2), 2) + self.assertEqual(next_accept.packet_type, CultNetRudpPacketType.ACCEPT) + # The Accept is unacknowledged and the only thing pending: 1 MiB fits above it. + sent = server.send("state", bytes(self.MIB), CultNetRudpSendOptions(reliable=True)) + self.assertEqual(sent.sequence, next_accept.sequence + 1) + def test_cultnet_rudp_socket_transport_handshakes_and_carries_reliable_ordered_schema_frames(self) -> None: server_socket = bind_udp_socket() client_socket = bind_udp_socket() diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 3ee6af66..9e8c4a1c 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -53,6 +53,13 @@ const RUDP_RECEIVE_AHEAD_WINDOW: u32 = 1_024; const RUDP_CONNECT_ATTEMPT_MS: u64 = 3_000; const RUDP_MAX_ORDERED_BUFFERED_FRAMES: usize = 1_024; const RUDP_MAX_ORDERED_BUFFERED_BYTES: usize = 4 * 1024 * 1024; +/// The sender's flow window: a reliable packet goes on the wire only while its +/// sequence is at most this far above the lowest unacked one, and the payload +/// above that sequence stays within the byte bound. A receiver holds +/// `RUDP_RECEIVE_AHEAD_WINDOW - 1` sequences and `RUDP_MAX_ORDERED_BUFFERED_BYTES` +/// beyond its watermark, so a sender that obeys this never overruns it. +const RUDP_FLOW_WINDOW_SEQUENCES: u32 = RUDP_RECEIVE_AHEAD_WINDOW - 1; +const RUDP_FLOW_WINDOW_BYTES: usize = RUDP_MAX_ORDERED_BUFFERED_BYTES; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum CultNetRudpPacketType { @@ -244,6 +251,11 @@ pub struct CultNetRudpSession { received_sequences: BTreeSet, latest_sequenced_by_channel: BTreeMap, pending_reliable: BTreeMap, + /// Payload sizes of acknowledged reliable sequences above the lowest + /// unacknowledged one. The receiver still holds those bytes behind the gap, + /// so the flow window keeps counting them until the lowest sequence + /// advances past them. + acked_above_lowest: BTreeMap, queued_reliable: VecDeque<(CultNetRudpPacket, Option)>, reliable_packets_expired: u64, /// Every reliable sequence up to and including this one has been received @@ -284,6 +296,7 @@ impl CultNetRudpSession { received_sequences: BTreeSet::new(), latest_sequenced_by_channel: BTreeMap::new(), pending_reliable: BTreeMap::new(), + acked_above_lowest: BTreeMap::new(), queued_reliable: VecDeque::new(), reliable_packets_expired: 0, received_through: None, @@ -368,6 +381,7 @@ impl CultNetRudpSession { self.awaiting_accept = false; self.generation += 1; self.pending_reliable.clear(); + self.acked_above_lowest.clear(); self.queued_reliable.clear(); } @@ -560,8 +574,14 @@ impl CultNetRudpSession { payload: Vec, options: CultNetRudpSendOptions, ) -> Result { + // Measure the window as `send_many` will: without the deadline-passed + // sends it is about to reclaim. + if options.reliable { + self.purge_expired_reliable(options.now_ms); + } if options.reliable - && self.pending_reliable.len() >= CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS + && (!self.queued_reliable.is_empty() + || !self.window_admits(self.next_sequence, payload.len())) { return Err(anyhow!( "RUDP reliable send window is full; receive acknowledgements before sending" @@ -1073,22 +1093,47 @@ impl CultNetRudpSession { self.queued_reliable .retain(|(_, expires_at_ms)| expires_at_ms.is_none_or(|at| now_ms <= at)); let after = self.pending_reliable.len() + self.queued_reliable.len(); + if after != before { + self.forget_acked_below_lowest(); + } self.reliable_packets_expired = self .reliable_packets_expired .saturating_add((before - after) as u64); } + /// Whether a reliable packet may go on the wire now: the window has a slot, + /// and its sequence and the payload above the lowest unacked sequence stay + /// inside the flow window. With nothing pending, any packet is admissible. + fn window_admits(&self, sequence: u32, payload_len: usize) -> bool { + if self.pending_reliable.len() >= CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS { + return false; + } + let Some((&lowest, _)) = self.pending_reliable.first_key_value() else { + return true; + }; + let pending_bytes: usize = self + .pending_reliable + .iter() + .filter(|(pending_sequence, _)| **pending_sequence > lowest) + .map(|(_, pending)| pending.packet.payload.len()) + .sum(); + let acked_bytes: usize = self.acked_above_lowest.values().sum(); + let bytes_above = pending_bytes + acked_bytes; + sequence.saturating_sub(lowest) <= RUDP_FLOW_WINDOW_SEQUENCES + && bytes_above + payload_len <= RUDP_FLOW_WINDOW_BYTES + } + fn admit_reliable_packets( &mut self, packets: Vec, now_ms: u64, expires_at_ms: Option, ) -> Vec { - let available = - CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS.saturating_sub(self.pending_reliable.len()); - let mut ready = Vec::with_capacity(available.min(packets.len())); + let mut ready = Vec::new(); for packet in packets { - if ready.len() < available { + if self.queued_reliable.is_empty() + && self.window_admits(packet.sequence, packet.payload.len()) + { self.track_reliable(packet.clone(), now_ms, expires_at_ms); ready.push(packet); } else { @@ -1099,10 +1144,12 @@ impl CultNetRudpSession { } fn promote_queued_reliable(&mut self, now_ms: u64) -> Vec { - let available = - CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS.saturating_sub(self.pending_reliable.len()); - let mut ready = Vec::with_capacity(available.min(self.queued_reliable.len())); - for _ in 0..available { + let mut ready = Vec::new(); + while self + .queued_reliable + .front() + .is_some_and(|(packet, _)| self.window_admits(packet.sequence, packet.payload.len())) + { let Some((packet, expires_at_ms)) = self.queued_reliable.pop_front() else { break; }; @@ -1133,12 +1180,29 @@ impl CultNetRudpSession { } fn apply_acknowledgements(&mut self, packet: &CultNetRudpPacket) { - self.pending_reliable.remove(&packet.ack); + self.acknowledge(packet.ack); for bit in 0..32 { if (packet.ack_mask & (1_u32 << bit)) != 0 && packet.ack > bit { - self.pending_reliable.remove(&(packet.ack - bit - 1)); + self.acknowledge(packet.ack - bit - 1); } } + self.forget_acked_below_lowest(); + } + + fn acknowledge(&mut self, sequence: u32) { + if let Some(pending) = self.pending_reliable.remove(&sequence) { + self.acked_above_lowest + .insert(sequence, pending.packet.payload.len()); + } + } + + /// Acknowledged sizes stop counting once the lowest unacknowledged + /// sequence passes them, and all of them stop when nothing is unacknowledged. + fn forget_acked_below_lowest(&mut self) { + match self.pending_reliable.first_key_value() { + Some((&lowest, _)) => self.acked_above_lowest.retain(|sequence, _| *sequence > lowest), + None => self.acked_above_lowest.clear(), + } } /// The handshake's one act on the watermark: the peer's Connect or Accept diff --git a/packages/cultnet-rs/tests/rudp_connect_generation.rs b/packages/cultnet-rs/tests/rudp_connect_generation.rs index b456162c..92f83322 100644 --- a/packages/cultnet-rs/tests/rudp_connect_generation.rs +++ b/packages/cultnet-rs/tests/rudp_connect_generation.rs @@ -575,9 +575,30 @@ fn a_flush_that_a_timeout_ended_names_the_timeout() -> Result<()> { // Hold-buffer boundaries: 1,024 frames and 64 channels are held; one more is // refused. +/// An acknowledgement that names only `sequence`. +fn ack_naming(sequence: u32) -> CultNetRudpPacket { + CultNetRudpPacket { + packet_type: CultNetRudpPacketType::Ack, + connection_id: CONNECTION_ID, + sequence: 0, + ack: sequence, + ack_mask: 0, + channel_id: "control".into(), + reliable: false, + ordered: false, + sequenced: false, + fragment_id: 0, + fragment_index: 0, + fragment_count: 0, + payload: Vec::new(), + } +} + /// A connected pair with one sequence the receiver will never see, so every -/// later ordered frame waits behind it. The sender's window is kept open by -/// acknowledging what the receiver has. +/// later ordered frame waits behind it. The sender's flow window would stop it +/// 1,023 sequences above its lowest unacked packet, so the missing one is +/// acknowledged by hand (the receiver never saw it) and the rest by what the +/// receiver has: the receiver's hold buffer is what this probes, not the sender. fn held_behind_a_gap( frames: usize, channel_of: impl Fn(usize) -> String, @@ -585,7 +606,8 @@ fn held_behind_a_gap( let mut sender = session(1); let mut receiver = session(500); handshake(&mut sender, &mut receiver)?; - let _missing = sender.send("gap", b"lost".to_vec(), ordered())?; + let missing = sender.send("gap", b"lost".to_vec(), ordered())?; + sender.receive(&ack_naming(missing.sequence), 1)?; let mut held = 0; for index in 0..frames { let packet = sender.send(&channel_of(index), vec![0], ordered())?; diff --git a/packages/cultnet-rs/tests/rudp_flow_window.rs b/packages/cultnet-rs/tests/rudp_flow_window.rs new file mode 100644 index 00000000..9f10eb50 --- /dev/null +++ b/packages/cultnet-rs/tests/rudp_flow_window.rs @@ -0,0 +1,310 @@ +//! The sender's flow window: a reliable packet goes on the wire only while its +//! sequence is at most 1,023 above the lowest unacked one and the payload above +//! that sequence stays within 4 MiB. Everything else waits in order in the queue. + +use anyhow::Result; +use cultnet_rs::*; + +const CONNECTION_ID: u32 = 0x464c_4f57; +const MIB: usize = 1024 * 1024; + +fn session(initial_sequence: u32) -> CultNetRudpSession { + CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: CONNECTION_ID, + initial_sequence: Some(initial_sequence), + resend_delay_ms: 25, + max_pending_reliable_packets: None, + }) +} + +/// A sender and a receiver that have shaken hands: the receiver's watermark is +/// seeded by the sender's Connect, and the sender's Connect is acknowledged, so +/// nothing is pending on the sender. +fn connected_pair( + sender_initial_sequence: u32, + receiver_initial_sequence: u32, +) -> Result<(CultNetRudpSession, CultNetRudpSession)> { + let mut sender = session(sender_initial_sequence); + let mut receiver = session(receiver_initial_sequence); + let connect = sender.create_connect(0, Vec::new())?; + let accept = receiver.accept_connect(&connect, 0, Vec::new())?; + sender.receive(&accept, 0)?; + Ok((sender, receiver)) +} + +fn connected(initial_sequence: u32) -> Result { + Ok(connected_pair(initial_sequence, 900)?.0) +} + +fn reliable() -> CultNetRudpSendOptions { + CultNetRudpSendOptions { + reliable: true, + ..Default::default() + } +} + +/// What `send_many` puts on the wire now; the rest of the packet waits queued. +fn send(session: &mut CultNetRudpSession, payload_bytes: usize) -> Result> { + session.send_many("state", vec![7; payload_bytes], reliable(), None) +} + +fn ack_for(sequence: u32) -> CultNetRudpPacket { + CultNetRudpPacket { + packet_type: CultNetRudpPacketType::Ack, + connection_id: CONNECTION_ID, + sequence: 0, + ack: sequence, + ack_mask: 0, + channel_id: "control".into(), + reliable: false, + ordered: false, + sequenced: false, + fragment_id: 0, + fragment_index: 0, + fragment_count: 0, + payload: Vec::new(), + } +} + +fn sequences(packets: &[CultNetRudpPacket]) -> Vec { + packets.iter().map(|packet| packet.sequence).collect() +} + +/// Sends `g`, then acks every packet up to and including the one 1,023 above +/// it, leaving `g` the only unacked sequence. Returns `g`. +fn lose_g_and_fill_the_span(session: &mut CultNetRudpSession) -> Result { + let g = send(session, 1)?.remove(0); + for offset in 1..=1_023u32 { + let sent = send(session, 1)?; + assert_eq!(sequences(&sent), vec![g.sequence + offset]); + session.receive(&ack_for(g.sequence + offset), 1)?; + } + Ok(g) +} + +#[test] +fn a_lost_packet_holds_the_sender_1023_sequences_ahead_and_is_still_delivered() -> Result<()> { + let (mut sender, mut receiver) = connected_pair(1, 100)?; + let g = send(&mut sender, 1)?.remove(0); + // g is lost. Everything after it is delivered and acked, until the sender stops. + let mut admitted_after_g = 0u32; + for _ in 0..4_200 { + for packet in send(&mut sender, 1)? { + admitted_after_g += 1; + receiver.receive(&packet, 1)?; + sender.receive(&receiver.create_ack(), 1)?; + } + } + assert_eq!(admitted_after_g, 1_023); + assert_eq!(sender.queued_reliable_packet_count(), 4_200 - 1_023); + + let retransmit = sender + .due_resends(1_000) + .into_iter() + .find(|packet| packet.sequence == g.sequence) + .expect("g is still pending and due"); + let delivered = receiver.receive(&retransmit, 1_000)?.delivered; + assert_eq!(delivered.len(), 1); + assert_eq!(delivered[0].sequence, g.sequence); + + let promoted = sender + .receive(&receiver.create_ack_for_received(g.sequence), 1_001)? + .ready_to_send; + assert_eq!(promoted.first().map(|packet| packet.sequence), Some(g.sequence + 1_024)); + Ok(()) +} + +#[test] +fn a_direct_send_past_the_span_is_refused_without_consuming_a_sequence() -> Result<()> { + let mut sender = connected(1)?; + let g = lose_g_and_fill_the_span(&mut sender)?; + assert!(sender.send("state", vec![7], reliable()).is_err()); + assert_eq!(sender.queued_reliable_packet_count(), 0); + sender.receive(&ack_for(g.sequence), 2)?; + let next = sender.send("state", vec![7], reliable())?; + assert_eq!(next.sequence, g.sequence + 1_024); + Ok(()) +} + +#[test] +fn payload_above_the_lowest_unacked_sequence_is_bounded_by_4_mib() -> Result<()> { + let mut sender = connected(1)?; + let g = send(&mut sender, MIB)?.remove(0); + // The lowest unacked packet's own bytes do not count: four more MiB fit above it. + for _ in 0..4 { + assert_eq!(send(&mut sender, MIB)?.len(), 1); + } + assert!(send(&mut sender, MIB)?.is_empty()); + assert!(send(&mut sender, MIB)?.is_empty()); + assert_eq!(sender.queued_reliable_packet_count(), 2); + + // Acking g moves the floor up one packet: exactly one MiB more fits. + let promoted = sender.receive(&ack_for(g.sequence), 1)?.ready_to_send; + assert_eq!(sequences(&promoted), vec![g.sequence + 5]); + assert_eq!(sender.queued_reliable_packet_count(), 1); + Ok(()) +} + +#[test] +fn a_packet_larger_than_the_window_goes_out_alone_and_waits_behind_anything_unacked() -> Result<()> { + let mut sender = connected(1)?; + assert_eq!(send(&mut sender, 5 * MIB)?.len(), 1); + let mut sender = connected(1)?; + let g = send(&mut sender, 1)?.remove(0); + assert!(send(&mut sender, 5 * MIB)?.is_empty()); + let promoted = sender.receive(&ack_for(g.sequence), 1)?.ready_to_send; + assert_eq!(sequences(&promoted), vec![g.sequence + 1]); + Ok(()) +} + +#[test] +fn a_small_packet_does_not_overtake_a_queued_large_one() -> Result<()> { + let mut sender = connected(1)?; + let g = send(&mut sender, 1)?.remove(0); + assert_eq!(send(&mut sender, 3 * MIB)?.len(), 1); + assert!(send(&mut sender, 2 * MIB)?.is_empty()); + // A direct send refuses instead of queueing behind it, and takes no sequence. + assert!(sender.send("state", vec![7], reliable()).is_err()); + assert_eq!(sender.queued_reliable_packet_count(), 1); + // One byte would fit above g, but the 2 MiB packet is ahead of it. + assert!(send(&mut sender, 1)?.is_empty()); + let promoted = sender.receive(&ack_for(g.sequence), 1)?.ready_to_send; + assert_eq!(sequences(&promoted), vec![g.sequence + 2, g.sequence + 3]); + Ok(()) +} + +#[test] +fn acknowledged_bytes_above_a_lost_packet_still_count_against_4_mib() -> Result<()> { + let mut sender = connected(1)?; + let g = send(&mut sender, 1)?.remove(0); + // g is lost. Each 1 MiB frame above it is delivered and acknowledged, and + // the receiver still holds it behind the gap, so it keeps counting. + for offset in 1..=4u32 { + let sent = send(&mut sender, MIB)?; + assert_eq!(sequences(&sent), vec![g.sequence + offset]); + sender.receive(&ack_for(g.sequence + offset), 1)?; + } + assert!(send(&mut sender, MIB)?.is_empty()); + assert_eq!(sender.queued_reliable_packet_count(), 1); + assert!(sender.send("state", vec![7; MIB], reliable()).is_err()); + + // g arrives: nothing is held behind a gap any more, and the queue drains. + let promoted = sender.receive(&ack_for(g.sequence), 2)?.ready_to_send; + assert_eq!(sequences(&promoted), vec![g.sequence + 5]); + // The bytes above the old gap no longer count: another MiB fits. + assert_eq!(sequences(&send(&mut sender, MIB)?), vec![g.sequence + 6]); + Ok(()) +} + +#[test] +fn acknowledged_bytes_count_against_the_lowest_unacked_sequence_as_it_advances() -> Result<()> { + let mut sender = connected(1)?; + let g = send(&mut sender, 1)?.remove(0); + let h = send(&mut sender, 1)?.remove(0); + for _ in 0..3 { + send(&mut sender, MIB)?; + } + // g and the 3 MiB above h are acknowledged; h is lost. + for offset in [2, 3, 4, 0] { + sender.receive(&ack_for(g.sequence + offset), 1)?; + } + assert_eq!(sequences(&send(&mut sender, MIB)?), vec![g.sequence + 5]); + // 4 MiB sit above h, acknowledged or not: one more byte does not fit. + assert!(send(&mut sender, 1)?.is_empty()); + // h arrives, the floor moves to g + 5, and the byte fits. + let promoted = sender.receive(&ack_for(h.sequence), 2)?.ready_to_send; + assert_eq!(sequences(&promoted), vec![g.sequence + 6]); + Ok(()) +} + +#[test] +fn acknowledged_bytes_stop_counting_when_the_lowest_unacked_packet_expires() -> Result<()> { + let mut sender = connected(1)?; + let expiring = CultNetRudpSendOptions { + reliable: true, + now_ms: 0, + reliable_expire_after_ms: Some(100), + ..Default::default() + }; + let g = sender + .send_many("state", vec![7], expiring, None)? + .remove(0); + for offset in 1..=4u32 { + assert_eq!(send(&mut sender, MIB)?.len(), 1); + sender.receive(&ack_for(g.sequence + offset), 1)?; + } + assert!(sender.send("state", vec![7; MIB], reliable()).is_err()); + + // g's deadline passes and it leaves the window; nothing is unacked, so the + // acknowledged bytes no longer sit above any gap. + let at = |now_ms| CultNetRudpSendOptions { + reliable: true, + now_ms, + ..Default::default() + }; + let first = sender.send_many("state", vec![7], at(200), None)?; + assert_eq!(first.len(), 1); + for _ in 0..4 { + assert_eq!(sender.send_many("state", vec![7; MIB], at(200), None)?.len(), 1); + } + Ok(()) +} + +#[test] +fn a_direct_send_is_not_refused_for_bytes_whose_lowest_packet_has_expired() -> Result<()> { + let mut sender = connected(1)?; + let expiring = CultNetRudpSendOptions { + reliable: true, + now_ms: 0, + reliable_expire_after_ms: Some(100), + ..Default::default() + }; + let g = sender + .send_many("state", vec![7], expiring, None)? + .remove(0); + for offset in 1..=4u32 { + assert_eq!(send(&mut sender, MIB)?.len(), 1); + sender.receive(&ack_for(g.sequence + offset), 1)?; + } + assert!(sender.send("state", vec![7; MIB], reliable()).is_err()); + + // The first call after g's deadline is a direct send: it measures the + // window without g, as `send_many` would, and goes on the wire. + let at_200 = CultNetRudpSendOptions { + reliable: true, + now_ms: 200, + ..Default::default() + }; + let sent = sender.send("state", vec![7; MIB], at_200)?; + assert_eq!(sent.sequence, g.sequence + 5); + assert_eq!(sender.outstanding_reliable_packet_count(), 1); + Ok(()) +} + +/// A server that accepts a fresh Connect may send before its Accept is +/// acknowledged, so no acknowledgement of the new generation has moved the +/// floor yet. Bytes the ended generation had acknowledged above its own lost +/// packet belong to no receiver's hold any more and must not count. +#[test] +fn a_new_generation_on_the_server_does_not_inherit_acknowledged_bytes() -> Result<()> { + let mut server = session(500); + let mut first_client = session(1); + let connect = first_client.create_connect(0, Vec::new())?; + let accept = server.accept_connect(&connect, 0, Vec::new())?; + server.receive(&ack_for(accept.sequence), 0)?; + let g = send(&mut server, 1)?.remove(0); + for offset in 1..=4u32 { + assert_eq!(sequences(&send(&mut server, MIB)?), vec![g.sequence + offset]); + server.receive(&ack_for(g.sequence + offset), 1)?; + } + assert!(send(&mut server, MIB)?.is_empty()); + + let mut second_client = session(200_000); + let connect = second_client.create_connect(2, Vec::new())?; + let accept = server.accept_connect(&connect, 2, Vec::new())?; + assert_eq!(accept.packet_type, CultNetRudpPacketType::Accept); + // The Accept is unacknowledged and the only thing pending: 1 MiB fits above it. + let sent = server.send("state", vec![7; MIB], reliable())?; + assert_eq!(sent.sequence, accept.sequence + 1); + Ok(()) +} diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index 23852a3c..5c1c9c5e 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -33,6 +33,11 @@ const RUDP_FIXED_HEADER_BYTES = 36; const MAX_CHANNEL_ID_BYTES = 255; export const CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS = 32; const RUDP_RECEIVED_SEQUENCE_WINDOW = 4_096; +// Flow window: a reliable packet is admitted only while its sequence is at most RUDP_FLOW_WINDOW_SEQUENCES above the +// lowest unacked one and the payload above that sequence stays within RUDP_FLOW_WINDOW_BYTES. +const RUDP_FLOW_WINDOW_SEQUENCES = 1_023; +const RUDP_FLOW_WINDOW_BYTES = 4 * 1024 * 1024; + /// How long a client keeps retransmitting a Connect nobody answered before it /// abandons that attempt for a fresh one. A Connect the server answers with an /// Ack, never an Accept, is one the server judged stale (see `acceptConnect`); @@ -212,6 +217,9 @@ export class CultNetRudpSession { readonly #receivedSequences = new Set(); readonly #latestSequencedByChannel = new Map(); readonly #pendingReliable = new Map(); + /// Payload sizes of acknowledged reliable sequences above the lowest unacknowledged one. The receiver still holds + /// those bytes behind the gap, so the flow window keeps counting them until the lowest sequence passes them. + readonly #ackedAboveLowest = new Map(); readonly #queuedReliable: CultNetRudpPacket[] = []; /// Every reliable sequence up to and including this one has been received /// since the peer state was last reset. Only the handshake seeds it: the @@ -294,6 +302,7 @@ export class CultNetRudpSession { this.#awaitingAccept = false; this.#generation += 1; this.#pendingReliable.clear(); + this.#ackedAboveLowest.clear(); this.#queuedReliable.splice(0); } @@ -420,7 +429,7 @@ export class CultNetRudpSession { payload: Uint8Array, options: { reliable?: boolean; ordered?: boolean; sequenced?: boolean; nowMs?: number } = {}, ): CultNetRudpPacket { - if (options.reliable && this.#pendingReliable.size >= CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS) { + if (options.reliable && (this.#queuedReliable.length > 0 || !this.#windowAdmits(this.#nextSequence, payload.length))) { throw new Error("RUDP reliable send window is full; receive acknowledgements before sending."); } return this.sendMany(channelId, payload, options)[0]!; @@ -753,18 +762,44 @@ export class CultNetRudpSession { }); } + /** + * Whether a reliable packet may go on the wire now: the window has a slot, and its sequence and the payload above + * the lowest unacked sequence stay inside the flow window. With nothing pending, any packet is admissible. + */ + #windowAdmits(sequence: number, payloadLength: number): boolean { + if (this.#pendingReliable.size >= CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS) return false; + if (this.#pendingReliable.size === 0) return true; + const lowest = Math.min(...this.#pendingReliable.keys()); + let bytesAbove = 0; + for (const [pendingSequence, pending] of this.#pendingReliable) { + if (pendingSequence > lowest) bytesAbove += pending.packet.payload?.length ?? 0; + } + for (const length of this.#ackedAboveLowest.values()) bytesAbove += length; + return sequence - lowest <= RUDP_FLOW_WINDOW_SEQUENCES && bytesAbove + payloadLength <= RUDP_FLOW_WINDOW_BYTES; + } + #admitReliablePackets(packets: CultNetRudpPacket[], nowMs: number): CultNetRudpPacket[] { - const available = Math.max(0, CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS - this.#pendingReliable.size); - const ready = packets.slice(0, available); - for (const packet of ready) this.#trackReliable(packet, nowMs); - this.#queuedReliable.push(...packets.slice(available)); + const ready: CultNetRudpPacket[] = []; + for (const packet of packets) { + if (this.#queuedReliable.length === 0 && this.#windowAdmits(packet.sequence, packet.payload?.length ?? 0)) { + this.#trackReliable(packet, nowMs); + ready.push(packet); + } else { + this.#queuedReliable.push(packet); + } + } return ready; } #promoteQueuedReliable(nowMs: number): CultNetRudpPacket[] { - const available = Math.max(0, CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS - this.#pendingReliable.size); - const ready = this.#queuedReliable.splice(0, available); - for (const packet of ready) this.#trackReliable(packet, nowMs); + const ready: CultNetRudpPacket[] = []; + while (this.#queuedReliable.length > 0) { + const packet = this.#queuedReliable[0]; + if (!this.#windowAdmits(packet.sequence, packet.payload?.length ?? 0)) break; + this.#queuedReliable.shift(); + this.#trackReliable(packet, nowMs); + ready.push(packet); + } return ready; } @@ -778,12 +813,33 @@ export class CultNetRudpSession { } #applyAcknowledgements(packet: CultNetRudpPacket): void { - this.#pendingReliable.delete(packet.ack); + this.#acknowledge(packet.ack); for (let bit = 0; bit < 32; bit += 1) { if ((packet.ackMask & (1 << bit)) !== 0) { - this.#pendingReliable.delete(packet.ack - bit - 1); + this.#acknowledge(packet.ack - bit - 1); } } + this.#forgetAckedBelowLowest(); + } + + #acknowledge(sequence: number): void { + const pending = this.#pendingReliable.get(sequence); + if (pending === undefined) return; + this.#pendingReliable.delete(sequence); + this.#ackedAboveLowest.set(sequence, pending.packet.payload?.length ?? 0); + } + + /// Acknowledged sizes stop counting once the lowest unacknowledged sequence passes them, and all of them stop + /// when nothing is unacknowledged. + #forgetAckedBelowLowest(): void { + if (this.#pendingReliable.size === 0) { + this.#ackedAboveLowest.clear(); + return; + } + const lowest = Math.min(...this.#pendingReliable.keys()); + for (const sequence of this.#ackedAboveLowest.keys()) { + if (sequence <= lowest) this.#ackedAboveLowest.delete(sequence); + } } /// The handshake's one act on the watermark: the peer's Connect or Accept is diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index 5f21fe72..e6b62bdb 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -1417,6 +1417,167 @@ test("rudp session advances large fragment sets through a bounded reliable windo assert.equal(oldAck.ackMask, 0); }); +// The sender's flow window: a reliable packet goes on the wire only while its sequence is at most 1,023 above the +// lowest unacked one and the payload above that sequence stays within 4 MiB. +const FLOW_CONNECTION_ID = 0x464c4f57; +const MIB = 1024 * 1024; + +function flowSession(initialSequence: number): CultNetRudpSession { + return new CultNetRudpSession({ connectionId: FLOW_CONNECTION_ID, initialSequence, resendDelayMs: 25 }); +} + +// A sender and a receiver that have shaken hands: the receiver's watermark is seeded by the sender's Connect, and the +// sender's Connect is acknowledged, so nothing is pending on the sender. +function connectedFlowPair(senderInitialSequence: number, receiverInitialSequence: number): [CultNetRudpSession, CultNetRudpSession] { + const sender = flowSession(senderInitialSequence); + const receiver = flowSession(receiverInitialSequence); + sender.receive(receiver.acceptConnect(sender.createConnect(0), 0), 0); + return [sender, receiver]; +} + +function connectedFlowSession(initialSequence: number): CultNetRudpSession { + return connectedFlowPair(initialSequence, 900)[0]; +} + +function flowSend(session: CultNetRudpSession, payloadBytes: number): CultNetRudpPacket[] { + return session.sendMany("state", new Uint8Array(payloadBytes), { reliable: true }); +} + +function flowAck(sequence: number): CultNetRudpPacket { + return { packetType: "ack", connectionId: FLOW_CONNECTION_ID, sequence: 0, ack: sequence, ackMask: 0, channelId: "control" }; +} + +const sequencesOf = (packets: CultNetRudpPacket[]): number[] => packets.map((packet) => packet.sequence); + +function loseGAndFillTheSpan(session: CultNetRudpSession): CultNetRudpPacket { + const g = flowSend(session, 1)[0]!; + for (let offset = 1; offset <= 1023; offset++) { + assert.deepEqual(sequencesOf(flowSend(session, 1)), [g.sequence + offset]); + session.receive(flowAck(g.sequence + offset), 1); + } + return g; +} + +test("a lost packet holds the sender 1023 sequences ahead and is still delivered", () => { + const [sender, receiver] = connectedFlowPair(1, 100); + const g = flowSend(sender, 1)[0]!; + let admittedAfterG = 0; + for (let index = 0; index < 4200; index++) { + for (const packet of flowSend(sender, 1)) { + admittedAfterG++; + receiver.receive(packet, 1); + sender.receive(receiver.createAck(), 1); + } + } + assert.equal(admittedAfterG, 1023); + assert.equal(sender.queuedReliablePacketCount, 4200 - 1023); + + const retransmit = sender.dueResends(1000).find((packet) => packet.sequence === g.sequence)!; + const delivered = receiver.receive(retransmit, 1000).delivered; + assert.deepEqual(delivered.map((frame) => frame.sequence), [g.sequence]); + + const promoted = sender.receive(receiver.createAckForReceived(g.sequence), 1001).readyToSend ?? []; + assert.equal(promoted[0]!.sequence, g.sequence + 1024); +}); + +test("a direct send past the span is refused without consuming a sequence", () => { + const sender = connectedFlowSession(1); + const g = loseGAndFillTheSpan(sender); + assert.throws(() => sender.send("state", new Uint8Array(1), { reliable: true })); + assert.equal(sender.queuedReliablePacketCount, 0); + sender.receive(flowAck(g.sequence), 2); + assert.equal(sender.send("state", new Uint8Array(1), { reliable: true }).sequence, g.sequence + 1024); +}); + +test("payload above the lowest unacked sequence is bounded by 4 MiB", () => { + const sender = connectedFlowSession(1); + const g = flowSend(sender, MIB)[0]!; + // The lowest unacked packet's own bytes do not count: four more MiB fit above it. + for (let index = 0; index < 4; index++) assert.equal(flowSend(sender, MIB).length, 1); + assert.equal(flowSend(sender, MIB).length, 0); + assert.equal(flowSend(sender, MIB).length, 0); + assert.equal(sender.queuedReliablePacketCount, 2); + + // Acking g moves the floor up one packet: exactly one MiB more fits. + const promoted = sender.receive(flowAck(g.sequence), 1).readyToSend ?? []; + assert.deepEqual(sequencesOf(promoted), [g.sequence + 5]); + assert.equal(sender.queuedReliablePacketCount, 1); +}); + +test("a packet larger than the window goes out alone and waits behind anything unacked", () => { + assert.equal(flowSend(connectedFlowSession(1), 5 * MIB).length, 1); + const sender = connectedFlowSession(1); + const g = flowSend(sender, 1)[0]!; + assert.equal(flowSend(sender, 5 * MIB).length, 0); + assert.deepEqual(sequencesOf(sender.receive(flowAck(g.sequence), 1).readyToSend ?? []), [g.sequence + 1]); +}); + +test("a small packet does not overtake a queued large one", () => { + const sender = connectedFlowSession(1); + const g = flowSend(sender, 1)[0]!; + assert.equal(flowSend(sender, 3 * MIB).length, 1); + assert.equal(flowSend(sender, 2 * MIB).length, 0); + // A direct send refuses instead of queueing behind it, and takes no sequence. + assert.throws(() => sender.send("state", new Uint8Array(1), { reliable: true })); + assert.equal(sender.queuedReliablePacketCount, 1); + // One byte would fit above g, but the 2 MiB packet is ahead of it. + assert.equal(flowSend(sender, 1).length, 0); + assert.deepEqual(sequencesOf(sender.receive(flowAck(g.sequence), 1).readyToSend ?? []), [g.sequence + 2, g.sequence + 3]); +}); + +test("acknowledged bytes above a lost packet still count against 4 MiB", () => { + const sender = connectedFlowSession(1); + const g = flowSend(sender, 1)[0]!; + // g is lost. Each 1 MiB frame above it is delivered and acknowledged, and the receiver still holds it behind the + // gap, so it keeps counting. + for (let offset = 1; offset <= 4; offset++) { + assert.deepEqual(sequencesOf(flowSend(sender, MIB)), [g.sequence + offset]); + sender.receive(flowAck(g.sequence + offset), 1); + } + assert.equal(flowSend(sender, MIB).length, 0); + assert.equal(sender.queuedReliablePacketCount, 1); + assert.throws(() => sender.send("state", new Uint8Array(MIB), { reliable: true })); + + // g arrives: nothing is held behind a gap any more, and the queue drains. + assert.deepEqual(sequencesOf(sender.receive(flowAck(g.sequence), 2).readyToSend ?? []), [g.sequence + 5]); + // The bytes above the old gap no longer count: another MiB fits. + assert.deepEqual(sequencesOf(flowSend(sender, MIB)), [g.sequence + 6]); +}); + +test("acknowledged bytes count against the lowest unacked sequence as it advances", () => { + const sender = connectedFlowSession(1); + const g = flowSend(sender, 1)[0]!; + const h = flowSend(sender, 1)[0]!; + for (let index = 0; index < 3; index++) flowSend(sender, MIB); + // g and the 3 MiB above h are acknowledged; h is lost. + for (const offset of [2, 3, 4, 0]) sender.receive(flowAck(g.sequence + offset), 1); + assert.deepEqual(sequencesOf(flowSend(sender, MIB)), [g.sequence + 5]); + // 4 MiB sit above h, acknowledged or not: one more byte does not fit. + assert.equal(flowSend(sender, 1).length, 0); + // h arrives, the floor moves to g + 5, and the byte fits. + assert.deepEqual(sequencesOf(sender.receive(flowAck(h.sequence), 2).readyToSend ?? []), [g.sequence + 6]); +}); + +// A server that accepts a fresh Connect may send before its Accept is acknowledged, so no acknowledgement of the new +// generation has moved the floor yet. Bytes the ended generation had acknowledged above its own lost packet belong to +// no receiver's hold any more and must not count. +test("a new generation on the server does not inherit acknowledged bytes", () => { + const server = flowSession(500); + const accept = server.acceptConnect(flowSession(1).createConnect(0), 0); + server.receive(flowAck(accept.sequence), 0); + const g = flowSend(server, 1)[0]!; + for (let offset = 1; offset <= 4; offset++) { + assert.deepEqual(sequencesOf(flowSend(server, MIB)), [g.sequence + offset]); + server.receive(flowAck(g.sequence + offset), 1); + } + assert.equal(flowSend(server, MIB).length, 0); + + const nextAccept = server.acceptConnect(flowSession(200_000).createConnect(2), 2); + assert.equal(nextAccept.packetType, "accept"); + // The Accept is unacknowledged and the only thing pending: 1 MiB fits above it. + assert.equal(server.send("state", new Uint8Array(MIB), { reliable: true }).sequence, nextAccept.sequence + 1); +}); + test("CultNet contracts encode legacy bytes without Node Buffer authority", () => { const originalBuffer = globalThis.Buffer; try { diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index 9799cb38..9731b7f9 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -885,6 +885,10 @@ namespace GameCult.Networking { /// Maximum reliable packets admitted to the wire before acknowledgements advance the window. public const int ReliableSendWindowPackets = 32; + // Flow window: a reliable packet is admitted only while its sequence is at most FlowWindowSequences above the + // lowest unacked one and the payload above that sequence stays within FlowWindowBytes. + private const uint FlowWindowSequences = 1023; + private const int FlowWindowBytes = 4 * 1024 * 1024; private const int ReceivedSequenceWindow = 4096; // How long a client keeps retransmitting a Connect nobody answered before it abandons that // attempt for a fresh one. A Connect the server answers with an Ack, never an Accept, is one @@ -935,6 +939,9 @@ namespace GameCult.Networking private readonly Dictionary _latestSequencedByChannel = new Dictionary(StringComparer.Ordinal); private readonly object _pendingReliableGate = new object(); private readonly Dictionary _pendingReliable = new Dictionary(); + // Payload sizes of acknowledged reliable sequences above the lowest unacknowledged one. The receiver still + // holds those bytes behind the gap, so the flow window keeps counting them until the lowest sequence passes them. + private readonly Dictionary _ackedAboveLowest = new Dictionary(); private readonly Queue _queuedReliable = new Queue(); // Every reliable sequence up to and including this one has been received since the peer state // was last reset. Only the handshake seeds it: the peer's Connect on the accepting side, the @@ -1156,7 +1163,7 @@ namespace GameCult.Networking options ??= new CultNetRudpSendOptions(); lock (_pendingReliableGate) { - if (options.Reliable && _pendingReliable.Count >= ReliableSendWindowPackets) + if (options.Reliable && (_queuedReliable.Count > 0 || !WindowAdmits(_nextSequence, payload.Length))) throw new InvalidOperationException("RUDP reliable send window is full; receive acknowledgements before sending."); } return SendMany(channelId, payload, options).First(); @@ -1420,6 +1427,7 @@ namespace GameCult.Networking lock (_pendingReliableGate) { _pendingReliable.Clear(); + _ackedAboveLowest.Clear(); _queuedReliable.Clear(); } } @@ -1573,17 +1581,39 @@ namespace GameCult.Networking } } + /// + /// Whether a reliable packet may go on the wire now: the window has a slot, and its sequence and the + /// payload above the lowest unacked sequence stay inside the flow window. With nothing pending, any + /// packet is admissible. Callers hold the pending gate. + /// + private bool WindowAdmits(uint sequence, int payloadLength) + { + if (_pendingReliable.Count >= ReliableSendWindowPackets) + return false; + if (_pendingReliable.Count == 0) + return true; + var lowest = _pendingReliable.Keys.Min(); + long bytesAbove = 0; + foreach (var pending in _pendingReliable) + { + if (pending.Key > lowest) + bytesAbove += pending.Value.Packet.Payload.Length; + } + foreach (var length in _ackedAboveLowest.Values) + bytesAbove += length; + return sequence - lowest <= FlowWindowSequences && bytesAbove + payloadLength <= FlowWindowBytes; + } + private IReadOnlyList AdmitReliablePackets( IReadOnlyList packets, long nowMs) { lock (_pendingReliableGate) { - var available = Math.Max(0, ReliableSendWindowPackets - _pendingReliable.Count); - var ready = new List(Math.Min(available, packets.Count)); + var ready = new List(); foreach (var packet in packets) { - if (ready.Count < available) + if (_queuedReliable.Count == 0 && WindowAdmits(packet.Sequence, packet.Payload.Length)) { _pendingReliable[packet.Sequence] = new PendingReliablePacket { @@ -1605,9 +1635,9 @@ namespace GameCult.Networking { lock (_pendingReliableGate) { - var available = Math.Max(0, ReliableSendWindowPackets - _pendingReliable.Count); - var ready = new List(Math.Min(available, _queuedReliable.Count)); - while (ready.Count < available && _queuedReliable.Count > 0) + var ready = new List(); + while (_queuedReliable.Count > 0 + && WindowAdmits(_queuedReliable.Peek().Sequence, _queuedReliable.Peek().Payload.Length)) { var packet = _queuedReliable.Dequeue(); _pendingReliable[packet.Sequence] = new PendingReliablePacket @@ -1641,17 +1671,36 @@ namespace GameCult.Networking { lock (_pendingReliableGate) { - _pendingReliable.Remove(packet.Ack); + Acknowledge(packet.Ack); for (var bit = 0; bit < 32; bit++) { if ((packet.AckMask & (1u << bit)) != 0 && packet.Ack > bit) { - _pendingReliable.Remove(packet.Ack - (uint)bit - 1); + Acknowledge(packet.Ack - (uint)bit - 1); } } + // Acknowledged sizes stop counting once the lowest unacknowledged sequence passes them, and all + // of them stop when nothing is unacknowledged. + if (_pendingReliable.Count == 0) + { + _ackedAboveLowest.Clear(); + } + else + { + var lowest = _pendingReliable.Keys.Min(); + foreach (var sequence in _ackedAboveLowest.Keys.Where(sequence => sequence <= lowest).ToArray()) + _ackedAboveLowest.Remove(sequence); + } } } + // Callers hold the pending gate. + private void Acknowledge(uint sequence) + { + if (_pendingReliable.Remove(sequence, out var pending)) + _ackedAboveLowest[sequence] = pending.Packet.Payload.Length; + } + private static bool PacketAcknowledges(CultNetRudpPacket packet, uint sequence) { if (packet.Ack == sequence) diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 8d01a7c3..5a1f570f 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1141,6 +1141,196 @@ namespace GameCult.Networking.Tests Assert.That(delivered[0].Payload, Is.EqualTo(payload)); } + // The sender's flow window: a reliable packet goes on the wire only while its sequence is at most 1,023 + // above the lowest unacked one and the payload above that sequence stays within 4 MiB. + private const int Mib = 1024 * 1024; + + private static CultNetRudpSession FlowSession(uint initialSequence) => + new CultNetRudpSession(new CultNetRudpSessionOptions + { + ConnectionId = 0x464c4f57, + InitialSequence = initialSequence, + ResendDelayMs = 25 + }); + + // A sender and a receiver that have shaken hands: the receiver's watermark is seeded by the sender's + // Connect, and the sender's Connect is acknowledged, so nothing is pending on the sender. + private static (CultNetRudpSession Sender, CultNetRudpSession Receiver) ConnectedFlowPair(uint senderInitialSequence, uint receiverInitialSequence) + { + var sender = FlowSession(senderInitialSequence); + var receiver = FlowSession(receiverInitialSequence); + sender.Receive(receiver.AcceptConnect(sender.CreateConnect(0), 0), 0); + return (sender, receiver); + } + + private static CultNetRudpSession ConnectedFlowSession(uint initialSequence) => ConnectedFlowPair(initialSequence, 900).Sender; + + private static IReadOnlyList FlowSend(CultNetRudpSession session, int payloadBytes) => + session.SendMany("state", new byte[payloadBytes], new CultNetRudpSendOptions { Reliable = true }); + + private static CultNetRudpPacket FlowAck(uint sequence) => + new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Ack, ConnectionId = 0x464c4f57, Ack = sequence, ChannelId = "control" }; + + private static CultNetRudpPacket LoseGAndFillTheSpan(CultNetRudpSession session) + { + var g = FlowSend(session, 1).Single(); + for (uint offset = 1; offset <= 1023; offset++) + { + Assert.That(FlowSend(session, 1).Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + offset })); + session.Receive(FlowAck(g.Sequence + offset), 1); + } + return g; + } + + [Test] + public void RudpSession_ALostPacketHoldsTheSender1023SequencesAheadAndIsStillDelivered() + { + var (sender, receiver) = ConnectedFlowPair(1, 100); + var g = FlowSend(sender, 1).Single(); + var admittedAfterG = 0; + for (var index = 0; index < 4200; index++) + { + foreach (var packet in FlowSend(sender, 1)) + { + admittedAfterG++; + receiver.Receive(packet, 1); + sender.Receive(receiver.CreateAck(), 1); + } + } + Assert.That(admittedAfterG, Is.EqualTo(1023)); + Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(4200 - 1023)); + + var retransmit = sender.DueResends(1000).Single(packet => packet.Sequence == g.Sequence); + var delivered = receiver.Receive(retransmit, 1000).Delivered; + Assert.That(delivered.Select(frame => frame.Sequence), Is.EqualTo(new[] { g.Sequence })); + + var promoted = sender.Receive(receiver.CreateAckForReceived(g.Sequence), 1001).ReadyToSend; + Assert.That(promoted.First().Sequence, Is.EqualTo(g.Sequence + 1024)); + } + + [Test] + public void RudpSession_ADirectSendPastTheSpanIsRefusedWithoutConsumingASequence() + { + var sender = ConnectedFlowSession(1); + var g = LoseGAndFillTheSpan(sender); + Assert.Throws(() => sender.Send("state", new byte[1], new CultNetRudpSendOptions { Reliable = true })); + Assert.That(sender.QueuedReliablePacketCount, Is.Zero); + sender.Receive(FlowAck(g.Sequence), 2); + var next = sender.Send("state", new byte[1], new CultNetRudpSendOptions { Reliable = true }); + Assert.That(next.Sequence, Is.EqualTo(g.Sequence + 1024)); + } + + [Test] + public void RudpSession_PayloadAboveTheLowestUnackedSequenceIsBoundedBy4MiB() + { + var sender = ConnectedFlowSession(1); + var g = FlowSend(sender, Mib).Single(); + // The lowest unacked packet's own bytes do not count: four more MiB fit above it. + for (var index = 0; index < 4; index++) + Assert.That(FlowSend(sender, Mib), Has.Count.EqualTo(1)); + Assert.That(FlowSend(sender, Mib), Is.Empty); + Assert.That(FlowSend(sender, Mib), Is.Empty); + Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(2)); + + // Acking g moves the floor up one packet: exactly one MiB more fits. + var promoted = sender.Receive(FlowAck(g.Sequence), 1).ReadyToSend; + Assert.That(promoted.Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 5 })); + Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(1)); + } + + [Test] + public void RudpSession_APacketLargerThanTheWindowGoesOutAloneAndWaitsBehindAnythingUnacked() + { + Assert.That(FlowSend(ConnectedFlowSession(1), 5 * Mib), Has.Count.EqualTo(1)); + var sender = ConnectedFlowSession(1); + var g = FlowSend(sender, 1).Single(); + Assert.That(FlowSend(sender, 5 * Mib), Is.Empty); + var promoted = sender.Receive(FlowAck(g.Sequence), 1).ReadyToSend; + Assert.That(promoted.Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 1 })); + } + + [Test] + public void RudpSession_ASmallPacketDoesNotOvertakeAQueuedLargeOne() + { + var sender = ConnectedFlowSession(1); + var g = FlowSend(sender, 1).Single(); + Assert.That(FlowSend(sender, 3 * Mib), Has.Count.EqualTo(1)); + Assert.That(FlowSend(sender, 2 * Mib), Is.Empty); + // A direct send refuses instead of queueing behind it, and takes no sequence. + Assert.Throws(() => sender.Send("state", new byte[1], new CultNetRudpSendOptions { Reliable = true })); + Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(1)); + // One byte would fit above g, but the 2 MiB packet is ahead of it. + Assert.That(FlowSend(sender, 1), Is.Empty); + var promoted = sender.Receive(FlowAck(g.Sequence), 1).ReadyToSend; + Assert.That(promoted.Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 2, g.Sequence + 3 })); + } + + [Test] + public void RudpSession_AcknowledgedBytesAboveALostPacketStillCountAgainst4MiB() + { + var sender = ConnectedFlowSession(1); + var g = FlowSend(sender, 1).Single(); + // g is lost. Each 1 MiB frame above it is delivered and acknowledged, and the receiver still holds it + // behind the gap, so it keeps counting. + for (uint offset = 1; offset <= 4; offset++) + { + Assert.That(FlowSend(sender, Mib).Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + offset })); + sender.Receive(FlowAck(g.Sequence + offset), 1); + } + Assert.That(FlowSend(sender, Mib), Is.Empty); + Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(1)); + Assert.Throws(() => sender.Send("state", new byte[Mib], new CultNetRudpSendOptions { Reliable = true })); + + // g arrives: nothing is held behind a gap any more, and the queue drains. + var promoted = sender.Receive(FlowAck(g.Sequence), 2).ReadyToSend; + Assert.That(promoted.Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 5 })); + // The bytes above the old gap no longer count: another MiB fits. + Assert.That(FlowSend(sender, Mib).Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 6 })); + } + + [Test] + public void RudpSession_AcknowledgedBytesCountAgainstTheLowestUnackedSequenceAsItAdvances() + { + var sender = ConnectedFlowSession(1); + var g = FlowSend(sender, 1).Single(); + var h = FlowSend(sender, 1).Single(); + for (var index = 0; index < 3; index++) + FlowSend(sender, Mib); + // g and the 3 MiB above h are acknowledged; h is lost. + foreach (uint offset in new uint[] { 2, 3, 4, 0 }) + sender.Receive(FlowAck(g.Sequence + offset), 1); + Assert.That(FlowSend(sender, Mib).Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 5 })); + // 4 MiB sit above h, acknowledged or not: one more byte does not fit. + Assert.That(FlowSend(sender, 1), Is.Empty); + // h arrives, the floor moves to g + 5, and the byte fits. + var promoted = sender.Receive(FlowAck(h.Sequence), 2).ReadyToSend; + Assert.That(promoted.Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 6 })); + } + + // A server that accepts a fresh Connect may send before its Accept is acknowledged, so no acknowledgement of + // the new generation has moved the floor yet. Bytes the ended generation had acknowledged above its own lost + // packet belong to no receiver's hold any more and must not count. + [Test] + public void RudpSession_ANewGenerationOnTheServerDoesNotInheritAcknowledgedBytes() + { + var server = FlowSession(500); + var accept = server.AcceptConnect(FlowSession(1).CreateConnect(0), 0); + server.Receive(FlowAck(accept.Sequence), 0); + var g = FlowSend(server, 1).Single(); + for (uint offset = 1; offset <= 4; offset++) + { + Assert.That(FlowSend(server, Mib).Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + offset })); + server.Receive(FlowAck(g.Sequence + offset), 1); + } + Assert.That(FlowSend(server, Mib), Is.Empty); + + var nextAccept = server.AcceptConnect(FlowSession(200_000).CreateConnect(2), 2); + Assert.That(nextAccept.PacketType, Is.EqualTo(CultNetRudpPacketType.Accept)); + // The Accept is unacknowledged and the only thing pending: 1 MiB fits above it. + var sent = server.Send("state", new byte[Mib], new CultNetRudpSendOptions { Reliable = true }); + Assert.That(sent.Sequence, Is.EqualTo(nextAccept.Sequence + 1)); + } + [Test] public void RudpSocketTransport_HandshakesAndCarriesReliableOrderedSchemaFrames() {