From 75c180782aeba7cfd22d6412877397708a4ed28f Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 23 Aug 2026 19:15:22 +0200 Subject: [PATCH] Acknowledge retransmits across RUDP horizons --- docs/cultnet-transport-parity.md | 7 ++++-- .../cultcache-py/src/cultmesh_py/server.py | 6 ++++- .../src/cultnet_py/interop_peer.py | 2 +- .../cultcache-py/src/cultnet_py/transport.py | 9 +++++++- packages/cultcache-py/tests/test_cultcache.py | 4 ++++ .../kotlin/org/gamecult/cultmesh/CultMesh.kt | 22 +++++++++++++++++-- packages/cultnet-rs/src/rudp.rs | 13 +++++++++-- packages/cultnet-rs/tests/cultnet.rs | 4 ++++ packages/cultnet-ts/src/operation-service.ts | 4 +++- packages/cultnet-ts/src/rudp.ts | 10 ++++++++- packages/cultnet-ts/test/cultnet.test.ts | 4 ++++ src/GameCult.Networking/CultNetTransport.cs | 16 ++++++++++++-- .../NetworkingTests.cs | 20 +++++++++++------ 13 files changed, 101 insertions(+), 20 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 35e33400..aa962c46 100644 --- a/docs/cultnet-transport-parity.md +++ b/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 diff --git a/packages/cultcache-py/src/cultmesh_py/server.py b/packages/cultcache-py/src/cultmesh_py/server.py index 8d6653af..6f124480 100644 --- a/packages/cultcache-py/src/cultmesh_py/server.py +++ b/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, diff --git a/packages/cultcache-py/src/cultnet_py/interop_peer.py b/packages/cultcache-py/src/cultnet_py/interop_peer.py index 2fe62b60..7ee9995e 100644 --- a/packages/cultcache-py/src/cultnet_py/interop_peer.py +++ b/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", diff --git a/packages/cultcache-py/src/cultnet_py/transport.py b/packages/cultcache-py/src/cultnet_py/transport.py index 2a83a218..9ce4e4de 100644 --- a/packages/cultcache-py/src/cultnet_py/transport.py +++ b/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: diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index e6896f47..2dbbeb2d 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/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() diff --git a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt index 0c84cfa9..6245fb9e 100644 --- a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt +++ b/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() { diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index a5c5107c..25382264 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/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) -> 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) diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index ce291be2..a66411f0 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/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(()) } diff --git a/packages/cultnet-ts/src/operation-service.ts b/packages/cultnet-ts/src/operation-service.ts index 0b6f6b31..ee13457f 100644 --- a/packages/cultnet-ts/src/operation-service.ts +++ b/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 } { diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index 6c69d1df..51b5cc5b 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/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))); diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index 8b64a38e..e213ef6f 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/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", () => { diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index f0e7f869..6e706fbc 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -1193,6 +1193,18 @@ namespace GameCult.Networking }; } + /// + /// Creates a cumulative acknowledgement when the received packet is inside the + /// acknowledgement horizon, or an exact acknowledgement for an older retransmit. + /// + public CultNetRudpPacket CreateAckForReceived(uint receivedSequence) + { + var (ack, _) = AckState(); + return ack >= receivedSequence && ack - receivedSequence <= 32 + ? CreateAck() + : CreateAck(receivedSequence); + } + /// /// Creates a packet carrying a keepalive ping payload. /// @@ -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) { diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 41bb2895..33aff57d 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/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();