From fda51fad50bee39106915fd37c842413f46efc7a Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 23 Aug 2026 19:00:33 +0200 Subject: [PATCH] Acknowledge reliable RUDP windows cumulatively --- .../cultcache-py/src/cultmesh_py/server.py | 2 +- .../src/cultnet_py/interop_peer.py | 2 +- .../cultcache-py/src/cultnet_py/transport.py | 13 ++++++++-- packages/cultcache-py/tests/test_cultcache.py | 8 ++++--- .../kotlin/org/gamecult/cultmesh/CultMesh.kt | 24 ++++++++++++++----- packages/cultnet-rs/src/rudp.rs | 6 ++--- packages/cultnet-rs/tests/cultnet.rs | 12 ++++++---- packages/cultnet-ts/test/cultnet.test.ts | 9 ++++--- src/GameCult.Networking/CultNetTransport.cs | 4 ++-- .../NetworkingTests.cs | 7 +++--- 10 files changed, 59 insertions(+), 28 deletions(-) diff --git a/packages/cultcache-py/src/cultmesh_py/server.py b/packages/cultcache-py/src/cultmesh_py/server.py index 0de99729..8d6653af 100644 --- a/packages/cultcache-py/src/cultmesh_py/server.py +++ b/packages/cultcache-py/src/cultmesh_py/server.py @@ -268,7 +268,7 @@ 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_for(packet.sequence)) + self._send_rudp_packet(rudp_socket, remote_addr, peer.session.create_ack()) 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 98fcf219..2fe62b60 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_for(packet.sequence)) + send_rudp_packet(sock, remote, peer.session.create_ack()) 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 70af8f48..2a83a218 100644 --- a/packages/cultcache-py/src/cultnet_py/transport.py +++ b/packages/cultcache-py/src/cultnet_py/transport.py @@ -599,7 +599,16 @@ class CultNetRudpSession: ) def create_ack(self) -> CultNetRudpPacket: - return self._create_packet(CultNetRudpPacketType.ACK, "control", b"") + ack, ack_mask = self._ack_state() + return CultNetRudpPacket( + packet_type=CultNetRudpPacketType.ACK, + connection_id=self.connection_id, + sequence=0, + ack=ack, + ack_mask=ack_mask, + channel_id="control", + payload=b"", + ) def create_ack_for(self, sequence: int) -> CultNetRudpPacket: return CultNetRudpPacket( @@ -955,7 +964,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_for(packet.sequence)) + self._send_packet(self.session.create_ack()) 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 c6dedd65..e6896f47 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -1178,9 +1178,11 @@ class CultCacheTests(unittest.TestCase): delivered = [] while wire: - packet = wire.pop(0) - delivered.extend(receiver.receive(packet, 2).delivered) - acknowledged = sender.receive(receiver.create_ack_for(packet.sequence), 3) + admitted = len(wire) + for _ in range(admitted): + packet = wire.pop(0) + delivered.extend(receiver.receive(packet, 2).delivered) + acknowledged = sender.receive(receiver.create_ack(), 3) wire.extend(acknowledged.ready_to_send) self.assertEqual(sender.outstanding_reliable_packet_count, 0) 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 bd0f269f..0c84cfa9 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 @@ -2764,7 +2764,17 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { ) } - fun createAck(): CultNetRudpPacket = createPacket(CultNetRudpPacketType.Ack, "control", ByteArray(0)) + fun createAck(): CultNetRudpPacket { + val (ack, ackMask) = ackState() + return CultNetRudpPacket( + packetType = CultNetRudpPacketType.Ack, + connectionId = connectionId, + sequence = 0, + ack = ack, + ackMask = ackMask, + channelId = "control", + ) + } fun createAckFor(sequence: Long): CultNetRudpPacket = CultNetRudpPacket( packetType = CultNetRudpPacketType.Ack, @@ -3205,7 +3215,7 @@ class CultNetRudpSocketTransportConnection( framesReceived += 1 } val frame = if (delivered.isEmpty()) null else delivered.removeFirst() - if (packet.reliable || packet.packetType == CultNetRudpPacketType.Accept || frame != null) sendPacket(session.createAckFor(packet.sequence)) + if (packet.reliable || packet.packetType == CultNetRudpPacketType.Accept || frame != null) sendPacket(session.createAck()) return frame } @@ -3972,7 +3982,7 @@ 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.createAckFor(packet.sequence)) + if (packet.reliable || packet.packetType == CultNetRudpPacketType.Data) sendInteropRudpPacket(socket, remote, session.createAck()) for (frame in result.delivered) { if (frame.channelId != "schema") continue val sender = InteropRudpSessionSender(socket, remote, session) @@ -5323,9 +5333,11 @@ private fun rudpSessionAdvancesLargeFragmentSetsThroughBoundedReliableWindow() { val delivered = mutableListOf() while (!wire.isEmpty()) { - val packet = wire.removeFirst() - delivered.addAll(receiver.receive(packet, 2).delivered) - val acknowledged = sender.receive(receiver.createAckFor(packet.sequence), 3) + val admitted = wire.size + repeat(admitted) { + delivered.addAll(receiver.receive(wire.removeFirst(), 2).delivered) + } + val acknowledged = sender.receive(receiver.createAck(), 3) acknowledged.readyToSend.forEach { wire.addLast(it) } } check(sender.outstandingReliablePacketCount == 0) diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index f84d8cae..a5c5107c 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -1288,7 +1288,7 @@ impl CultNetRudpServerHub { let reply = peer .session .pending_accept_for_resend(now_ms()) - .unwrap_or_else(|| peer.session.create_ack_for(packet.sequence)); + .unwrap_or_else(|| peer.session.create_ack()); self.send_packet(remote_addr, &reply)?; return Ok(None); } @@ -1340,7 +1340,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_for(packet.sequence)) + Some(peer.session.create_ack()) } else { None }; @@ -1657,7 +1657,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_for(packet.sequence); + let ack = self.session.create_ack(); self.send_packet(&ack)?; } Ok(frame) diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index 1669fd0e..8705cb4b 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -1308,10 +1308,14 @@ fn rudp_session_advances_large_fragment_sets_through_a_bounded_reliable_window() let mut wire = VecDeque::from(initial); let mut delivered = Vec::new(); - while let Some(packet) = wire.pop_front() { - let received = receiver.receive(&packet, 2)?; - delivered.extend(received.delivered); - let acknowledged = sender.receive(&receiver.create_ack_for(packet.sequence), 3)?; + while !wire.is_empty() { + let admitted = wire.len(); + for _ in 0..admitted { + let packet = wire.pop_front().expect("admitted packet remains queued"); + let received = receiver.receive(&packet, 2)?; + delivered.extend(received.delivered); + } + let acknowledged = sender.receive(&receiver.create_ack(), 3)?; wire.extend(acknowledged.ready_to_send); } diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index 621949aa..8b64a38e 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -1322,9 +1322,12 @@ test("rudp session advances large fragment sets through a bounded reliable windo const delivered = []; while (wire.length > 0) { - const packet = wire.shift()!; - delivered.push(...receiver.receive(packet, 2).delivered); - const acknowledged = sender.receive(receiver.createAckFor(packet.sequence), 3); + const admitted = wire.length; + for (let index = 0; index < admitted; index += 1) { + const packet = wire.shift()!; + delivered.push(...receiver.receive(packet, 2).delivered); + } + const acknowledged = sender.receive(receiver.createAck(), 3); wire.push(...(acknowledged.readyToSend ?? [])); } diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index e65eb966..f0e7f869 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -2077,7 +2077,7 @@ namespace GameCult.Networking SendPackets(result.ReadyToSend); if (packet.PacketType == CultNetRudpPacketType.Accept || packet.PacketType == CultNetRudpPacketType.Data) - acknowledgement = _session.CreateAck(packet.Sequence); + acknowledgement = _session.CreateAck(); } if (result.Pong) { @@ -2486,7 +2486,7 @@ namespace GameCult.Networking SendPacket(existingPeer.RemoteEndPoint, result.Reply); SendPackets(existingPeer, result.ReadyToSend); if (packet.PacketType == CultNetRudpPacketType.Data) - acknowledgement = existingPeer.Session.CreateAck(packet.Sequence); + acknowledgement = existingPeer.Session.CreateAck(); } if (result.Disconnected) { diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 12d397f0..41bb2895 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1096,9 +1096,10 @@ namespace GameCult.Networking.Tests var delivered = new List(); while (wire.Count > 0) { - var packet = wire.Dequeue(); - delivered.AddRange(receiver.Receive(packet, 2).Delivered); - var acknowledged = sender.Receive(receiver.CreateAck(packet.Sequence), 3); + var admitted = wire.Count; + for (var index = 0; index < admitted; index++) + delivered.AddRange(receiver.Receive(wire.Dequeue(), 2).Delivered); + var acknowledged = sender.Receive(receiver.CreateAck(), 3); foreach (var ready in acknowledged.ReadyToSend) wire.Enqueue(ready); }