Acknowledge retransmits across RUDP horizons

docs/cultnet-transport-parity.md

@@ -259,8 +259,11 @@ Current progress:
   state transitions.
 - TypeScript, C#, Rust, Python, and Kotlin now share bounded reliable-send
   backpressure for RUDP sessions. At most 32 reliable packets are admitted to
-  the wire at once; later fragments remain in a session-owned FIFO and exact
-  acknowledgements promote the next packets. This keeps the send frontier well
+  the wire at once; later fragments remain in a session-owned FIFO and
+  acknowledgements promote the next packets. Receivers emit a cumulative ACK
+  for packets inside the 32-sequence ACK horizon and an exact ACK for an older
+  retransmit, so unrelated unreliable traffic cannot strand reliable work.
+  This keeps the send frontier well
   inside the 4,096-sequence receive history instead of blasting an entire large
   document and hoping the UDP buffers impersonate flow control.
   `maxPendingReliablePackets` bounds both queued and in-flight packets before a

packages/cultcache-py/src/cultmesh_py/server.py

@@ -268,7 +268,11 @@ class CultMeshLocalServer:
                         msgpack.packb(response, use_bin_type=True),
                     )
             if packet.reliable or packet.packet_type == CultNetRudpPacketType.DATA or result.delivered:
-                self._send_rudp_packet(rudp_socket, remote_addr, peer.session.create_ack())
+                self._send_rudp_packet(
+                    rudp_socket,
+                    remote_addr,
+                    peer.session.create_ack_for_received(packet.sequence),
+                )
 
     def _poll_rudp_resends(
         self,

packages/cultcache-py/src/cultnet_py/interop_peer.py

@@ -320,7 +320,7 @@ def rudp_loop(
                 for response in handle_server_message(state, message, peer.subscriptions):
                     send_rudp_schema_frame(sock, remote, peer.session, response)
             if packet.reliable or packet.packet_type == CultNetRudpPacketType.DATA or result.delivered:
-                send_rudp_packet(sock, remote, peer.session.create_ack())
+                send_rudp_packet(sock, remote, peer.session.create_ack_for_received(packet.sequence))
         except Exception as error:
             sys.stderr.write(json.dumps({
                 "event": "rudpMessageError",

packages/cultcache-py/src/cultnet_py/transport.py

@@ -621,6 +621,13 @@ class CultNetRudpSession:
             payload=b"",
         )
 
+    def create_ack_for_received(self, sequence: int) -> CultNetRudpPacket:
+        received_sequence = _uint32(sequence, "received sequence")
+        ack, _ = self._ack_state()
+        if ack >= received_sequence and ack - received_sequence <= 32:
+            return self.create_ack()
+        return self.create_ack_for(received_sequence)
+
     def create_ping(self, payload: bytes = b"") -> CultNetRudpPacket:
         return self._create_packet(CultNetRudpPacketType.PING, "control", payload)
 
@@ -964,7 +971,7 @@ class CultNetRudpSocketTransportConnection:
 
         frame = self._delivered_frames.popleft() if self._delivered_frames else None
         if packet.reliable or packet.packet_type == CultNetRudpPacketType.ACCEPT or frame is not None:
-            self._send_packet(self.session.create_ack())
+            self._send_packet(self.session.create_ack_for_received(packet.sequence))
         return frame
 
     def receive(self, timeout_seconds: float | None = None) -> CultNetTransportFrame:

packages/cultcache-py/tests/test_cultcache.py

@@ -1173,6 +1173,7 @@ class CultCacheTests(unittest.TestCase):
             max_fragment_bytes=8,
         ))
         self.assertEqual(len(wire), CultNetRudpSession.RELIABLE_SEND_WINDOW_PACKETS)
+        oldest_sequence = wire[0].sequence
         self.assertEqual(len(sender.pending_reliable_sequences), len(wire))
         self.assertEqual(sender.queued_reliable_packet_count, 17)
 
@@ -1188,6 +1189,9 @@ class CultCacheTests(unittest.TestCase):
         self.assertEqual(sender.outstanding_reliable_packet_count, 0)
         self.assertEqual(len(delivered), 1)
         self.assertEqual(delivered[0].payload, payload)
+        old_ack = receiver.create_ack_for_received(oldest_sequence)
+        self.assertEqual(old_ack.ack, oldest_sequence)
+        self.assertEqual(old_ack.ack_mask, 0)
 
     def test_cultnet_rudp_socket_transport_handshakes_and_carries_reliable_ordered_schema_frames(self) -> None:
         server_socket = bind_udp_socket()

packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt

@@ -2785,6 +2785,16 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) {
         channelId = "control",
     )
 
+    fun createAckForReceived(sequence: Long): CultNetRudpPacket {
+        val receivedSequence = uint32(sequence, "received sequence")
+        val (ack, _) = ackState()
+        return if (ack >= receivedSequence && ack - receivedSequence <= 32L) {
+            createAck()
+        } else {
+            createAckFor(receivedSequence)
+        }
+    }
+
     fun createPing(payload: ByteArray = ByteArray(0)): CultNetRudpPacket =
         createPacket(CultNetRudpPacketType.Ping, "control", payload)
 
@@ -3215,7 +3225,9 @@ class CultNetRudpSocketTransportConnection(
             framesReceived += 1
         }
         val frame = if (delivered.isEmpty()) null else delivered.removeFirst()
-        if (packet.reliable || packet.packetType == CultNetRudpPacketType.Accept || frame != null) sendPacket(session.createAck())
+        if (packet.reliable || packet.packetType == CultNetRudpPacketType.Accept || frame != null) {
+            sendPacket(session.createAckForReceived(packet.sequence))
+        }
         return frame
     }
 
