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

Bound RUDP reliable transmission windows

Browse source
This commit is contained in:
Meta Vens 2026-08-23 18:35:49 +02:00
commit f4dabfefb0
18 changed files with 846 additions and 79 deletions
Download patch file Download diff file Expand all files Collapse all files

15
docs/cultnet-transport-parity.md
View file

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

6
packages/cultcache-py/src/cultmesh_py/server.py
View file

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

6
packages/cultcache-py/src/cultnet_py/interop_peer.py
View file

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

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

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

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

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

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

@ -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<CultNetRudpDeliveredFrame> = emptyList(),
val readyToSend: List<CultNetRudpPacket> = 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<Long>()
private val pendingReliable = linkedMapOf<Long, PendingReliablePacket>()
private val queuedReliable = ArrayDeque<CultNetRudpPacket>()
private val orderedNextSequenceByChannel = linkedMapOf<String, Long>()
private val orderedBuffers = linkedMapOf<String, TreeMap<Long, PendingOrderedFrame>>()
private val fragmentBuffers = linkedMapOf<Pair<String, Int>, FragmentBuffer>()
val pendingReliableSequences: List<Long>
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<CultNetRudpPacket>, nowMs: Long): List<CultNetRudpPacket> {
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<CultNetRudpPacket> {
val available = maxOf(0, ReliableSendWindowPackets - pendingReliable.size)
val ready = mutableListOf<CultNetRudpPacket>()
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<Long, Long> {
@ -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<CultNetTransportFrame>()
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<String>) {
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<CultNetRudpDeliveredFrame>()
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(

5
packages/cultmesh-ts/src/index.ts
View file

@ -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();
}

3
packages/cultnet-rs/examples/cultnet_interop_peer.rs
View file

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

127
packages/cultnet-rs/src/rudp.rs
View file

@ -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<CultNetRudpDeliveredFrame>,
pub ready_to_send: Vec<CultNetRudpPacket>,
pub reply: Option<CultNetRudpPacket>,
pub pong: bool,
pub pong_payload: Vec<u8>,
@ -139,6 +140,7 @@ pub struct CultNetRudpSession {
highest_received_sequence: Option<u32>,
received_sequences: BTreeSet<u32>,
pending_reliable: BTreeMap<u32, PendingReliablePacket>,
queued_reliable: VecDeque<CultNetRudpPacket>,
ordered_next_sequence_by_channel: BTreeMap<String, u32>,
ordered_buffers: BTreeMap<String, BTreeMap<u32, PendingOrderedFrame>>,
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<CultNetRudpPacket> {
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<u8>,
options: CultNetRudpSendOptions,
) -> Result<CultNetRudpPacket> {
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<CultNetRudpReceiveResult> {
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<CultNetRudpPacket>,
now_ms: u64,
) -> Vec<CultNetRudpPacket> {
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<CultNetRudpPacket> {
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<Option<CultNetTransportFrame>> {
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);
}

132
packages/cultnet-rs/tests/cultnet.rs
View file

@ -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::<Vec<_>>();
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;

1
packages/cultnet-ts/src/index.ts
View file

@ -83,6 +83,7 @@ export {
createRudpTransportProfile,
decodeRudpPacket,
encodeRudpPacket,
CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS,
CultNetRudpReconnectLoop,
CultNetRudpSession,
CultNetRudpSocketTransportConnection,

1
packages/cultnet-ts/src/operation-service.ts
View file

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

4
packages/cultnet-ts/src/peer.ts
View file

@ -106,6 +106,10 @@ export class CultNetPeer extends EventEmitter {
this.#stream?.write(encodeFrame(payload));
}
async flush(timeoutMs?: number): Promise<void> {
await this.#transport?.flush?.(timeoutMs);
}
sendHello(message: CultNetHelloMessage): void {
this.send(message);
}

99
packages/cultnet-ts/src/rudp.ts
View file

@ -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<number>();
readonly #pendingReliable = new Map<number, PendingReliablePacket>();
readonly #queuedReliable: CultNetRudpPacket[] = [];
readonly #orderedNextSequenceByChannel = new Map<string, number>();
readonly #orderedBuffers = new Map<string, Map<number, PendingOrderedFrame>>();
readonly #fragmentBuffers = new Map<string, FragmentBuffer>();
@ -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<void> {
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() });
}

1
packages/cultnet-ts/src/transport.ts
View file

@ -131,6 +131,7 @@ export interface CultNetTransportConnection {
readonly profile: CultNetTransportProfile;
readonly stats: CultNetTransportStats;
send(channelId: string, payload: Uint8Array): void;
flush?(timeoutMs?: number): Promise<void>;
close(): void;
on<EventName extends keyof CultNetTransportConnectionEvents>(
eventName: EventName,

36
packages/cultnet-ts/test/cultnet.test.ts
View file

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

155
src/GameCult.Networking/CultNetTransport.cs
View file

@ -723,6 +723,10 @@ namespace GameCult.Networking
/// </summary>
public IReadOnlyList<CultNetRudpDeliveredFrame> Delivered { get; set; } = Array.Empty<CultNetRudpDeliveredFrame>();
/// <summary>
/// Gets or sets queued reliable packets admitted by newly received acknowledgements.
/// </summary>
public IReadOnlyList<CultNetRudpPacket> ReadyToSend { get; set; } = Array.Empty<CultNetRudpPacket>();
/// <summary>
/// Gets or sets an optional immediate reply packet.
/// </summary>
public CultNetRudpPacket? Reply { get; set; }
@ -865,6 +869,9 @@ namespace GameCult.Networking
/// </summary>
public sealed class CultNetRudpSession
{
/// <summary>Maximum reliable packets admitted to the wire before acknowledgements advance the window.</summary>
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<uint> _receivedSequences = new HashSet<uint>();
private readonly object _pendingReliableGate = new object();
private readonly Dictionary<uint, PendingReliablePacket> _pendingReliable = new Dictionary<uint, PendingReliablePacket>();
private readonly Queue<CultNetRudpPacket> _queuedReliable = new Queue<CultNetRudpPacket>();
private readonly Dictionary<string, uint> _orderedNextSequenceByChannel = new Dictionary<string, uint>(StringComparer.Ordinal);
private readonly Dictionary<string, SortedDictionary<uint, PendingOrderedFrame>> _orderedBuffers =
new Dictionary<string, SortedDictionary<uint, PendingOrderedFrame>>(StringComparer.Ordinal);
@ -947,6 +955,18 @@ namespace GameCult.Networking
}
}
/// <summary>Gets the number of reliable packets waiting for admission to the wire.</summary>
public int QueuedReliablePacketCount
{
get { lock (_pendingReliableGate) return _queuedReliable.Count; }
}
/// <summary>Gets all reliable packets that have not yet been acknowledged.</summary>
public int OutstandingReliablePacketCount
{
get { lock (_pendingReliableGate) return _pendingReliable.Count + _queuedReliable.Count; }
}
/// <summary>
/// Creates a reliable ordered connect packet.
/// </summary>
@ -982,6 +1002,12 @@ namespace GameCult.Networking
/// </summary>
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<byte>()
@ -1107,6 +1134,7 @@ namespace GameCult.Networking
_connected = false;
return new CultNetRudpReceiveResult
{
ReadyToSend = readyToSend,
Disconnected = true,
DisconnectReason = packet.Payload ?? Array.Empty<byte>()
};
@ -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<CultNetRudpPacket> AdmitReliablePackets(
IReadOnlyList<CultNetRudpPacket> packets,
long nowMs)
{
lock (_pendingReliableGate)
{
var available = Math.Max(0, ReliableSendWindowPackets - _pendingReliable.Count);
var ready = new List<CultNetRudpPacket>(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<CultNetRudpPacket> PromoteQueuedReliable(long nowMs)
{
lock (_pendingReliableGate)
{
var available = Math.Max(0, ReliableSendWindowPackets - _pendingReliable.Count);
var ready = new List<CultNetRudpPacket>(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++;
}
/// <summary>
/// Pumps acknowledgements and resends until every queued reliable packet is acknowledged.
/// </summary>
public void FlushReliable(TimeSpan? timeout = null)
{
var deadline = DateTimeOffset.UtcNow.Add(timeout ?? TimeSpan.FromSeconds(30));
var preserved = new List<CultNetTransportFrame>();
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);
}
}
/// <summary>
/// Sends a reliable ordered schema-channel payload.
/// </summary>
@ -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
/// <summary>
/// Gets the number of reliable packets still awaiting acknowledgement.
/// </summary>
public int PendingReliablePacketCount { get { lock (SessionGate) return Session.PendingReliableSequences.Count; } }
public int PendingReliablePacketCount { get { lock (SessionGate) return Session.OutstandingReliablePacketCount; } }
/// <summary>
/// Gets the last transport-level remote disconnect reason, if one was received.
/// </summary>
@ -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);
}

45
tests/GameCult.Networking.Tests/NetworkingTests.cs
View file

@ -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<CultNetRudpPacket>(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<CultNetRudpDeliveredFrame>();
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()
{