GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Finish the sequence-domain split in Python and Kotlin

Browse source 
Separating the reliable and lossy sequence domains landed in Rust, TypeScript
and C# on a branch. The Python and Kotlin halves were written at the same time
and left uncommitted in the worktree, so three runtimes spoke the new contract
and two spoke the old one -- the exact cross-runtime disagreement this
repository exists to prevent.

Both were already complete and implement the same three rules as the others:
ordered delivery requires reliability, sequenced channels carry their own
per-channel counters with latest-wins delivery, and duplicate suppression is
gated on packet.reliable so lossy datagrams are remembered nowhere.

Python also gains the three tests Rust, TypeScript and C# each received and it
did not: lossy packets cannot create reliable ordered gaps, unreliable
sequenced delivery is scoped to its channel, and unreliable ordered delivery is
rejected. Python passes 141.

Kotlin is committed unverified. This machine has no JVM or Gradle, and the
package carries no runtime-local test file, so its only gate is the interop
lane in CI. The change mirrors the four runtimes proven here line for line, but
mirroring is not evidence and CI is the witness.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-04 00:40:05 +02:00
commit fd676ce74d
3 changed files with 99 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

40
packages/cultcache-py/src/cultnet_py/transport.py
View file

@ -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,

46
packages/cultcache-py/tests/test_cultcache.py
View file

@ -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))

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

@ -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<String, Long>()
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<Long>()
private val latestSequencedByChannel = linkedMapOf<String, Long>()
private val pendingReliable = linkedMapOf<Long, PendingReliablePacket>()
private val queuedReliable = ArrayDeque<CultNetRudpPacket>()
private val orderedNextSequenceByChannel = linkedMapOf<String, Long>()
@ -2687,6 +2689,7 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) {
maxFragmentBytes: Int? = null,
): List<CultNetRudpPacket> {
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())
}