@@ -3982,7 +3994,9 @@ private fun startInteropRudpServer(
                 val result = session.receive(packet, nowMs())
                 result.reply?.let { sendInteropRudpPacket(socket, remote, it) }
                 result.readyToSend.forEach { sendInteropRudpPacket(socket, remote, it) }
-                if (packet.reliable || packet.packetType == CultNetRudpPacketType.Data) sendInteropRudpPacket(socket, remote, session.createAck())
+                if (packet.reliable || packet.packetType == CultNetRudpPacketType.Data) {
+                    sendInteropRudpPacket(socket, remote, session.createAckForReceived(packet.sequence))
+                }
                 for (frame in result.delivered) {
                     if (frame.channelId != "schema") continue
                     val sender = InteropRudpSessionSender(socket, remote, session)
@@ -5328,6 +5342,7 @@ private fun rudpSessionAdvancesLargeFragmentSetsThroughBoundedReliableWindow() {
         maxFragmentBytes = 8,
     ))
     check(wire.size == CultNetRudpSession.ReliableSendWindowPackets)
+    val oldestSequence = wire.first().sequence
     check(sender.pendingReliableSequences.size == wire.size)
     check(sender.queuedReliablePacketCount == 17)
 
@@ -5343,6 +5358,9 @@ private fun rudpSessionAdvancesLargeFragmentSetsThroughBoundedReliableWindow() {
     check(sender.outstandingReliablePacketCount == 0)
     check(delivered.size == 1)
     check(delivered.first().payload.contentEquals(payload))
+    val oldAck = receiver.createAckForReceived(oldestSequence)
+    check(oldAck.ack == oldestSequence)
+    check(oldAck.ackMask == 0L)
 }
 
 private fun rudpSocketTransportErgonomicFactoriesCarrySchemaFrames() {

packages/cultnet-rs/src/rudp.rs

@@ -542,6 +542,15 @@ impl CultNetRudpSession {
         }
     }
 
+    pub fn create_ack_for_received(&mut self, sequence: u32) -> CultNetRudpPacket {
+        let (ack, _) = self.ack_state();
+        if ack >= sequence && ack - sequence <= 32 {
+            self.create_ack()
+        } else {
+            self.create_ack_for(sequence)
+        }
+    }
+
     pub fn create_ping(&mut self, payload: Vec<u8>) -> CultNetRudpPacket {
         self.create_packet(
             CultNetRudpPacketType::Ping,
@@ -1340,7 +1349,7 @@ impl CultNetRudpServerHub {
         let result = peer.session.receive(&packet, now_ms())?;
         let context = peer.context.clone();
         let ack = if packet.reliable {
-            Some(peer.session.create_ack())
+            Some(peer.session.create_ack_for_received(packet.sequence))
         } else {
             None
         };
@@ -1657,7 +1666,7 @@ impl CultNetRudpSocketTransportConnection {
         let frame = self.delivered_frames.pop_front();
         if packet.reliable || packet.packet_type == CultNetRudpPacketType::Accept || frame.is_some()
         {
-            let ack = self.session.create_ack();
+            let ack = self.session.create_ack_for_received(packet.sequence);
             self.send_packet(&ack)?;
         }
         Ok(frame)

packages/cultnet-rs/tests/cultnet.rs

@@ -1303,6 +1303,7 @@ fn rudp_session_advances_large_fragment_sets_through_a_bounded_reliable_window()
         Some(8),
     )?;
     assert_eq!(initial.len(), CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS);
+    let oldest_sequence = initial[0].sequence;
     assert_eq!(sender.pending_reliable_sequences().len(), initial.len());
     assert_eq!(sender.queued_reliable_packet_count(), 17);
 
@@ -1322,6 +1323,9 @@ fn rudp_session_advances_large_fragment_sets_through_a_bounded_reliable_window()
     assert_eq!(sender.outstanding_reliable_packet_count(), 0);
     assert_eq!(delivered.len(), 1);
     assert_eq!(delivered[0].payload, payload);
+    let old_ack = receiver.create_ack_for_received(oldest_sequence);
+    assert_eq!(old_ack.ack, oldest_sequence);
+    assert_eq!(old_ack.ack_mask, 0);
     Ok(())
 }
 

packages/cultnet-ts/src/operation-service.ts

@@ -163,7 +163,9 @@ async function handleServerDatagram(
       maxFragmentBytes: options.maxFragmentBytes ?? 2048,
     })) sendPacket(remote, responsePacket);
   }
