From d7eacd1011948b690fa454ee6c847c99b77d29d8 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 12:04:19 +0200 Subject: [PATCH] Ack Cut 3 F1: the 4 MiB flow bound counts acknowledged bytes above the lowest unacknowledged sequence The bound summed only packets still awaiting an ack. Once frames above a lost packet were acknowledged they stopped counting, so a receiver could be made to hold more than 4 MiB behind one loss, and its hold buffer ended the session. The sender now keeps the payload size of each acknowledged sequence above the lowest pending one until the lowest advances past it, in Rust, TypeScript, Python and C#. Tests per runtime: one lost packet, four acknowledged 1 MiB frames, the fifth refused until the gap closes; and the floor moving to a new lowest sequence. Co-Authored-By: Claude Sonnet 5.5 --- .../cultnet-py/src/cultnet_py/transport.py | 21 ++++++++-- packages/cultnet-py/tests/test_cultnet.py | 33 +++++++++++++++ packages/cultnet-rs/src/rudp.rs | 33 +++++++++++++-- packages/cultnet-rs/tests/rudp_flow_window.rs | 42 +++++++++++++++++++ packages/cultnet-ts/src/rudp.ts | 30 ++++++++++++- packages/cultnet-ts/test/cultnet.test.ts | 31 ++++++++++++++ src/GameCult.Networking/CultNetTransport.cs | 29 ++++++++++++- .../NetworkingTests.cs | 40 ++++++++++++++++++ 8 files changed, 249 insertions(+), 10 deletions(-) diff --git a/packages/cultnet-py/src/cultnet_py/transport.py b/packages/cultnet-py/src/cultnet_py/transport.py index a5927070..1e718aab 100644 --- a/packages/cultnet-py/src/cultnet_py/transport.py +++ b/packages/cultnet-py/src/cultnet_py/transport.py @@ -482,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 @@ -544,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: @@ -950,7 +954,7 @@ class CultNetRudpSession: 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( @@ -984,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 8105c83d..f11eb1ce 100644 --- a/packages/cultnet-py/tests/test_cultnet.py +++ b/packages/cultnet-py/tests/test_cultnet.py @@ -980,6 +980,39 @@ class CultNetTests(unittest.TestCase): 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]) + + 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_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 7d8e6f12..f4b0616f 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -251,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 @@ -291,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, @@ -375,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(); } @@ -1080,6 +1087,7 @@ impl CultNetRudpSession { .retain(|_, pending| pending.expires_at_ms.is_none_or(|at| now_ms <= at)); self.queued_reliable .retain(|(_, expires_at_ms)| expires_at_ms.is_none_or(|at| now_ms <= at)); + self.forget_acked_below_lowest(); let after = self.pending_reliable.len() + self.queued_reliable.len(); self.reliable_packets_expired = self .reliable_packets_expired @@ -1096,12 +1104,14 @@ impl CultNetRudpSession { let Some((&lowest, _)) = self.pending_reliable.first_key_value() else { return true; }; - let bytes_above: usize = self + 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 } @@ -1163,12 +1173,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_flow_window.rs b/packages/cultnet-rs/tests/rudp_flow_window.rs index 73b7f8e6..bc7493cc 100644 --- a/packages/cultnet-rs/tests/rudp_flow_window.rs +++ b/packages/cultnet-rs/tests/rudp_flow_window.rs @@ -172,3 +172,45 @@ fn a_small_packet_does_not_overtake_a_queued_large_one() -> Result<()> { 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]); + 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(()) +} diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index e072e225..ae696536 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -203,6 +203,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 @@ -285,6 +288,7 @@ export class CultNetRudpSession { this.#awaitingAccept = false; this.#generation += 1; this.#pendingReliable.clear(); + this.#ackedAboveLowest.clear(); this.#queuedReliable.splice(0); } @@ -756,6 +760,7 @@ export class CultNetRudpSession { 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; } @@ -794,12 +799,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 f3c1612b..c921cb0d 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -1524,6 +1524,37 @@ test("a small packet does not overtake a queued large one", () => { 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]); +}); + +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]); +}); + 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 6501fd3d..9731b7f9 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -939,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 @@ -1424,6 +1427,7 @@ namespace GameCult.Networking lock (_pendingReliableGate) { _pendingReliable.Clear(); + _ackedAboveLowest.Clear(); _queuedReliable.Clear(); } } @@ -1595,6 +1599,8 @@ namespace GameCult.Networking 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; } @@ -1665,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 15091ec3..fbd422ce 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1265,6 +1265,46 @@ namespace GameCult.Networking.Tests 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 })); + } + + [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 })); + } + [Test] public void RudpSocketTransport_HandshakesAndCarriesReliableOrderedSchemaFrames() {