diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 5f0fec91..e9fd0ef8 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -258,11 +258,16 @@ Current progress: echo against C#/Kotlin/Python/Rust UDP peers; local runtime tests prove timeout state transitions. - TypeScript, C#, Rust, Python, and Kotlin now share bounded reliable-send - backpressure for RUDP sessions. `maxPendingReliablePackets` is advertised on - RUDP transport-profile channels and enforced by the session before reliable - connect, accept, single-packet sends, or fragmented sends enqueue anything; - TCP-framed profiles do not publish the field because they do not own the RUDP - pending-reliable queue. + backpressure for RUDP sessions. At most 64 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 + 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 + reliable frame is accepted. Socket transports expose an explicit reliable + flush where the runtime supports one, so callers that require confirmed + transport delivery do not substitute a fixed sleep. TCP-framed profiles do + not publish the field because they do not own the RUDP pending-reliable queue. - TypeScript, C#, Rust, Python, and Kotlin now share a portable `cultnet.reconnect_policy.v0` document and deterministic exponential delay helper, and RUDP transport profiles advertise that policy under diff --git a/packages/cultcache-py/src/cultmesh_py/server.py b/packages/cultcache-py/src/cultmesh_py/server.py index d560cc83..0de99729 100644 --- a/packages/cultcache-py/src/cultmesh_py/server.py +++ b/packages/cultcache-py/src/cultmesh_py/server.py @@ -245,6 +245,8 @@ class CultMeshLocalServer: result = peer.session.receive(packet, now_ms) if result.reply is not None: self._send_rudp_packet(rudp_socket, remote_addr, result.reply) + for ready in result.ready_to_send: + self._send_rudp_packet(rudp_socket, remote_addr, ready) if result.disconnected: peers.pop(remote_addr, None) continue @@ -265,8 +267,8 @@ class CultMeshLocalServer: peer.session, msgpack.packb(response, use_bin_type=True), ) - if packet.packet_type == CultNetRudpPacketType.DATA or result.delivered: - self._send_rudp_packet(rudp_socket, remote_addr, peer.session.create_ack()) + 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)) 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 d26f1b70..98fcf219 100644 --- a/packages/cultcache-py/src/cultnet_py/interop_peer.py +++ b/packages/cultcache-py/src/cultnet_py/interop_peer.py @@ -306,6 +306,8 @@ def rudp_loop( result = peer.session.receive(packet, now_ms()) if result.reply is not None: send_rudp_packet(sock, remote, result.reply) + for ready in result.ready_to_send: + send_rudp_packet(sock, remote, ready) if result.disconnected: peers.pop(remote, None) continue @@ -317,8 +319,8 @@ def rudp_loop( continue for response in handle_server_message(state, message, peer.subscriptions): send_rudp_schema_frame(sock, remote, peer.session, response) - if packet.packet_type == CultNetRudpPacketType.DATA or result.delivered: - send_rudp_packet(sock, remote, peer.session.create_ack()) + if packet.reliable or packet.packet_type == CultNetRudpPacketType.DATA or result.delivered: + send_rudp_packet(sock, remote, peer.session.create_ack_for(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 25a4c891..ca877a2a 100644 --- a/packages/cultcache-py/src/cultnet_py/transport.py +++ b/packages/cultcache-py/src/cultnet_py/transport.py @@ -280,6 +280,7 @@ class CultNetRudpDeliveredFrame: @dataclass(frozen=True) class CultNetRudpReceiveResult: delivered: tuple[CultNetRudpDeliveredFrame, ...] = field(default_factory=tuple) + ready_to_send: tuple[CultNetRudpPacket, ...] = field(default_factory=tuple) reply: CultNetRudpPacket | None = None pong: bool = False pong_payload: bytes = b"" @@ -395,6 +396,9 @@ def create_rudp_transport_profile( class CultNetRudpSession: + RELIABLE_SEND_WINDOW_PACKETS = 64 + RECEIVED_SEQUENCE_WINDOW = 4_096 + def __init__(self, options: CultNetRudpSessionOptions) -> None: self.connection_id = _uint32(options.connection_id, "connection_id") self.resend_delay_ms = options.resend_delay_ms @@ -408,6 +412,7 @@ class CultNetRudpSession: self._highest_received_sequence: int | None = None self._received_sequences: set[int] = set() self._pending_reliable: dict[int, _PendingReliablePacket] = {} + self._queued_reliable: deque[CultNetRudpPacket] = deque() self._ordered_next_sequence_by_channel: dict[str, int] = {} self._ordered_buffers: dict[str, dict[int, tuple[CultNetRudpDeliveredFrame, int]]] = {} self._fragment_buffers: dict[tuple[str, int], dict[str, Any]] = {} @@ -424,6 +429,14 @@ class CultNetRudpSession: def pending_reliable_sequences(self) -> tuple[int, ...]: return tuple(sorted(self._pending_reliable)) + @property + def queued_reliable_packet_count(self) -> int: + return len(self._queued_reliable) + + @property + def outstanding_reliable_packet_count(self) -> int: + return len(self._pending_reliable) + len(self._queued_reliable) + def create_connect(self, now_ms: int = 0, payload: bytes = b"") -> CultNetRudpPacket: self._ensure_reliable_capacity(1) packet = self._create_packet( @@ -465,7 +478,10 @@ class CultNetRudpSession: payload: bytes, options: CultNetRudpSendOptions | None = None, ) -> CultNetRudpPacket: - return self.send_many(channel_id, payload, options)[0] + resolved = options or CultNetRudpSendOptions() + if resolved.reliable and len(self._pending_reliable) >= self.RELIABLE_SEND_WINDOW_PACKETS: + raise ValueError("RUDP reliable send window is full; receive acknowledgements before sending") + return self.send_many(channel_id, payload, resolved)[0] def send_many( self, @@ -492,9 +508,7 @@ class CultNetRudpSession: ordered=options.ordered, sequenced=options.sequenced, ) - if packet.reliable: - self._track_reliable(packet, options.now_ms) - return (packet,) + return self._admit_reliable_packets((packet,), options.now_ms) if packet.reliable else (packet,) fragment_count = (len(payload) + max_fragment_bytes - 1) // max_fragment_bytes if fragment_count > 0xFFFF: @@ -516,14 +530,13 @@ class CultNetRudpSession: fragment_index=index, fragment_count=fragment_count, ) - if packet.reliable: - self._track_reliable(packet, options.now_ms) packets.append(packet) - return tuple(packets) + return self._admit_reliable_packets(tuple(packets), options.now_ms) if options.reliable else tuple(packets) def receive(self, packet: CultNetRudpPacket, now_ms: int = 0) -> CultNetRudpReceiveResult: self._require_connection(packet) self._apply_acknowledgements(packet) + ready_to_send = self._promote_queued_reliable(now_ms) self._last_received_at_ms = now_ms expected_sequence_if_uninitialized = ( packet.sequence @@ -534,17 +547,19 @@ class CultNetRudpSession: if packet.packet_type == CultNetRudpPacketType.ACCEPT: self._remember_received(packet.sequence) self._connected = True - return CultNetRudpReceiveResult() + return CultNetRudpReceiveResult(ready_to_send=ready_to_send) if packet.packet_type == CultNetRudpPacketType.PING: self._remember_received(packet.sequence) return CultNetRudpReceiveResult( + ready_to_send=ready_to_send, reply=self._create_packet(CultNetRudpPacketType.PONG, "control", packet.payload) ) if packet.packet_type in (CultNetRudpPacketType.ACK, CultNetRudpPacketType.PONG): self._remember_received(packet.sequence) return CultNetRudpReceiveResult( + ready_to_send=ready_to_send, pong=packet.packet_type == CultNetRudpPacketType.PONG, pong_payload=bytes(packet.payload) if packet.packet_type == CultNetRudpPacketType.PONG else b"", ) @@ -553,33 +568,50 @@ class CultNetRudpSession: self._remember_received(packet.sequence) self._connected = False return CultNetRudpReceiveResult( + ready_to_send=ready_to_send, disconnected=True, disconnect_reason=bytes(packet.payload), ) if packet.packet_type != CultNetRudpPacketType.DATA: - return CultNetRudpReceiveResult() + return CultNetRudpReceiveResult(ready_to_send=ready_to_send) - duplicate = packet.sequence in self._received_sequences + duplicate = packet.sequence in self._received_sequences or ( + self._highest_received_sequence is not None + and packet.sequence < self._highest_received_sequence + and self._highest_received_sequence - packet.sequence >= self.RECEIVED_SEQUENCE_WINDOW + ) self._remember_received(packet.sequence) if duplicate: - return CultNetRudpReceiveResult() + return CultNetRudpReceiveResult(ready_to_send=ready_to_send) reassembled = self._reassemble(packet) if reassembled is None: - return CultNetRudpReceiveResult() + return CultNetRudpReceiveResult(ready_to_send=ready_to_send) frame, ordered, next_sequence = reassembled if not ordered: - return CultNetRudpReceiveResult(delivered=(frame,)) + return CultNetRudpReceiveResult(delivered=(frame,), ready_to_send=ready_to_send) return CultNetRudpReceiveResult( delivered=tuple( self._deliver_ordered(frame, next_sequence, expected_sequence_if_uninitialized) - ) + ), + ready_to_send=ready_to_send, ) def create_ack(self) -> CultNetRudpPacket: return self._create_packet(CultNetRudpPacketType.ACK, "control", b"") + def create_ack_for(self, sequence: int) -> CultNetRudpPacket: + return CultNetRudpPacket( + packet_type=CultNetRudpPacketType.ACK, + connection_id=self.connection_id, + sequence=0, + ack=_uint32(sequence, "ack sequence"), + ack_mask=0, + channel_id="control", + payload=b"", + ) + def create_ping(self, payload: bytes = b"") -> CultNetRudpPacket: return self._create_packet(CultNetRudpPacketType.PING, "control", payload) @@ -655,10 +687,31 @@ class CultNetRudpSession: last_sent_at_ms=now_ms, ) + def _admit_reliable_packets( + self, + packets: tuple[CultNetRudpPacket, ...], + now_ms: int, + ) -> tuple[CultNetRudpPacket, ...]: + available = max(0, self.RELIABLE_SEND_WINDOW_PACKETS - len(self._pending_reliable)) + ready = packets[:available] + for packet in ready: + self._track_reliable(packet, now_ms) + self._queued_reliable.extend(packets[available:]) + return ready + + def _promote_queued_reliable(self, now_ms: int) -> tuple[CultNetRudpPacket, ...]: + available = max(0, self.RELIABLE_SEND_WINDOW_PACKETS - len(self._pending_reliable)) + ready: list[CultNetRudpPacket] = [] + while len(ready) < available and self._queued_reliable: + packet = self._queued_reliable.popleft() + self._track_reliable(packet, now_ms) + ready.append(packet) + return tuple(ready) + def _ensure_reliable_capacity(self, packet_count: int) -> None: if packet_count == 0 or self.max_pending_reliable_packets is None: return - if len(self._pending_reliable) + packet_count > self.max_pending_reliable_packets: + if self.outstanding_reliable_packet_count + packet_count > self.max_pending_reliable_packets: raise ValueError("RUDP reliable send queue is full") def _apply_acknowledgements(self, packet: CultNetRudpPacket) -> None: @@ -671,6 +724,9 @@ class CultNetRudpSession: self._received_sequences.add(sequence) if self._highest_received_sequence is None or sequence > self._highest_received_sequence: self._highest_received_sequence = sequence + if len(self._received_sequences) > self.RECEIVED_SEQUENCE_WINDOW: + keep_from = max(0, (self._highest_received_sequence or sequence) - self.RECEIVED_SEQUENCE_WINDOW + 1) + self._received_sequences = {received for received in self._received_sequences if received >= keep_from} def _ack_state(self) -> tuple[int, int]: ack = self._highest_received_sequence or 0 @@ -838,6 +894,10 @@ class CultNetRudpSocketTransportConnection: frames_sent=self._frames_sent, ) + @property + def outstanding_reliable_packet_count(self) -> int: + return self.session.outstanding_reliable_packet_count + def connect(self, payload: bytes = b"") -> None: if self.mode != CultNetRudpSocketMode.CLIENT: raise ValueError("Only a client RUDP socket transport can initiate connect") @@ -879,6 +939,8 @@ class CultNetRudpSocketTransportConnection: result = self.session.receive(packet, _now_ms()) if result.reply is not None: self._send_packet(result.reply) + for ready in result.ready_to_send: + self._send_packet(ready) if result.pong: self.pong_payloads.append(result.pong_payload) if result.disconnected: @@ -892,8 +954,8 @@ class CultNetRudpSocketTransportConnection: self._frames_received += 1 frame = self._delivered_frames.popleft() if self._delivered_frames else None - if packet.packet_type == CultNetRudpPacketType.ACCEPT or frame is not None: - self._send_packet(self.session.create_ack()) + if packet.reliable or packet.packet_type == CultNetRudpPacketType.ACCEPT or frame is not None: + self._send_packet(self.session.create_ack_for(packet.sequence)) return frame def receive(self, timeout_seconds: float | None = None) -> CultNetTransportFrame: @@ -906,6 +968,29 @@ class CultNetRudpSocketTransportConnection: if deadline is not None and time.monotonic() >= deadline: raise TimeoutError("Timed out waiting for RUDP schema frame") + def flush_reliable(self, timeout_seconds: float = 30.0) -> None: + deadline = time.monotonic() + max(0.0, timeout_seconds) + original_timeout = self.socket.gettimeout() + poll_timeout = min(original_timeout, 0.01) if original_timeout is not None else 0.01 + preserved = deque(self._delivered_frames) + self._delivered_frames.clear() + self.socket.settimeout(poll_timeout) + try: + while self.session.outstanding_reliable_packet_count > 0: + if time.monotonic() >= deadline: + raise TimeoutError( + "RUDP reliable flush timed out with " + f"{self.session.outstanding_reliable_packet_count} packets outstanding" + ) + frame = self.receive_once() + if frame is not None: + preserved.append(frame) + self.poll_resends() + finally: + preserved.extend(self._delivered_frames) + self._delivered_frames = preserved + self.socket.settimeout(original_timeout) + def disconnect(self, reason: bytes = b"") -> None: self._send_packet(self.session.create_disconnect(reason)) diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 82a00845..c6dedd65 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -1157,6 +1157,36 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(delivered[0].payload, b"fragment-me-please") self.assertEqual(delivered[0].sequence, packets[0].sequence) + def test_cultnet_rudp_session_advances_large_fragment_sets_through_bounded_window(self) -> None: + connection_id = 457 + sender = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id, initial_sequence=1)) + receiver = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id, initial_sequence=100)) + sender.receive(CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, connection_id, 0, 0, 0, "control")) + receiver.receive(CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, connection_id, 0, 0, 0, "control")) + + fragment_count = CultNetRudpSession.RELIABLE_SEND_WINDOW_PACKETS + 17 + payload = bytes(index % 251 for index in range(fragment_count * 8)) + wire = list(sender.send_many( + "schema", + payload, + CultNetRudpSendOptions(reliable=True, ordered=True, now_ms=1), + max_fragment_bytes=8, + )) + self.assertEqual(len(wire), CultNetRudpSession.RELIABLE_SEND_WINDOW_PACKETS) + self.assertEqual(len(sender.pending_reliable_sequences), len(wire)) + self.assertEqual(sender.queued_reliable_packet_count, 17) + + delivered = [] + while wire: + packet = wire.pop(0) + delivered.extend(receiver.receive(packet, 2).delivered) + acknowledged = sender.receive(receiver.create_ack_for(packet.sequence), 3) + wire.extend(acknowledged.ready_to_send) + + self.assertEqual(sender.outstanding_reliable_packet_count, 0) + self.assertEqual(len(delivered), 1) + self.assertEqual(delivered[0].payload, payload) + 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/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt index 968e7669..a7f90610 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 @@ -2587,6 +2587,7 @@ data class CultNetRudpPacket( data class CultNetRudpDeliveredFrame(val channelId: String, val payload: ByteArray, val sequence: Long) data class CultNetRudpReceiveResult( val delivered: List = emptyList(), + val readyToSend: List = emptyList(), val reply: CultNetRudpPacket? = null, val pong: Boolean = false, val pongPayload: ByteArray = ByteArray(0), @@ -2619,6 +2620,11 @@ private data class FragmentBuffer( ) class CultNetRudpSession(options: CultNetRudpSessionOptions) { + companion object { + const val ReliableSendWindowPackets = 64 + private const val ReceivedSequenceWindow = 4_096 + } + val connectionId: Long = uint32(options.connectionId, "connectionId") val resendDelayMs: Long = options.resendDelayMs private val maxPendingReliablePackets: Int? = options.maxPendingReliablePackets @@ -2631,12 +2637,17 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { private var highestReceivedSequence: Long? = null private val receivedSequences = linkedSetOf() private val pendingReliable = linkedMapOf() + private val queuedReliable = ArrayDeque() private val orderedNextSequenceByChannel = linkedMapOf() private val orderedBuffers = linkedMapOf>() private val fragmentBuffers = linkedMapOf, FragmentBuffer>() val pendingReliableSequences: List get() = pendingReliable.keys.sorted() + val queuedReliablePacketCount: Int + get() = queuedReliable.size + val outstandingReliablePacketCount: Int + get() = pendingReliable.size + queuedReliable.size init { if (maxPendingReliablePackets != null && maxPendingReliablePackets <= 0) { @@ -2663,6 +2674,9 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { } fun send(channelId: String, payload: ByteArray, options: CultNetRudpSendOptions = CultNetRudpSendOptions()): CultNetRudpPacket { + if (options.reliable && pendingReliable.size >= ReliableSendWindowPackets) { + throw IOException("RUDP reliable send window is full; receive acknowledgements before sending") + } return sendMany(channelId, payload, options).first() } @@ -2679,7 +2693,7 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { if (fragmentCount > 0xffff) throw IOException("RUDP payload requires more than 65535 fragments") ensureReliableCapacity(if (options.reliable) fragmentCount else 0) val fragmentId = allocateFragmentId() - return (0 until fragmentCount).map { index -> + val packets = (0 until fragmentCount).map { index -> val start = index * maxFragmentBytes val packet = createPacket( channelId = channelId, @@ -2692,14 +2706,13 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { fragmentIndex = index, fragmentCount = fragmentCount, ) - if (packet.reliable) trackReliable(packet, options.nowMs) packet } + return if (options.reliable) admitReliablePackets(packets, options.nowMs) else packets } ensureReliableCapacity(if (options.reliable) 1 else 0) val packet = createPacket(channelId = channelId, packetType = CultNetRudpPacketType.Data, payload = payload, reliable = options.reliable, ordered = options.ordered, sequenced = options.sequenced) - if (packet.reliable) trackReliable(packet, options.nowMs) - return listOf(packet) + return if (packet.reliable) admitReliablePackets(listOf(packet), options.nowMs) else listOf(packet) } fun receive(packet: CultNetRudpPacket, nowMs: Long = 0): CultNetRudpReceiveResult { @@ -2707,21 +2720,23 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { val ignoredNow = nowMs requireConnection(packet) applyAcknowledgements(packet) + val readyToSend = promoteQueuedReliable(nowMs) lastReceivedAtMs = nowMs val expectedSequenceIfUninitialized = (highestReceivedSequence ?: (packet.sequence - 1)) + 1 when (packet.packetType) { CultNetRudpPacketType.Accept -> { rememberReceived(packet.sequence) connected = true - return CultNetRudpReceiveResult() + return CultNetRudpReceiveResult(readyToSend = readyToSend) } CultNetRudpPacketType.Ping -> { rememberReceived(packet.sequence) - return CultNetRudpReceiveResult(reply = createPacket(CultNetRudpPacketType.Pong, "control", packet.payload)) + return CultNetRudpReceiveResult(readyToSend = readyToSend, reply = createPacket(CultNetRudpPacketType.Pong, "control", packet.payload)) } CultNetRudpPacketType.Ack, CultNetRudpPacketType.Pong -> { rememberReceived(packet.sequence) return CultNetRudpReceiveResult( + readyToSend = readyToSend, pong = packet.packetType == CultNetRudpPacketType.Pong, pongPayload = if (packet.packetType == CultNetRudpPacketType.Pong) packet.payload.copyOf() else ByteArray(0), ) @@ -2729,21 +2744,37 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { CultNetRudpPacketType.Disconnect -> { rememberReceived(packet.sequence) connected = false - return CultNetRudpReceiveResult(disconnected = true, disconnectReason = packet.payload.copyOf()) + return CultNetRudpReceiveResult(readyToSend = readyToSend, disconnected = true, disconnectReason = packet.payload.copyOf()) } CultNetRudpPacketType.Data -> Unit - else -> return CultNetRudpReceiveResult() + else -> return CultNetRudpReceiveResult(readyToSend = readyToSend) } - val duplicate = receivedSequences.contains(packet.sequence) + val duplicate = receivedSequences.contains(packet.sequence) || ( + highestReceivedSequence != null && + packet.sequence < highestReceivedSequence!! && + highestReceivedSequence!! - packet.sequence >= ReceivedSequenceWindow.toLong() + ) rememberReceived(packet.sequence) - if (duplicate) return CultNetRudpReceiveResult() - val reassembled = reassemble(packet) ?: return CultNetRudpReceiveResult() - return CultNetRudpReceiveResult(delivered = if (reassembled.ordered) deliverOrdered(reassembled.frame, reassembled.nextSequence, expectedSequenceIfUninitialized) else listOf(reassembled.frame)) + if (duplicate) return CultNetRudpReceiveResult(readyToSend = readyToSend) + val reassembled = reassemble(packet) ?: return CultNetRudpReceiveResult(readyToSend = readyToSend) + return CultNetRudpReceiveResult( + delivered = if (reassembled.ordered) deliverOrdered(reassembled.frame, reassembled.nextSequence, expectedSequenceIfUninitialized) else listOf(reassembled.frame), + readyToSend = readyToSend, + ) } fun createAck(): CultNetRudpPacket = createPacket(CultNetRudpPacketType.Ack, "control", ByteArray(0)) + fun createAckFor(sequence: Long): CultNetRudpPacket = CultNetRudpPacket( + packetType = CultNetRudpPacketType.Ack, + connectionId = connectionId, + sequence = 0, + ack = uint32(sequence, "ack sequence"), + ackMask = 0, + channelId = "control", + ) + fun createPing(payload: ByteArray = ByteArray(0)): CultNetRudpPacket = createPacket(CultNetRudpPacketType.Ping, "control", payload) @@ -2787,9 +2818,28 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { pendingReliable[packet.sequence] = PendingReliablePacket(packet.copy(payload = packet.payload.copyOf()), nowMs) } + private fun admitReliablePackets(packets: List, nowMs: Long): List { + val available = maxOf(0, ReliableSendWindowPackets - pendingReliable.size) + val ready = packets.take(available) + ready.forEach { trackReliable(it, nowMs) } + packets.drop(available).forEach { queuedReliable.addLast(it.copy(payload = it.payload.copyOf())) } + return ready + } + + private fun promoteQueuedReliable(nowMs: Long): List { + val available = maxOf(0, ReliableSendWindowPackets - pendingReliable.size) + val ready = mutableListOf() + while (ready.size < available && !queuedReliable.isEmpty()) { + val packet = queuedReliable.removeFirst() + trackReliable(packet, nowMs) + ready.add(packet) + } + return ready + } + private fun ensureReliableCapacity(packetCount: Int) { if (packetCount == 0 || maxPendingReliablePackets == null) return - if (pendingReliable.size + packetCount > maxPendingReliablePackets) { + if (outstandingReliablePacketCount + packetCount > maxPendingReliablePackets) { throw IOException("RUDP reliable send queue is full") } } @@ -2806,6 +2856,10 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { private fun rememberReceived(sequence: Long) { receivedSequences.add(sequence) if (highestReceivedSequence == null || sequence > highestReceivedSequence!!) highestReceivedSequence = sequence + if (receivedSequences.size > ReceivedSequenceWindow) { + val keepFrom = maxOf(0L, (highestReceivedSequence ?: sequence) - ReceivedSequenceWindow.toLong() + 1) + receivedSequences.removeIf { it < keepFrom } + } } private fun ackState(): Pair { @@ -3010,6 +3064,7 @@ class CultNetRudpSocketTransportConnection( maxPendingReliablePackets = maxPendingReliablePackets, ) val connected: Boolean get() = session.connected + val outstandingReliablePacketCount: Int get() = session.outstandingReliablePacketCount val localPort: Int get() = socket.localPort override val stats: CultNetTransportStats get() = CultNetTransportStats(bytesReceived, bytesSent, framesReceived, framesSent) @@ -3093,6 +3148,27 @@ class CultNetRudpSocketTransportConnection( override fun receiveSchemaMessage(timeoutMs: Long, pollIntervalMs: Long): CultNetMessage? = receiveSchema(timeoutMs, pollIntervalMs)?.let { parseCultNetMessage(it) } + fun flushReliable(timeoutMs: Long = 30_000, pollIntervalMs: Long = 5) { + val deadline = System.nanoTime() + timeoutMs.coerceAtLeast(0) * 1_000_000L + val preserved = ArrayDeque() + while (!delivered.isEmpty()) preserved.addLast(delivered.removeFirst()) + try { + while (session.outstandingReliablePacketCount > 0) { + if (System.nanoTime() >= deadline) { + throw SocketTimeoutException( + "RUDP reliable flush timed out with ${session.outstandingReliablePacketCount} packets outstanding" + ) + } + receiveOnce()?.let { preserved.addLast(it) } + pollResends() + Thread.sleep(pollIntervalMs) + } + } finally { + while (!delivered.isEmpty()) preserved.addLast(delivered.removeFirst()) + while (!preserved.isEmpty()) delivered.addLast(preserved.removeFirst()) + } + } + fun receiveOnce(): CultNetTransportFrame? { if (!delivered.isEmpty()) return delivered.removeFirst() val buffer = ByteArray(65535) @@ -3116,6 +3192,7 @@ class CultNetRudpSocketTransportConnection( } val result = session.receive(packet, nowMs()) result.reply?.let { sendPacket(it) } + result.readyToSend.forEach { sendPacket(it) } if (result.pong) { pongPayloads.add(result.pongPayload.copyOf()) } @@ -3128,7 +3205,7 @@ class CultNetRudpSocketTransportConnection( framesReceived += 1 } val frame = if (delivered.isEmpty()) null else delivered.removeFirst() - if (packet.packetType == CultNetRudpPacketType.Accept || frame != null) sendPacket(session.createAck()) + if (packet.reliable || packet.packetType == CultNetRudpPacketType.Accept || frame != null) sendPacket(session.createAckFor(packet.sequence)) return frame } @@ -3252,6 +3329,7 @@ fun main(args: Array) { rudpSessionPingsAndDetectsReceiveTimeout() rudpSessionBoundsPendingReliablePacketsBeforeEnqueue() rudpSessionFragmentsAndReassemblesReliableOrderedPayloads() + rudpSessionAdvancesLargeFragmentSetsThroughBoundedReliableWindow() rudpSocketTransportErgonomicFactoriesCarrySchemaFrames() rudpSocketTransportHandshakesAndCarriesReliableOrderedSchemaFrames() rudpSocketTransportCarriesFragmentedReliableOrderedSchemaFrames() @@ -3893,7 +3971,8 @@ private fun startInteropRudpServer( } val result = session.receive(packet, nowMs()) result.reply?.let { sendInteropRudpPacket(socket, remote, it) } - if (packet.packetType == CultNetRudpPacketType.Data) sendInteropRudpPacket(socket, remote, session.createAck()) + result.readyToSend.forEach { sendInteropRudpPacket(socket, remote, it) } + if (packet.reliable || packet.packetType == CultNetRudpPacketType.Data) sendInteropRudpPacket(socket, remote, session.createAckFor(packet.sequence)) for (frame in result.delivered) { if (frame.channelId != "schema") continue val sender = InteropRudpSessionSender(socket, remote, session) @@ -5223,6 +5302,37 @@ private fun rudpSessionFragmentsAndReassemblesReliableOrderedPayloads() { check(delivered.first().sequence == packets.first().sequence) } +private fun rudpSessionAdvancesLargeFragmentSetsThroughBoundedReliableWindow() { + val connectionId = 457L + val sender = CultNetRudpSession(CultNetRudpSessionOptions(connectionId = connectionId, initialSequence = 1)) + val receiver = CultNetRudpSession(CultNetRudpSessionOptions(connectionId = connectionId, initialSequence = 100)) + sender.receive(CultNetRudpPacket(CultNetRudpPacketType.Accept, connectionId, 0, 0, 0, "control")) + receiver.receive(CultNetRudpPacket(CultNetRudpPacketType.Accept, connectionId, 0, 0, 0, "control")) + + val fragmentCount = CultNetRudpSession.ReliableSendWindowPackets + 17 + val payload = ByteArray(fragmentCount * 8) { index -> (index % 251).toByte() } + val wire = ArrayDeque(sender.sendMany( + "schema", + payload, + CultNetRudpSendOptions(reliable = true, ordered = true, nowMs = 1), + maxFragmentBytes = 8, + )) + check(wire.size == CultNetRudpSession.ReliableSendWindowPackets) + check(sender.pendingReliableSequences.size == wire.size) + check(sender.queuedReliablePacketCount == 17) + + 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) + acknowledged.readyToSend.forEach { wire.addLast(it) } + } + check(sender.outstandingReliablePacketCount == 0) + check(delivered.size == 1) + check(delivered.first().payload.contentEquals(payload)) +} + private fun rudpSocketTransportErgonomicFactoriesCarrySchemaFrames() { val connectionId = 0x10203042L CultMesh.createRudpServer( diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index 0d1ba965..2ed6e4b7 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -5816,6 +5816,9 @@ export class CultMesh { if (result.reply) { socket.send(encodeRudpPacket(result.reply), record.remote.port, record.remote.address); } + for (const ready of result.readyToSend ?? []) { + socket.send(encodeRudpPacket(ready), record.remote.port, record.remote.address); + } for (const frame of result.delivered) { if (frame.channelId !== "schema") { continue; @@ -6078,7 +6081,7 @@ export class CultMesh { tags: options.tags, }, )); - await new Promise((resolve) => setTimeout(resolve, Math.max(0, options.flushTimeoutMs ?? 150))); + await peer.flush(options.flushTimeoutMs ?? 30_000); } finally { peer.close(); } diff --git a/packages/cultnet-rs/examples/cultnet_interop_peer.rs b/packages/cultnet-rs/examples/cultnet_interop_peer.rs index ab00c6c8..1ed70a35 100644 --- a/packages/cultnet-rs/examples/cultnet_interop_peer.rs +++ b/packages/cultnet-rs/examples/cultnet_interop_peer.rs @@ -986,6 +986,9 @@ fn run_rudp_server( if let Some(reply) = result.reply { send_rudp_packet(&socket, remote_addr, &reply)?; } + for ready in result.ready_to_send { + send_rudp_packet(&socket, remote_addr, &ready)?; + } if packet.packet_type == CultNetRudpPacketType::Data { let ack = peer.session.create_ack(); send_rudp_packet(&socket, remote_addr, &ack)?; diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index ae1ef00b..7cae0b28 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -6,8 +6,7 @@ use std::collections::VecDeque; use std::io::ErrorKind; use std::net::SocketAddr; use std::net::UdpSocket; -use std::time::SystemTime; -use std::time::UNIX_EPOCH; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use crate::CultNetMessage; use crate::CultNetReconnectController; @@ -30,6 +29,7 @@ const RUDP_MAGIC: [u8; 4] = [0x43, 0x4e, 0x52, 0x30]; const RUDP_VERSION: u8 = 0; const RUDP_FIXED_HEADER_BYTES: usize = 36; pub const CULTNET_RUDP_DEFAULT_MAX_PAYLOAD_BYTES: usize = 16 * 1024 * 1024; +pub const CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS: usize = 64; const RUDP_RECEIVED_SEQUENCE_WINDOW: usize = 4_096; #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -70,6 +70,7 @@ pub struct CultNetRudpDeliveredFrame { #[derive(Clone, Debug, PartialEq, Eq)] pub struct CultNetRudpReceiveResult { pub delivered: Vec, + pub ready_to_send: Vec, pub reply: Option, pub pong: bool, pub pong_payload: Vec, @@ -139,6 +140,7 @@ pub struct CultNetRudpSession { highest_received_sequence: Option, received_sequences: BTreeSet, pending_reliable: BTreeMap, + queued_reliable: VecDeque, ordered_next_sequence_by_channel: BTreeMap, ordered_buffers: BTreeMap>, fragment_buffers: BTreeMap<(String, u16), FragmentBuffer>, @@ -160,6 +162,7 @@ impl CultNetRudpSession { highest_received_sequence: None, received_sequences: BTreeSet::new(), pending_reliable: BTreeMap::new(), + queued_reliable: VecDeque::new(), ordered_next_sequence_by_channel: BTreeMap::new(), ordered_buffers: BTreeMap::new(), fragment_buffers: BTreeMap::new(), @@ -182,6 +185,14 @@ impl CultNetRudpSession { self.pending_reliable.keys().copied().collect() } + pub fn queued_reliable_packet_count(&self) -> usize { + self.queued_reliable.len() + } + + pub fn outstanding_reliable_packet_count(&self) -> usize { + self.pending_reliable.len() + self.queued_reliable.len() + } + fn pending_accept_for_resend(&mut self, now_ms: u64) -> Option { let pending = self .pending_reliable @@ -220,6 +231,7 @@ impl CultNetRudpSession { self.highest_received_sequence = None; self.received_sequences.clear(); self.pending_reliable.clear(); + self.queued_reliable.clear(); self.ordered_next_sequence_by_channel.clear(); self.ordered_buffers.clear(); self.fragment_buffers.clear(); @@ -275,6 +287,13 @@ impl CultNetRudpSession { payload: Vec, options: CultNetRudpSendOptions, ) -> Result { + if options.reliable + && self.pending_reliable.len() >= CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS + { + return Err(anyhow!( + "RUDP reliable send window is full; receive acknowledgements before sending" + )); + } self.send_many(channel_id, payload, options, None)? .into_iter() .next() @@ -321,12 +340,13 @@ impl CultNetRudpSession { index as u16, fragment_count as u16, ); - if packet.reliable { - self.track_reliable(packet.clone(), options.now_ms); - } packets.push(packet); } - return Ok(packets); + return Ok(if options.reliable { + self.admit_reliable_packets(packets, options.now_ms) + } else { + packets + }); } } @@ -340,7 +360,7 @@ impl CultNetRudpSession { options.sequenced, ); if packet.reliable { - self.track_reliable(packet.clone(), options.now_ms); + return Ok(self.admit_reliable_packets(vec![packet], options.now_ms)); } Ok(vec![packet]) } @@ -352,6 +372,7 @@ impl CultNetRudpSession { ) -> Result { self.require_connection(packet)?; self.apply_acknowledgements(packet); + let ready_to_send = self.promote_queued_reliable(now_ms); self.last_received_at_ms = Some(now_ms); let expected_sequence_if_uninitialized = self .highest_received_sequence @@ -363,6 +384,7 @@ impl CultNetRudpSession { self.connected = true; return Ok(CultNetRudpReceiveResult { delivered: Vec::new(), + ready_to_send, reply: None, pong: false, pong_payload: Vec::new(), @@ -375,6 +397,7 @@ impl CultNetRudpSession { self.remember_received(packet.sequence); return Ok(CultNetRudpReceiveResult { delivered: Vec::new(), + ready_to_send, reply: Some(self.create_packet( CultNetRudpPacketType::Pong, "control", @@ -398,6 +421,7 @@ impl CultNetRudpSession { } return Ok(CultNetRudpReceiveResult { delivered: Vec::new(), + ready_to_send, reply: None, pong: packet.packet_type == CultNetRudpPacketType::Pong, pong_payload: if packet.packet_type == CultNetRudpPacketType::Pong { @@ -415,6 +439,7 @@ impl CultNetRudpSession { self.connected = false; return Ok(CultNetRudpReceiveResult { delivered: Vec::new(), + ready_to_send, reply: None, pong: false, pong_payload: Vec::new(), @@ -426,6 +451,7 @@ impl CultNetRudpSession { if packet.packet_type != CultNetRudpPacketType::Data { return Ok(CultNetRudpReceiveResult { delivered: Vec::new(), + ready_to_send, reply: None, pong: false, pong_payload: Vec::new(), @@ -443,6 +469,7 @@ impl CultNetRudpSession { if duplicate { return Ok(CultNetRudpReceiveResult { delivered: Vec::new(), + ready_to_send, reply: None, pong: false, pong_payload: Vec::new(), @@ -454,6 +481,7 @@ impl CultNetRudpSession { let Some((frame, ordered, next_sequence)) = self.reassemble(packet)? else { return Ok(CultNetRudpReceiveResult { delivered: Vec::new(), + ready_to_send, reply: None, pong: false, pong_payload: Vec::new(), @@ -468,6 +496,7 @@ impl CultNetRudpSession { }; Ok(CultNetRudpReceiveResult { delivered, + ready_to_send, reply: None, pong: false, pong_payload: Vec::new(), @@ -630,12 +659,45 @@ impl CultNetRudpSession { ); } + fn admit_reliable_packets( + &mut self, + packets: Vec, + now_ms: u64, + ) -> Vec { + let available = + CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS.saturating_sub(self.pending_reliable.len()); + let mut ready = Vec::with_capacity(available.min(packets.len())); + for packet in packets { + if ready.len() < available { + self.track_reliable(packet.clone(), now_ms); + ready.push(packet); + } else { + self.queued_reliable.push_back(packet); + } + } + ready + } + + fn promote_queued_reliable(&mut self, now_ms: u64) -> Vec { + let available = + CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS.saturating_sub(self.pending_reliable.len()); + let mut ready = Vec::with_capacity(available.min(self.queued_reliable.len())); + for _ in 0..available { + let Some(packet) = self.queued_reliable.pop_front() else { + break; + }; + self.track_reliable(packet.clone(), now_ms); + ready.push(packet); + } + ready + } + fn ensure_reliable_capacity(&self, packet_count: usize) -> Result<()> { if packet_count == 0 { return Ok(()); } if let Some(limit) = self.max_pending_reliable_packets { - if self.pending_reliable.len() + packet_count > limit { + if self.outstanding_reliable_packet_count() + packet_count > limit { return Err(anyhow!("RUDP reliable send queue is full")); } } @@ -1283,6 +1345,9 @@ impl CultNetRudpServerHub { if let Some(reply) = result.reply { self.send_packet(remote_addr, &reply)?; } + for ready in result.ready_to_send { + self.send_packet(remote_addr, &ready)?; + } if let Some(ack) = ack { self.send_packet(remote_addr, &ack)?; } @@ -1474,6 +1539,49 @@ impl CultNetRudpSocketTransportConnection { self.session.check_timeout(now_ms(), timeout_ms) } + pub fn pending_reliable_packet_count(&self) -> usize { + self.session.pending_reliable_sequences().len() + } + + pub fn outstanding_reliable_packet_count(&self) -> usize { + self.session.outstanding_reliable_packet_count() + } + + pub fn flush_reliable(&mut self, timeout: Duration) -> Result<()> { + let original_timeout = self.socket.read_timeout()?; + let poll_timeout = original_timeout + .map(|configured| configured.min(Duration::from_millis(10))) + .unwrap_or(Duration::from_millis(10)); + self.socket.set_read_timeout(Some(poll_timeout))?; + + let deadline = Instant::now() + timeout; + let mut preserved_frames = VecDeque::new(); + let result = (|| { + while self.outstanding_reliable_packet_count() > 0 { + if Instant::now() >= deadline { + return Err(anyhow!( + "RUDP reliable flush timed out with {} packets outstanding", + self.outstanding_reliable_packet_count() + )); + } + if let Some(frame) = self.receive_once()? { + preserved_frames.push_back(frame); + } + self.poll_resends()?; + } + Ok(()) + })(); + + preserved_frames.append(&mut self.delivered_frames); + self.delivered_frames = preserved_frames; + let restore = self.socket.set_read_timeout(original_timeout); + match (result, restore) { + (Err(error), _) => Err(error), + (Ok(()), Err(error)) => Err(error.into()), + (Ok(()), Ok(())) => Ok(()), + } + } + pub fn receive_once(&mut self) -> Result> { if let Some(frame) = self.delivered_frames.pop_front() { return Ok(Some(frame)); @@ -1524,6 +1632,9 @@ impl CultNetRudpSocketTransportConnection { if let Some(reply) = result.reply { self.send_packet(&reply)?; } + for ready in result.ready_to_send { + self.send_packet(&ready)?; + } if result.pong { self.pong_payloads.push_back(result.pong_payload); } diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index 6511fdbd..32e643d3 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -4,6 +4,7 @@ use chrono::Utc; use cultcache_rs::CultCache; use cultcache_rs::DatabaseEntry; use cultcache_rs::SingleFileMessagePackBackingStore; +use cultnet_rs::CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS; use cultnet_rs::CultMesh; use cultnet_rs::CultMeshAuthorityLease; use cultnet_rs::CultMeshPeerCard; @@ -60,6 +61,7 @@ use cultnet_rs::encode_frame; use cultnet_rs::encode_rudp_packet; use pretty_assertions::assert_eq; use std::cell::RefCell; +use std::collections::VecDeque; use std::net::UdpSocket; use std::rc::Rc; use std::sync::atomic::{AtomicBool, Ordering}; @@ -1248,6 +1250,136 @@ fn rudp_socket_transport_carries_fragmented_reliable_ordered_schema_frames() -> Ok(()) } +#[test] +fn rudp_session_advances_large_fragment_sets_through_a_bounded_reliable_window() -> Result<()> { + let connection_id = 0x1020_3042; + let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id, + initial_sequence: 1, + resend_delay_ms: 25, + max_pending_reliable_packets: None, + }); + let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id, + initial_sequence: 100, + resend_delay_ms: 25, + max_pending_reliable_packets: None, + }); + for session in [&mut sender, &mut receiver] { + session.receive( + &CultNetRudpPacket { + packet_type: CultNetRudpPacketType::Accept, + connection_id, + sequence: 0, + ack: 0, + ack_mask: 0, + channel_id: "control".into(), + reliable: false, + ordered: false, + sequenced: false, + fragment_id: 0, + fragment_index: 0, + fragment_count: 0, + payload: Vec::new(), + }, + 0, + )?; + } + + let fragment_count = CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS + 17; + let payload = (0..fragment_count * 8) + .map(|index| (index % 251) as u8) + .collect::>(); + let initial = sender.send_many( + "schema", + payload.clone(), + CultNetRudpSendOptions { + reliable: true, + ordered: true, + now_ms: 1, + ..CultNetRudpSendOptions::default() + }, + Some(8), + )?; + assert_eq!(initial.len(), CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS); + assert_eq!(sender.pending_reliable_sequences().len(), initial.len()); + assert_eq!(sender.queued_reliable_packet_count(), 17); + + 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)?; + wire.extend(acknowledged.ready_to_send); + } + + assert_eq!(sender.outstanding_reliable_packet_count(), 0); + assert_eq!(delivered.len(), 1); + assert_eq!(delivered[0].payload, payload); + Ok(()) +} + +#[test] +fn rudp_socket_flush_waits_for_large_fragment_delivery() -> Result<()> { + let server_socket = bind_udp_socket()?; + let client_socket = bind_udp_socket()?; + let server_addr = server_socket.local_addr()?; + let connection_id = 0x1020_3043; + let mut server = + CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + runtime_id: "rust-rudp-window-server".into(), + socket: server_socket, + mode: CultNetRudpSocketMode::Server, + remote_addr: None, + connection_id, + initial_sequence: 100, + resend_delay_ms: 25, + transport_id: None, + max_payload_bytes: None, + max_fragment_bytes: Some(8), + max_pending_reliable_packets: None, + reconnect_policy: None, + })?; + let mut client = + CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + runtime_id: "rust-rudp-window-client".into(), + socket: client_socket, + mode: CultNetRudpSocketMode::Client, + remote_addr: Some(server_addr), + connection_id, + initial_sequence: 1, + resend_delay_ms: 25, + transport_id: None, + max_payload_bytes: None, + max_fragment_bytes: Some(8), + max_pending_reliable_packets: None, + reconnect_policy: None, + })?; + client.connect(b"join".to_vec())?; + pump_rudp_handshake(&mut client, &mut server)?; + + let payload = vec![0x5a; (CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS + 17) * 8]; + client.send("schema", payload.clone())?; + assert_eq!( + client.pending_reliable_packet_count(), + CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS + ); + assert_eq!( + client.outstanding_reliable_packet_count(), + CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS + 17 + ); + + let server_worker = thread::spawn(move || receive_rudp_frame(&mut server)); + client.flush_reliable(StdDuration::from_secs(5))?; + let delivered = server_worker + .join() + .map_err(|_| anyhow::anyhow!("RUDP server worker panicked"))??; + assert_eq!(delivered.payload, payload); + assert_eq!(client.outstanding_reliable_packet_count(), 0); + Ok(()) +} + #[test] fn cultmesh_facade_creates_rudp_client_from_peer_endpoint() -> Result<()> { let connection_id = 0x2030_4050; diff --git a/packages/cultnet-ts/src/index.ts b/packages/cultnet-ts/src/index.ts index 64d9c2fd..731cafb5 100644 --- a/packages/cultnet-ts/src/index.ts +++ b/packages/cultnet-ts/src/index.ts @@ -83,6 +83,7 @@ export { createRudpTransportProfile, decodeRudpPacket, encodeRudpPacket, + CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS, CultNetRudpReconnectLoop, CultNetRudpSession, CultNetRudpSocketTransportConnection, diff --git a/packages/cultnet-ts/src/operation-service.ts b/packages/cultnet-ts/src/operation-service.ts index 362e4a06..0b6f6b31 100644 --- a/packages/cultnet-ts/src/operation-service.ts +++ b/packages/cultnet-ts/src/operation-service.ts @@ -145,6 +145,7 @@ async function handleServerDatagram( if (!peer) return; const result = peer.session.receive(packet, Date.now()); if (result.reply) sendPacket(remote, result.reply); + for (const ready of result.readyToSend ?? []) sendPacket(remote, ready); if (result.disconnected) { sessions.delete(key); return; diff --git a/packages/cultnet-ts/src/peer.ts b/packages/cultnet-ts/src/peer.ts index 816fcfd6..0e54ee58 100644 --- a/packages/cultnet-ts/src/peer.ts +++ b/packages/cultnet-ts/src/peer.ts @@ -106,6 +106,10 @@ export class CultNetPeer extends EventEmitter { this.#stream?.write(encodeFrame(payload)); } + async flush(timeoutMs?: number): Promise { + await this.#transport?.flush?.(timeoutMs); + } + sendHello(message: CultNetHelloMessage): void { this.send(message); } diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index 7a38cb1d..7a1a9e17 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -16,6 +16,8 @@ const RUDP_MAGIC = [0x43, 0x4e, 0x52, 0x30] as const; // CNR0 const RUDP_VERSION = 0; const RUDP_FIXED_HEADER_BYTES = 36; const MAX_CHANNEL_ID_BYTES = 255; +export const CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS = 64; +const RUDP_RECEIVED_SEQUENCE_WINDOW = 4_096; export type CultNetRudpPacketType = | "connect" @@ -57,6 +59,7 @@ export interface CultNetRudpSessionOptions { export interface CultNetRudpReceiveResult { delivered: CultNetRudpDeliveredFrame[]; + readyToSend?: CultNetRudpPacket[]; reply?: CultNetRudpPacket; pong?: boolean; pongPayload?: Uint8Array; @@ -151,6 +154,7 @@ export class CultNetRudpSession { #highestReceivedSequence: number | undefined; readonly #receivedSequences = new Set(); readonly #pendingReliable = new Map(); + readonly #queuedReliable: CultNetRudpPacket[] = []; readonly #orderedNextSequenceByChannel = new Map(); readonly #orderedBuffers = new Map>(); readonly #fragmentBuffers = new Map(); @@ -173,6 +177,14 @@ export class CultNetRudpSession { return [...this.#pendingReliable.keys()].sort((left, right) => left - right); } + get queuedReliablePacketCount(): number { + return this.#queuedReliable.length; + } + + get outstandingReliablePacketCount(): number { + return this.#pendingReliable.size + this.#queuedReliable.length; + } + get lastReceivedAtMs(): number | undefined { return this.#lastReceivedAtMs; } @@ -183,6 +195,7 @@ export class CultNetRudpSession { this.#highestReceivedSequence = undefined; this.#receivedSequences.clear(); this.#pendingReliable.clear(); + this.#queuedReliable.splice(0); this.#orderedNextSequenceByChannel.clear(); this.#orderedBuffers.clear(); this.#fragmentBuffers.clear(); @@ -238,6 +251,9 @@ export class CultNetRudpSession { payload: Uint8Array, options: { reliable?: boolean; ordered?: boolean; sequenced?: boolean; nowMs?: number } = {}, ): CultNetRudpPacket { + if (options.reliable && this.#pendingReliable.size >= CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS) { + throw new Error("RUDP reliable send window is full; receive acknowledgements before sending."); + } return this.sendMany(channelId, payload, options)[0]!; } @@ -271,10 +287,9 @@ export class CultNetRudpSession { ordered: options.ordered, sequenced: options.sequenced, }); - if (packet.reliable) { - this.#trackReliable(packet, options.nowMs ?? 0); - } - return [packet]; + return packet.reliable + ? this.#admitReliablePackets([packet], options.nowMs ?? 0) + : [packet]; } const fragmentCount = Math.ceil(payload.byteLength / maxFragmentBytes); @@ -298,17 +313,17 @@ export class CultNetRudpSession { fragmentIndex: index, fragmentCount, }); - if (packet.reliable) { - this.#trackReliable(packet, options.nowMs ?? 0); - } packets.push(packet); } - return packets; + return options.reliable + ? this.#admitReliablePackets(packets, options.nowMs ?? 0) + : packets; } receive(packet: CultNetRudpPacket, nowMs = 0): CultNetRudpReceiveResult { this.#requireConnection(packet); this.#applyAcknowledgements(packet); + const readyToSend = this.#promoteQueuedReliable(nowMs); this.#lastReceivedAtMs = nowMs; const expectedSequenceIfUninitialized = this.#highestReceivedSequence === undefined ? packet.sequence @@ -317,13 +332,14 @@ export class CultNetRudpSession { if (packet.packetType === "accept") { this.#rememberReceived(packet.sequence); this.#connected = true; - return { delivered: [] }; + return { delivered: [], readyToSend }; } if (packet.packetType === "ping") { this.#rememberReceived(packet.sequence); return { delivered: [], + readyToSend, reply: this.#createPacket({ packetType: "pong", channelId: "control", @@ -336,6 +352,7 @@ export class CultNetRudpSession { if (packet.packetType === "pong") this.#rememberReceived(packet.sequence); return { delivered: this.#drainUnblockedOrdered(), + readyToSend, pong: packet.packetType === "pong", pongPayload: packet.packetType === "pong" ? packet.payload ?? new Uint8Array() : undefined, }; @@ -346,31 +363,38 @@ export class CultNetRudpSession { this.#connected = false; return { delivered: [], + readyToSend, disconnected: true, disconnectReason: packet.payload ?? new Uint8Array(), }; } if (packet.packetType !== "data") { - return { delivered: [] }; + return { delivered: [], readyToSend }; } - const isDuplicate = this.#receivedSequences.has(packet.sequence); + const isDuplicate = this.#receivedSequences.has(packet.sequence) + || (this.#highestReceivedSequence !== undefined + && packet.sequence < this.#highestReceivedSequence + && this.#highestReceivedSequence - packet.sequence >= RUDP_RECEIVED_SEQUENCE_WINDOW); this.#rememberReceived(packet.sequence); if (isDuplicate) { - return { delivered: [] }; + return { delivered: [], readyToSend }; } const reassembled = this.#reassemble(packet); if (!reassembled) { - return { delivered: [] }; + return { delivered: [], readyToSend }; } if (!reassembled.ordered) { - return { delivered: [reassembled.frame] }; + return { delivered: [reassembled.frame], readyToSend }; } - return { delivered: this.#deliverOrdered(reassembled.frame, reassembled.nextSequence, expectedSequenceIfUninitialized) }; + return { + delivered: this.#deliverOrdered(reassembled.frame, reassembled.nextSequence, expectedSequenceIfUninitialized), + readyToSend, + }; } createAck(): CultNetRudpPacket { @@ -487,11 +511,26 @@ export class CultNetRudpSession { }); } + #admitReliablePackets(packets: CultNetRudpPacket[], nowMs: number): CultNetRudpPacket[] { + const available = Math.max(0, CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS - this.#pendingReliable.size); + const ready = packets.slice(0, available); + for (const packet of ready) this.#trackReliable(packet, nowMs); + this.#queuedReliable.push(...packets.slice(available)); + return ready; + } + + #promoteQueuedReliable(nowMs: number): CultNetRudpPacket[] { + const available = Math.max(0, CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS - this.#pendingReliable.size); + const ready = this.#queuedReliable.splice(0, available); + for (const packet of ready) this.#trackReliable(packet, nowMs); + return ready; + } + #ensureReliableCapacity(packetCount: number): void { if (packetCount === 0 || this.#maxPendingReliablePackets === undefined) { return; } - if (this.#pendingReliable.size + packetCount > this.#maxPendingReliablePackets) { + if (this.outstandingReliablePacketCount + packetCount > this.#maxPendingReliablePackets) { throw new Error("RUDP reliable send queue is full."); } } @@ -510,6 +549,12 @@ export class CultNetRudpSession { if (this.#highestReceivedSequence === undefined || sequence > this.#highestReceivedSequence) { this.#highestReceivedSequence = sequence; } + if (this.#receivedSequences.size > RUDP_RECEIVED_SEQUENCE_WINDOW) { + const keepFrom = Math.max(0, (this.#highestReceivedSequence ?? sequence) - RUDP_RECEIVED_SEQUENCE_WINDOW + 1); + for (const received of this.#receivedSequences) { + if (received < keepFrom) this.#receivedSequences.delete(received); + } + } } #ackState(): { ack: number; ackMask: number } { @@ -754,6 +799,10 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement return { ...this.#stats }; } + get outstandingReliablePacketCount(): number { + return this.#session.outstandingReliablePacketCount; + } + connect(payload = new Uint8Array()): void { if (this.#mode !== "client") { throw new Error("Only a client RUDP socket transport can initiate connect."); @@ -773,6 +822,21 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement this.#stats.framesSent += 1; } + async flush(timeoutMs = 30_000): Promise { + const deadline = Date.now() + Math.max(0, timeoutMs); + while (this.#session.outstandingReliablePacketCount > 0) { + if (this.#closed) { + throw new Error("RUDP transport closed before reliable packets were acknowledged."); + } + if (Date.now() >= deadline) { + throw new Error( + `RUDP reliable flush timed out with ${this.#session.outstandingReliablePacketCount} packets outstanding.`, + ); + } + await new Promise(resolve => setTimeout(resolve, 5)); + } + } + ping(payload = new Uint8Array()): void { this.#sendPacket(this.#session.createPing(payload)); } @@ -834,6 +898,9 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement if (result.reply) { this.#sendPacket(result.reply); } + for (const ready of result.readyToSend ?? []) { + this.#sendPacket(ready); + } if (result.pong) { this.emit("pong", { payload: result.pongPayload ?? new Uint8Array() }); } diff --git a/packages/cultnet-ts/src/transport.ts b/packages/cultnet-ts/src/transport.ts index 53f64127..a1b9b3ce 100644 --- a/packages/cultnet-ts/src/transport.ts +++ b/packages/cultnet-ts/src/transport.ts @@ -131,6 +131,7 @@ export interface CultNetTransportConnection { readonly profile: CultNetTransportProfile; readonly stats: CultNetTransportStats; send(channelId: string, payload: Uint8Array): void; + flush?(timeoutMs?: number): Promise; close(): void; on( eventName: EventName, diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index 945e0a3b..621949aa 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -21,6 +21,7 @@ import { CultNetRudpReconnectLoop, CultNetRudpSession, CultNetRudpSocketTransportConnection, + CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS, CultNetSchemaCatalog, CultNetSchemaRegistry, CultNetSecret, @@ -1297,6 +1298,41 @@ test("operation envelopes preserve typed service routing and payload correlation assert.equal(response.messageId, request.messageId); }); +test("rudp session advances large fragment sets through a bounded reliable window", () => { + const connectionId = 457; + const sender = new CultNetRudpSession({ connectionId, initialSequence: 1, resendDelayMs: 25 }); + const receiver = new CultNetRudpSession({ connectionId, initialSequence: 100, resendDelayMs: 25 }); + sender.receive({ packetType: "accept", connectionId, sequence: 0, ack: 0, ackMask: 0, channelId: "control" }); + receiver.receive({ packetType: "accept", connectionId, sequence: 0, ack: 0, ackMask: 0, channelId: "control" }); + + const fragmentCount = CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS + 17; + const payload = Uint8Array.from( + { length: fragmentCount * 8 }, + (_, index) => index % 251, + ); + const wire = sender.sendMany("schema", payload, { + reliable: true, + ordered: true, + nowMs: 1, + maxFragmentBytes: 8, + }); + assert.equal(wire.length, CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS); + assert.equal(sender.pendingReliableSequences.length, wire.length); + assert.equal(sender.queuedReliablePacketCount, 17); + + 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); + wire.push(...(acknowledged.readyToSend ?? [])); + } + + assert.equal(sender.outstandingReliablePacketCount, 0); + assert.equal(delivered.length, 1); + assert.deepEqual(delivered[0]?.payload, payload); +}); + 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 fc72b2d4..b64399b0 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -723,6 +723,10 @@ namespace GameCult.Networking /// public IReadOnlyList Delivered { get; set; } = Array.Empty(); /// + /// Gets or sets queued reliable packets admitted by newly received acknowledgements. + /// + public IReadOnlyList ReadyToSend { get; set; } = Array.Empty(); + /// /// Gets or sets an optional immediate reply packet. /// public CultNetRudpPacket? Reply { get; set; } @@ -865,6 +869,9 @@ namespace GameCult.Networking /// public sealed class CultNetRudpSession { + /// Maximum reliable packets admitted to the wire before acknowledgements advance the window. + public const int ReliableSendWindowPackets = 64; + private const int ReceivedSequenceWindow = 4096; private sealed class PendingReliablePacket { public CultNetRudpPacket Packet { get; set; } = new CultNetRudpPacket(); @@ -895,6 +902,7 @@ namespace GameCult.Networking private readonly HashSet _receivedSequences = new HashSet(); private readonly object _pendingReliableGate = new object(); private readonly Dictionary _pendingReliable = new Dictionary(); + private readonly Queue _queuedReliable = new Queue(); private readonly Dictionary _orderedNextSequenceByChannel = new Dictionary(StringComparer.Ordinal); private readonly Dictionary> _orderedBuffers = new Dictionary>(StringComparer.Ordinal); @@ -947,6 +955,18 @@ namespace GameCult.Networking } } + /// Gets the number of reliable packets waiting for admission to the wire. + public int QueuedReliablePacketCount + { + get { lock (_pendingReliableGate) return _queuedReliable.Count; } + } + + /// Gets all reliable packets that have not yet been acknowledged. + public int OutstandingReliablePacketCount + { + get { lock (_pendingReliableGate) return _pendingReliable.Count + _queuedReliable.Count; } + } + /// /// Creates a reliable ordered connect packet. /// @@ -982,6 +1002,12 @@ namespace GameCult.Networking /// public CultNetRudpPacket Send(string channelId, byte[] payload, CultNetRudpSendOptions? options = null) { + options ??= new CultNetRudpSendOptions(); + lock (_pendingReliableGate) + { + if (options.Reliable && _pendingReliable.Count >= ReliableSendWindowPackets) + throw new InvalidOperationException("RUDP reliable send window is full; receive acknowledgements before sending."); + } return SendMany(channelId, payload, options).First(); } @@ -1029,13 +1055,11 @@ namespace GameCult.Networking fragmentId, (ushort)index, (ushort)fragmentCount); - if (fragmentPacket.Reliable) - { - TrackReliable(fragmentPacket, options.NowMs); - } packets.Add(fragmentPacket); } - return packets; + return options.Reliable + ? AdmitReliablePackets(packets, options.NowMs) + : packets; } EnsureReliableCapacity(options.Reliable ? 1 : 0); @@ -1048,7 +1072,7 @@ namespace GameCult.Networking options.Sequenced); if (packet.Reliable) { - TrackReliable(packet, options.NowMs); + return AdmitReliablePackets(new[] { packet }, options.NowMs); } return new[] { packet }; } @@ -1060,6 +1084,7 @@ namespace GameCult.Networking { RequireConnection(packet); ApplyAcknowledgements(packet); + var readyToSend = PromoteQueuedReliable(nowMs); _lastReceivedAtMs = nowMs; var expectedSequenceIfUninitialized = _highestReceivedSequence.HasValue ? _highestReceivedSequence.Value + 1 @@ -1069,7 +1094,7 @@ namespace GameCult.Networking { RememberReceived(packet.Sequence); _connected = true; - return new CultNetRudpReceiveResult(); + return new CultNetRudpReceiveResult { ReadyToSend = readyToSend }; } if (packet.PacketType == CultNetRudpPacketType.Ping) @@ -1077,6 +1102,7 @@ namespace GameCult.Networking RememberReceived(packet.Sequence); return new CultNetRudpReceiveResult { + ReadyToSend = readyToSend, Reply = CreatePacket( CultNetRudpPacketType.Pong, "control", @@ -1094,6 +1120,7 @@ namespace GameCult.Networking return new CultNetRudpReceiveResult { Delivered = DrainUnblockedOrdered(), + ReadyToSend = readyToSend, Pong = packet.PacketType == CultNetRudpPacketType.Pong, PongPayload = packet.PacketType == CultNetRudpPacketType.Pong ? packet.Payload ?? Array.Empty() @@ -1107,6 +1134,7 @@ namespace GameCult.Networking _connected = false; return new CultNetRudpReceiveResult { + ReadyToSend = readyToSend, Disconnected = true, DisconnectReason = packet.Payload ?? Array.Empty() }; @@ -1114,24 +1142,28 @@ namespace GameCult.Networking if (packet.PacketType != CultNetRudpPacketType.Data) { - return new CultNetRudpReceiveResult(); + return new CultNetRudpReceiveResult { ReadyToSend = readyToSend }; } - var duplicate = _receivedSequences.Contains(packet.Sequence); + var duplicate = _receivedSequences.Contains(packet.Sequence) + || (_highestReceivedSequence.HasValue + && packet.Sequence < _highestReceivedSequence.Value + && _highestReceivedSequence.Value - packet.Sequence >= ReceivedSequenceWindow); RememberReceived(packet.Sequence); if (duplicate) { - return new CultNetRudpReceiveResult(); + return new CultNetRudpReceiveResult { ReadyToSend = readyToSend }; } var reassembled = Reassemble(packet); if (reassembled == null) { - return new CultNetRudpReceiveResult(); + return new CultNetRudpReceiveResult { ReadyToSend = readyToSend }; } return new CultNetRudpReceiveResult { + ReadyToSend = readyToSend, Delivered = reassembled.Ordered ? DeliverOrdered(reassembled.Frame, reassembled.NextSequence, expectedSequenceIfUninitialized) : new[] { reassembled.Frame } @@ -1271,6 +1303,54 @@ namespace GameCult.Networking } } + private IReadOnlyList AdmitReliablePackets( + IReadOnlyList packets, + long nowMs) + { + lock (_pendingReliableGate) + { + var available = Math.Max(0, ReliableSendWindowPackets - _pendingReliable.Count); + var ready = new List(Math.Min(available, packets.Count)); + foreach (var packet in packets) + { + if (ready.Count < available) + { + _pendingReliable[packet.Sequence] = new PendingReliablePacket + { + Packet = ClonePacket(packet), + LastSentAtMs = nowMs + }; + ready.Add(packet); + } + else + { + _queuedReliable.Enqueue(ClonePacket(packet)); + } + } + return ready; + } + } + + private IReadOnlyList PromoteQueuedReliable(long nowMs) + { + lock (_pendingReliableGate) + { + var available = Math.Max(0, ReliableSendWindowPackets - _pendingReliable.Count); + var ready = new List(Math.Min(available, _queuedReliable.Count)); + while (ready.Count < available && _queuedReliable.Count > 0) + { + var packet = _queuedReliable.Dequeue(); + _pendingReliable[packet.Sequence] = new PendingReliablePacket + { + Packet = ClonePacket(packet), + LastSentAtMs = nowMs + }; + ready.Add(packet); + } + return ready; + } + } + private void EnsureReliableCapacity(int packetCount) { if (packetCount == 0 || !_maxPendingReliablePackets.HasValue) @@ -1280,7 +1360,7 @@ namespace GameCult.Networking lock (_pendingReliableGate) { - if (_pendingReliable.Count + packetCount > _maxPendingReliablePackets.Value) + if (_pendingReliable.Count + _queuedReliable.Count + packetCount > _maxPendingReliablePackets.Value) { throw new InvalidOperationException("RUDP reliable send queue is full."); } @@ -1309,6 +1389,13 @@ namespace GameCult.Networking { _highestReceivedSequence = sequence; } + if (_receivedSequences.Count > ReceivedSequenceWindow) + { + var keepFrom = (_highestReceivedSequence ?? sequence) >= ReceivedSequenceWindow - 1 + ? (_highestReceivedSequence ?? sequence) - (uint)(ReceivedSequenceWindow - 1) + : 0; + _receivedSequences.RemoveWhere(received => received < keepFrom); + } } private (uint Ack, uint AckMask) AckState() @@ -1698,6 +1785,46 @@ namespace GameCult.Networking _stats.FramesSent++; } + /// + /// Pumps acknowledgements and resends until every queued reliable packet is acknowledged. + /// + public void FlushReliable(TimeSpan? timeout = null) + { + var deadline = DateTimeOffset.UtcNow.Add(timeout ?? TimeSpan.FromSeconds(30)); + var preserved = new List(); + while (_deliveredFrames.Count > 0) + preserved.Add(_deliveredFrames.Dequeue()); + try + { + while (true) + { + lock (_sessionGate) + { + if (_session.OutstandingReliablePacketCount == 0) + return; + } + if (DateTimeOffset.UtcNow >= deadline) + { + int outstanding; + lock (_sessionGate) + outstanding = _session.OutstandingReliablePacketCount; + throw new TimeoutException($"RUDP reliable flush timed out with {outstanding} packets outstanding."); + } + if (TryReceiveOnce(out var delivered) && delivered != null) + preserved.Add(delivered); + PollResends(); + Thread.Sleep(5); + } + } + finally + { + while (_deliveredFrames.Count > 0) + preserved.Add(_deliveredFrames.Dequeue()); + foreach (var frame in preserved) + _deliveredFrames.Enqueue(frame); + } + } + /// /// Sends a reliable ordered schema-channel payload. /// @@ -1947,6 +2074,7 @@ namespace GameCult.Networking result = _session.Receive(packet, NowMs()); if (result.Reply != null) SendPacket(result.Reply); + SendPackets(result.ReadyToSend); if (packet.PacketType == CultNetRudpPacketType.Accept || packet.PacketType == CultNetRudpPacketType.Data) acknowledgement = _session.CreateAck(packet.Sequence); @@ -2126,7 +2254,7 @@ namespace GameCult.Networking /// /// Gets the number of reliable packets still awaiting acknowledgement. /// - public int PendingReliablePacketCount { get { lock (SessionGate) return Session.PendingReliableSequences.Count; } } + public int PendingReliablePacketCount { get { lock (SessionGate) return Session.OutstandingReliablePacketCount; } } /// /// Gets the last transport-level remote disconnect reason, if one was received. /// @@ -2356,6 +2484,7 @@ namespace GameCult.Networking result = existingPeer.Session.Receive(packet, NowMs()); if (result.Reply != null) SendPacket(existingPeer.RemoteEndPoint, result.Reply); + SendPackets(existingPeer, result.ReadyToSend); if (packet.PacketType == CultNetRudpPacketType.Data) acknowledgement = existingPeer.Session.CreateAck(packet.Sequence); } diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index dd4a63c9..12d397f0 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1063,6 +1063,51 @@ namespace GameCult.Networking.Tests Assert.That(delivered[0].Sequence, Is.EqualTo(packets[0].Sequence)); } + [Test] + public void RudpSession_AdvancesLargeFragmentSetsThroughBoundedReliableWindow() + { + const uint connectionId = 457; + var sender = new CultNetRudpSession(new CultNetRudpSessionOptions + { + ConnectionId = connectionId, + InitialSequence = 1, + ResendDelayMs = 25 + }); + var receiver = new CultNetRudpSession(new CultNetRudpSessionOptions + { + ConnectionId = connectionId, + InitialSequence = 100, + ResendDelayMs = 25 + }); + sender.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = connectionId, Sequence = 0, ChannelId = "control" }); + receiver.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = connectionId, Sequence = 0, ChannelId = "control" }); + + var fragmentCount = CultNetRudpSession.ReliableSendWindowPackets + 17; + var payload = Enumerable.Range(0, fragmentCount * 8).Select(index => (byte)(index % 251)).ToArray(); + var wire = new Queue(sender.SendMany( + "schema", + payload, + new CultNetRudpSendOptions { Reliable = true, Ordered = true, NowMs = 1 }, + maxFragmentBytes: 8)); + Assert.That(wire, Has.Count.EqualTo(CultNetRudpSession.ReliableSendWindowPackets)); + Assert.That(sender.PendingReliableSequences, Has.Count.EqualTo(wire.Count)); + Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(17)); + + 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); + foreach (var ready in acknowledged.ReadyToSend) + wire.Enqueue(ready); + } + + Assert.That(sender.OutstandingReliablePacketCount, Is.Zero); + Assert.That(delivered, Has.Count.EqualTo(1)); + Assert.That(delivered[0].Payload, Is.EqualTo(payload)); + } + [Test] public void RudpSocketTransport_HandshakesAndCarriesReliableOrderedSchemaFrames() {