-  if (packet.packetType === "data" || result.delivered.length > 0) sendPacket(remote, peer.session.createAck());
+  if (packet.packetType === "data" || result.delivered.length > 0) {
+    sendPacket(remote, peer.session.createAckForReceived(packet.sequence));
+  }
 }
 
 function parseRudpEndpoint(endpoint: string): { host: string; port: number } {

packages/cultnet-ts/src/rudp.ts

@@ -434,6 +434,14 @@ export class CultNetRudpSession {
     };
   }
 
+  createAckForReceived(sequence: number): CultNetRudpPacket {
+    const receivedSequence = toUint32(sequence, "received sequence");
+    const { ack } = this.#ackState();
+    return ack >= receivedSequence && ack - receivedSequence <= 32
+      ? this.createAck()
+      : this.createAckFor(receivedSequence);
+  }
+
   createPing(payload = new Uint8Array()): CultNetRudpPacket {
     return this.#createPacket({
       packetType: "ping",
@@ -917,7 +925,7 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement
         return;
       }
       if (packet.packetType === "accept" || packet.packetType === "data" || result.delivered.length > 0) {
-        this.#sendPacket(this.#session.createAck());
+        this.#sendPacket(this.#session.createAckForReceived(packet.sequence));
       }
     } catch (error) {
       this.emit("error", error instanceof Error ? error : new Error(String(error)));

packages/cultnet-ts/test/cultnet.test.ts

@@ -1317,6 +1317,7 @@ test("rudp session advances large fragment sets through a bounded reliable windo
     maxFragmentBytes: 8,
   });
   assert.equal(wire.length, CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS);
+  const oldestSequence = wire[0]!.sequence;
   assert.equal(sender.pendingReliableSequences.length, wire.length);
   assert.equal(sender.queuedReliablePacketCount, 17);
 
@@ -1334,6 +1335,9 @@ test("rudp session advances large fragment sets through a bounded reliable windo
   assert.equal(sender.outstandingReliablePacketCount, 0);
   assert.equal(delivered.length, 1);
   assert.deepEqual(delivered[0]?.payload, payload);
+  const oldAck = receiver.createAckForReceived(oldestSequence);
+  assert.equal(oldAck.ack, oldestSequence);
+  assert.equal(oldAck.ackMask, 0);
 });
 
 test("CultNet contracts encode legacy bytes without Node Buffer authority", () => {

src/GameCult.Networking/CultNetTransport.cs

@@ -1193,6 +1193,18 @@ namespace GameCult.Networking
             };
         }
 
