GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Ack Cut 3: the sender's flow window (1,023 sequences and 4 MiB above the lowest unacked)

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 00:55:14 +02:00
commit eeda6d7fd6
5 changed files with 146 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

9
docs/cultnet-transport-parity.md
View file

@ -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

39
packages/cultnet-py/src/cultnet_py/transport.py
View file

@ -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)

48
packages/cultnet-rs/src/rudp.rs
View file

@ -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<CultNetRudpPacket> {
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<CultNetRudpPacket>,
now_ms: u64,
expires_at_ms: Option<u64>,
) -> Vec<CultNetRudpPacket> {
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<CultNetRudpPacket> {
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;
};

45
packages/cultnet-ts/src/rudp.ts
View file

@ -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;
}

38
src/GameCult.Networking/CultNetTransport.cs
View file

@ -881,6 +881,10 @@ namespace GameCult.Networking
{
/// <summary>Maximum reliable packets admitted to the wire before acknowledgements advance the window.</summary>
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
}
}
/// <summary>
/// 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.
/// </summary>
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<CultNetRudpPacket> AdmitReliablePackets(
IReadOnlyList<CultNetRudpPacket> packets,
long nowMs)
{
lock (_pendingReliableGate)
{
var available = Math.Max(0, ReliableSendWindowPackets - _pendingReliable.Count);
var ready = new List<CultNetRudpPacket>(Math.Min(available, packets.Count));
var ready = new List<CultNetRudpPacket>();
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<CultNetRudpPacket>(Math.Min(available, _queuedReliable.Count));
while (ready.Count < available && _queuedReliable.Count > 0)
var ready = new List<CultNetRudpPacket>();
while (_queuedReliable.Count > 0
&& WindowAdmits(_queuedReliable.Peek().Sequence, _queuedReliable.Peek().Payload.Length))
{
var packet = _queuedReliable.Dequeue();
_pendingReliable[packet.Sequence] = new PendingReliablePacket