From eeda6d7fd61984594059c52d67399bc861faf243 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 00:55:14 +0200 Subject: [PATCH] Ack Cut 3: the sender's flow window (1,023 sequences and 4 MiB above the lowest unacked) Admission in Rust, C#, TypeScript and Python goes through one predicate, window_admits: the 32-packet slot, a sequence at most 1,023 above the lowest unacked one, and at most 4 MiB of payload above it. With nothing pending any packet is admissible. Admission is FIFO: nothing overtakes a queued packet. The single-packet send refuses on the same predicate instead of queueing a packet and then reporting that it produced none. The receive window is stated in docs/cultnet-transport-parity.md. Kotlin trails (Cut K3). Co-Authored-By: Claude Sonnet 5.5 --- docs/cultnet-transport-parity.md | 9 ++++ .../cultnet-py/src/cultnet_py/transport.py | 39 +++++++++++---- packages/cultnet-rs/src/rudp.rs | 48 +++++++++++++++---- packages/cultnet-ts/src/rudp.ts | 45 +++++++++++++---- src/GameCult.Networking/CultNetTransport.cs | 38 ++++++++++++--- 5 files changed, 146 insertions(+), 33 deletions(-) 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 0f378613..6b0ce026 100644 --- a/packages/cultnet-py/src/cultnet_py/transport.py +++ b/packages/cultnet-py/src/cultnet_py/transport.py @@ -402,6 +402,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") @@ -531,7 +535,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] @@ -773,22 +777,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 + ) + 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) diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 3b348fe3..ac4054a9 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -46,6 +46,13 @@ const RUDP_MAX_TRACKED_CHANNELS: usize = 64; const RUDP_RECEIVE_AHEAD_WINDOW: u32 = 1_024; 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 { @@ -441,7 +448,8 @@ impl CultNetRudpSession { options: CultNetRudpSendOptions, ) -> Result { 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" @@ -907,17 +915,37 @@ impl CultNetRudpSession { .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 bytes_above: usize = self + .pending_reliable + .iter() + .filter(|(pending_sequence, _)| **pending_sequence > lowest) + .map(|(_, pending)| pending.packet.payload.len()) + .sum(); + 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 { @@ -928,10 +956,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; }; diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index ae7ea838..10a6b9a2 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -18,6 +18,10 @@ 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; export type CultNetRudpPacketType = | "connect" @@ -295,7 +299,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]!; @@ -583,18 +587,43 @@ 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; + } + 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; } diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index e98fdc03..de61d1af 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -881,6 +881,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; private sealed class PendingReliablePacket { @@ -1063,7 +1067,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(); @@ -1436,17 +1440,37 @@ 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; + } + 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 { @@ -1468,9 +1492,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