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 <noreply@anthropic.com>

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

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

packages/cultnet-rs/src/rudp.rs

@@ -251,6 +251,11 @@ pub struct CultNetRudpSession {
     received_sequences: BTreeSet<u32>,
     latest_sequenced_by_channel: BTreeMap<String, u32>,
     pending_reliable: BTreeMap<u32, 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
+    /// advances past them.
+    acked_above_lowest: BTreeMap<u32, usize>,
     queued_reliable: VecDeque<(CultNetRudpPacket, Option<u64>)>,
     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

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(())
+}

packages/cultnet-ts/src/rudp.ts

@@ -203,6 +203,9 @@ export class CultNetRudpSession {
   readonly #receivedSequences = new Set<number>();
   readonly #latestSequencedByChannel = new Map<string, number>();
   readonly #pendingReliable = new Map<number, 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.
+  readonly #ackedAboveLowest = new Map<number, number>();
   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

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 {

src/GameCult.Networking/CultNetTransport.cs

@@ -939,6 +939,9 @@ namespace GameCult.Networking
         private readonly Dictionary<string, uint> _latestSequencedByChannel = new Dictionary<string, uint>(StringComparer.Ordinal);
         private readonly object _pendingReliableGate = new object();
         private readonly Dictionary<uint, PendingReliablePacket> _pendingReliable = new Dictionary<uint, 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.
+        private readonly Dictionary<uint, int> _ackedAboveLowest = new Dictionary<uint, int>();
         private readonly Queue<CultNetRudpPacket> _queuedReliable = new Queue<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 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)

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<InvalidOperationException>(() => 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()
         {