+        /// <summary>
+        /// Creates a cumulative acknowledgement when the received packet is inside the
+        /// acknowledgement horizon, or an exact acknowledgement for an older retransmit.
+        /// </summary>
+        public CultNetRudpPacket CreateAckForReceived(uint receivedSequence)
+        {
+            var (ack, _) = AckState();
+            return ack >= receivedSequence && ack - receivedSequence <= 32
+                ? CreateAck()
+                : CreateAck(receivedSequence);
+        }
+
         /// <summary>
         /// Creates a packet carrying a keepalive ping payload.
         /// </summary>
@@ -2077,7 +2089,7 @@ namespace GameCult.Networking
                 SendPackets(result.ReadyToSend);
                 if (packet.PacketType == CultNetRudpPacketType.Accept ||
                     packet.PacketType == CultNetRudpPacketType.Data)
-                    acknowledgement = _session.CreateAck();
+                    acknowledgement = _session.CreateAckForReceived(packet.Sequence);
             }
             if (result.Pong)
             {
@@ -2486,7 +2498,7 @@ namespace GameCult.Networking
                     SendPacket(existingPeer.RemoteEndPoint, result.Reply);
                 SendPackets(existingPeer, result.ReadyToSend);
                 if (packet.PacketType == CultNetRudpPacketType.Data)
-                    acknowledgement = existingPeer.Session.CreateAck();
+                    acknowledgement = existingPeer.Session.CreateAckForReceived(packet.Sequence);
             }
             if (result.Disconnected)
             {

tests/GameCult.Networking.Tests/NetworkingTests.cs

@@ -790,12 +790,15 @@ namespace GameCult.Networking.Tests
             receiver.Receive(new CultNetRudpPacket
                 { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 991, Sequence = 0, ChannelId = "control" });
 
-            var packets = sender.SendMany(
+            var oldPacket = sender.Send(
                 "snapshot",
-                Enumerable.Range(0, 80).Select(index => (byte)index).ToArray(),
-                new CultNetRudpSendOptions { Reliable = true, Ordered = true, NowMs = 0 },
-                maxFragmentBytes: 1);
-            foreach (var packet in packets)
+                new byte[] { 1 },
+                new CultNetRudpSendOptions { Reliable = true, Ordered = true, NowMs = 0 });
+            receiver.Receive(oldPacket);
+            foreach (var packet in Enumerable.Range(0, 40).Select(index => sender.Send(
+                "activity",
+                new byte[] { (byte)index },
+                new CultNetRudpSendOptions { Reliable = false, Ordered = false, NowMs = 0 })))
                 receiver.Receive(packet);
 
             sender.Receive(receiver.CreateAck());
@@ -803,7 +806,10 @@ namespace GameCult.Networking.Tests
 
             var oldRetransmit = sender.DueResends(10).Single(packet => packet.Sequence == 1);
             receiver.Receive(oldRetransmit);
-            sender.Receive(receiver.CreateAck(oldRetransmit.Sequence));
+            var exactAck = receiver.CreateAckForReceived(oldRetransmit.Sequence);
+            Assert.That(exactAck.Ack, Is.EqualTo(oldRetransmit.Sequence));
+            Assert.That(exactAck.AckMask, Is.Zero);
+            sender.Receive(exactAck);
 
             Assert.That(sender.PendingReliableSequences, Does.Not.Contain(1u));
         }
@@ -824,7 +830,7 @@ namespace GameCult.Networking.Tests
                 Sequence = 5000,
                 ChannelId = "control"
             });
-            var packets = Enumerable.Range(0, 1000)
+            var packets = Enumerable.Range(0, CultNetRudpSession.ReliableSendWindowPackets)
                 .Select(index => session.Send("schema", Encoding.UTF8.GetBytes(index.ToString()),
                     new CultNetRudpSendOptions { Reliable = true, Ordered = true, NowMs = 0 }))
                 .ToArray();