diff --git a/packages/cultcache-py/src/cultnet_py/transport.py b/packages/cultcache-py/src/cultnet_py/transport.py index 9ce4e4de..2b9fe462 100644 --- a/packages/cultcache-py/src/cultnet_py/transport.py +++ b/packages/cultcache-py/src/cultnet_py/transport.py @@ -406,11 +406,13 @@ class CultNetRudpSession: raise ValueError("RUDP max_pending_reliable_packets must be greater than zero") self.max_pending_reliable_packets = options.max_pending_reliable_packets self._next_sequence = _uint32(options.initial_sequence, "initial_sequence") + self._next_sequenced_by_channel: dict[str, int] = {} self._next_fragment_id = 1 self._connected = False self._last_received_at_ms: int | None = None self._highest_received_sequence: int | None = None self._received_sequences: set[int] = set() + self._latest_sequenced_by_channel: dict[str, int] = {} self._pending_reliable: dict[int, _PendingReliablePacket] = {} self._queued_reliable: deque[CultNetRudpPacket] = deque() self._ordered_next_sequence_by_channel: dict[str, int] = {} @@ -495,6 +497,8 @@ class CultNetRudpSession: raise ValueError("Cannot send RUDP data before the session is connected") options = options or CultNetRudpSendOptions() + if options.ordered and not options.reliable: + raise ValueError("RUDP ordered delivery requires reliability") if max_fragment_bytes is not None and max_fragment_bytes <= 0: raise ValueError("RUDP max_fragment_bytes must be greater than zero") @@ -550,14 +554,12 @@ class CultNetRudpSession: 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, @@ -565,7 +567,6 @@ class CultNetRudpSession: ) if packet.packet_type == CultNetRudpPacketType.DISCONNECT: - self._remember_received(packet.sequence) self._connected = False return CultNetRudpReceiveResult( ready_to_send=ready_to_send, @@ -576,12 +577,15 @@ class CultNetRudpSession: if packet.packet_type != CultNetRudpPacketType.DATA: return CultNetRudpReceiveResult(ready_to_send=ready_to_send) - 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 + duplicate = packet.reliable and ( + 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 packet.reliable: + self._remember_received(packet.sequence) if duplicate: return CultNetRudpReceiveResult(ready_to_send=ready_to_send) @@ -589,6 +593,13 @@ class CultNetRudpSession: if reassembled is None: return CultNetRudpReceiveResult(ready_to_send=ready_to_send) frame, ordered, next_sequence = reassembled + if not ordered and packet.sequenced: + newest_sequence = next_sequence - 1 + latest_sequence = self._latest_sequenced_by_channel.get(frame.channel_id) + if latest_sequence is not None and newest_sequence <= latest_sequence: + return CultNetRudpReceiveResult(ready_to_send=ready_to_send) + self._latest_sequenced_by_channel[frame.channel_id] = newest_sequence + return CultNetRudpReceiveResult(delivered=(frame,), ready_to_send=ready_to_send) if not ordered: return CultNetRudpReceiveResult(delivered=(frame,), ready_to_send=ready_to_send) return CultNetRudpReceiveResult( @@ -664,8 +675,17 @@ class CultNetRudpSession: fragment_index: int = 0, fragment_count: int = 0, ) -> CultNetRudpPacket: - sequence = self._next_sequence - self._next_sequence = _uint32(self._next_sequence + 1, "sequence") + if reliable: + sequence = self._next_sequence + self._next_sequence = _uint32(self._next_sequence + 1, "reliable sequence") + elif sequenced: + sequence = self._next_sequenced_by_channel.get(channel_id, 1) + self._next_sequenced_by_channel[channel_id] = _uint32( + sequence + 1, + "sequenced channel sequence", + ) + else: + sequence = 0 ack, ack_mask = self._ack_state() return CultNetRudpPacket( packet_type=packet_type, diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 2dbbeb2d..29fdefef 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -1110,6 +1110,52 @@ class CultCacheTests(unittest.TestCase): ["second", "third"], ) + def test_cultnet_rudp_lossy_packets_cannot_create_reliable_ordered_gaps(self) -> None: + sender = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=198, initial_sequence=1)) + receiver = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=198, initial_sequence=100)) + sender.receive(CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 198, 90, 0, 0, "control")) + receiver.receive(CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 198, 91, 0, 0, "control")) + + realtime = sender.send("realtime", b"discarded realtime", CultNetRudpSendOptions()) + latest = sender.send("latest", b"discarded latest state", CultNetRudpSendOptions(sequenced=True)) + schema = sender.send("schema", b"committed response", CultNetRudpSendOptions(reliable=True, ordered=True)) + + self.assertEqual(realtime.sequence, 0) + self.assertEqual(latest.sequence, 1) + self.assertEqual( + [frame.payload.decode("utf-8") for frame in receiver.receive(schema).delivered], + ["committed response"], + ) + + def test_cultnet_rudp_unreliable_sequenced_delivery_is_scoped_to_its_channel(self) -> None: + sender = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=197, initial_sequence=50)) + receiver = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=197, initial_sequence=100)) + sender.receive(CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 197, 90, 0, 0, "control")) + receiver.receive(CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 197, 91, 0, 0, "control")) + + older = sender.send("latest", b"older", CultNetRudpSendOptions(sequenced=True)) + newer = sender.send("latest", b"newer", CultNetRudpSendOptions(sequenced=True)) + + self.assertEqual(older.sequence, 1) + self.assertEqual(newer.sequence, 2) + self.assertEqual( + [frame.payload.decode("utf-8") for frame in receiver.receive(newer).delivered], + ["newer"], + ) + self.assertEqual(receiver.receive(older).delivered, ()) + + def test_cultnet_rudp_rejects_unreliable_ordered_delivery(self) -> None: + session = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=196, initial_sequence=1)) + session.receive(CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 196, 50, 0, 0, "control")) + + with self.assertRaises(ValueError) as raised: + session.send( + "schema", + b"cannot order what will not retransmit", + CultNetRudpSendOptions(ordered=True), + ) + self.assertIn("ordered delivery requires reliability", str(raised.exception)) + def test_cultnet_rudp_session_skips_control_packets_while_ordering_schema_payloads(self) -> None: sender = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=124, initial_sequence=1)) receiver = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=124, initial_sequence=100)) 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 6245fb9e..104bd1b9 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 @@ -2629,6 +2629,7 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { val resendDelayMs: Long = options.resendDelayMs private val maxPendingReliablePackets: Int? = options.maxPendingReliablePackets private var nextSequence = uint32(options.initialSequence, "initialSequence") + private val nextSequencedByChannel = linkedMapOf() private var nextFragmentId = 1 var connected: Boolean = false private set @@ -2636,6 +2637,7 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { private set private var highestReceivedSequence: Long? = null private val receivedSequences = linkedSetOf() + private val latestSequencedByChannel = linkedMapOf() private val pendingReliable = linkedMapOf() private val queuedReliable = ArrayDeque() private val orderedNextSequenceByChannel = linkedMapOf() @@ -2687,6 +2689,7 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { maxFragmentBytes: Int? = null, ): List { if (!connected) throw IOException("Cannot send RUDP data before the session is connected") + if (options.ordered && !options.reliable) throw IOException("RUDP ordered delivery requires reliability") if (maxFragmentBytes != null && maxFragmentBytes <= 0) throw IOException("RUDP maxFragmentBytes must be greater than zero") if (maxFragmentBytes != null && payload.size > maxFragmentBytes) { val fragmentCount = (payload.size + maxFragmentBytes - 1) / maxFragmentBytes @@ -2730,11 +2733,9 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { return CultNetRudpReceiveResult(readyToSend = readyToSend) } CultNetRudpPacketType.Ping -> { - rememberReceived(packet.sequence) 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, @@ -2742,7 +2743,6 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { ) } CultNetRudpPacketType.Disconnect -> { - rememberReceived(packet.sequence) connected = false return CultNetRudpReceiveResult(readyToSend = readyToSend, disconnected = true, disconnectReason = packet.payload.copyOf()) } @@ -2750,14 +2750,22 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { else -> return CultNetRudpReceiveResult(readyToSend = readyToSend) } - val duplicate = receivedSequences.contains(packet.sequence) || ( + val duplicate = packet.reliable && (receivedSequences.contains(packet.sequence) || ( highestReceivedSequence != null && packet.sequence < highestReceivedSequence!! && highestReceivedSequence!! - packet.sequence >= ReceivedSequenceWindow.toLong() - ) - rememberReceived(packet.sequence) + )) + if (packet.reliable) rememberReceived(packet.sequence) if (duplicate) return CultNetRudpReceiveResult(readyToSend = readyToSend) val reassembled = reassemble(packet) ?: return CultNetRudpReceiveResult(readyToSend = readyToSend) + if (!reassembled.ordered && packet.sequenced) { + val newestSequence = reassembled.nextSequence - 1 + val latestSequence = latestSequencedByChannel[reassembled.frame.channelId] + if (latestSequence != null && newestSequence <= latestSequence) { + return CultNetRudpReceiveResult(readyToSend = readyToSend) + } + latestSequencedByChannel[reassembled.frame.channelId] = newestSequence + } return CultNetRudpReceiveResult( delivered = if (reassembled.ordered) deliverOrdered(reassembled.frame, reassembled.nextSequence, expectedSequenceIfUninitialized) else listOf(reassembled.frame), readyToSend = readyToSend, @@ -2828,8 +2836,15 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { fragmentIndex: Int = 0, fragmentCount: Int = 0, ): CultNetRudpPacket { - val sequence = nextSequence - nextSequence = uint32(nextSequence + 1, "sequence") + val sequence = when { + reliable -> nextSequence.also { + nextSequence = uint32(nextSequence + 1, "reliable sequence") + } + sequenced -> (nextSequencedByChannel[channelId] ?: 1).also { + nextSequencedByChannel[channelId] = uint32(it + 1, "sequenced channel sequence") + } + else -> 0 + } val (ack, ackMask) = ackState() return CultNetRudpPacket(packetType, connectionId, sequence, ack, ackMask, channelId, reliable, ordered, sequenced, fragmentId, fragmentIndex, fragmentCount, payload.copyOf()) }