From 62cd1db32b68cf0d2cb9c50a948432b5e9748c9e Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 14 Jun 2026 22:35:40 +0100 Subject: [PATCH] Port RUDP reliability core to C# --- docs/cultnet-transport-parity.md | 12 +- src/GameCult.Networking/CultNetTransport.cs | 413 ++++++++++++++++++ .../NetworkingTests.cs | 126 ++++++ 3 files changed, 545 insertions(+), 6 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index eeba171c..5dca5f0c 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -160,15 +160,15 @@ Current progress: the RUDP runtime yet; it is the binary packet language the runtimes must converge on before resend loops, windows, and timeout behavior are allowed to claim parity. -- TypeScript now owns the first deterministic RUDP reliability state machine: - connect/accept handshake, packet-level ack/ack-mask accounting, reliable - resend scheduling, duplicate suppression, and reliable ordered channel - delivery. It is in-memory and socket-free so the behavior can be ported before - any runtime binds it to UDP I/O. +- TypeScript and C# now share the first deterministic RUDP reliability state + machine: connect/accept handshake, packet-level ack/ack-mask accounting, + reliable resend scheduling, duplicate suppression, and reliable ordered + channel delivery. It is in-memory and socket-free so the behavior can be + ported before any runtime binds it to UDP I/O. - The remaining parity work is to add the equivalent port to Kotlin, deepen Python's server-side use of its port, and move each runtime's existing TCP/LiteNetLib/WebSocket bodies behind those ports, port the RUDP reliability - state machine into C#/Rust/Python, then bind it to UDP sockets. + state machine into Rust/Python, then bind it to UDP sockets. ## RUDP Packet Contract V0 diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index 423d7675..81fa400a 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -1,5 +1,7 @@ using System; using System.Buffers.Binary; +using System.Collections.Generic; +using System.Linq; using System.IO; using System.Text; using System.Threading; @@ -408,6 +410,417 @@ namespace GameCult.Networking } } + /// + /// Payload delivered by the CultNet RUDP reliability state machine. + /// + public sealed class CultNetRudpDeliveredFrame + { + /// + /// Gets or sets the logical channel id. + /// + public string ChannelId { get; set; } = string.Empty; + /// + /// Gets or sets the delivered payload bytes. + /// + public byte[] Payload { get; set; } = Array.Empty(); + /// + /// Gets or sets the packet sequence that delivered the frame. + /// + public uint Sequence { get; set; } + } + + /// + /// Result of feeding one packet into a CultNet RUDP session. + /// + public sealed class CultNetRudpReceiveResult + { + /// + /// Gets or sets delivered frames. + /// + public IReadOnlyList Delivered { get; set; } = Array.Empty(); + /// + /// Gets or sets an optional immediate reply packet. + /// + public CultNetRudpPacket? Reply { get; set; } + } + + /// + /// Options for the in-memory CultNet RUDP reliability state machine. + /// + public sealed class CultNetRudpSessionOptions + { + /// + /// Gets or sets the connection/session binding id. + /// + public uint ConnectionId { get; set; } + /// + /// Gets or sets the first local packet sequence. + /// + public uint InitialSequence { get; set; } = 1; + /// + /// Gets or sets the resend delay in milliseconds. + /// + public long ResendDelayMs { get; set; } = 250; + } + + /// + /// Send options for RUDP data packets. + /// + public sealed class CultNetRudpSendOptions + { + /// + /// Gets or sets whether the packet participates in reliable delivery. + /// + public bool Reliable { get; set; } + /// + /// Gets or sets whether the packet participates in ordered delivery. + /// + public bool Ordered { get; set; } + /// + /// Gets or sets whether the packet is latest-state sequenced. + /// + public bool Sequenced { get; set; } + /// + /// Gets or sets the current logical time in milliseconds. + /// + public long NowMs { get; set; } + } + + /// + /// Socket-free reliability state machine for CultNet RUDP. + /// + public sealed class CultNetRudpSession + { + private sealed class PendingReliablePacket + { + public CultNetRudpPacket Packet { get; set; } = new CultNetRudpPacket(); + public long LastSentAtMs { get; set; } + } + + private uint _nextSequence; + private bool _connected; + private uint? _highestReceivedSequence; + private readonly HashSet _receivedSequences = new HashSet(); + private readonly Dictionary _pendingReliable = new Dictionary(); + private readonly Dictionary _orderedNextSequenceByChannel = new Dictionary(StringComparer.Ordinal); + private readonly Dictionary> _orderedBuffers = + new Dictionary>(StringComparer.Ordinal); + + /// + /// Initializes a RUDP reliability session. + /// + public CultNetRudpSession(CultNetRudpSessionOptions options) + { + if (options == null) throw new ArgumentNullException(nameof(options)); + ConnectionId = options.ConnectionId; + _nextSequence = options.InitialSequence; + ResendDelayMs = options.ResendDelayMs; + } + + /// + /// Gets the connection/session binding id. + /// + public uint ConnectionId { get; } + /// + /// Gets the resend delay in milliseconds. + /// + public long ResendDelayMs { get; } + /// + /// Gets whether the session has completed the connect/accept handshake. + /// + public bool Connected => _connected; + /// + /// Gets reliable packet sequences awaiting acknowledgement. + /// + public IReadOnlyList PendingReliableSequences => _pendingReliable.Keys.OrderBy(value => value).ToArray(); + + /// + /// Creates a reliable ordered connect packet. + /// + public CultNetRudpPacket CreateConnect(long nowMs = 0, byte[]? payload = null) + { + var packet = CreatePacket(CultNetRudpPacketType.Connect, "control", payload ?? Array.Empty(), reliable: true, ordered: true, sequenced: false); + TrackReliable(packet, nowMs); + return packet; + } + + /// + /// Accepts a connect packet and returns a reliable ordered accept packet. + /// + public CultNetRudpPacket AcceptConnect(CultNetRudpPacket packet, long nowMs = 0, byte[]? payload = null) + { + RequireConnection(packet); + if (packet.PacketType != CultNetRudpPacketType.Connect) + { + throw new InvalidOperationException($"Expected RUDP connect packet, got {packet.PacketType}."); + } + + RememberReceived(packet.Sequence); + _connected = true; + var response = CreatePacket(CultNetRudpPacketType.Accept, "control", payload ?? Array.Empty(), reliable: true, ordered: true, sequenced: false); + TrackReliable(response, nowMs); + return response; + } + + /// + /// Creates a data packet. + /// + public CultNetRudpPacket Send(string channelId, byte[] payload, CultNetRudpSendOptions? options = null) + { + if (!_connected) + { + throw new InvalidOperationException("Cannot send RUDP data before the session is connected."); + } + + options ??= new CultNetRudpSendOptions(); + var packet = CreatePacket( + CultNetRudpPacketType.Data, + channelId, + payload, + options.Reliable, + options.Ordered, + options.Sequenced); + if (packet.Reliable) + { + TrackReliable(packet, options.NowMs); + } + return packet; + } + + /// + /// Applies a remote packet to the session. + /// + public CultNetRudpReceiveResult Receive(CultNetRudpPacket packet, long nowMs = 0) + { + RequireConnection(packet); + ApplyAcknowledgements(packet); + + if (packet.PacketType == CultNetRudpPacketType.Accept) + { + RememberReceived(packet.Sequence); + _connected = true; + return new CultNetRudpReceiveResult(); + } + + if (packet.PacketType == CultNetRudpPacketType.Ping) + { + RememberReceived(packet.Sequence); + return new CultNetRudpReceiveResult + { + Reply = CreatePacket( + CultNetRudpPacketType.Pong, + "control", + packet.Payload ?? Array.Empty(), + reliable: false, + ordered: false, + sequenced: false) + }; + } + + if (packet.PacketType == CultNetRudpPacketType.Ack || packet.PacketType == CultNetRudpPacketType.Pong) + { + RememberReceived(packet.Sequence); + return new CultNetRudpReceiveResult(); + } + + if (packet.PacketType != CultNetRudpPacketType.Data) + { + return new CultNetRudpReceiveResult(); + } + + var duplicate = _receivedSequences.Contains(packet.Sequence); + RememberReceived(packet.Sequence); + if (duplicate) + { + return new CultNetRudpReceiveResult(); + } + + var frame = new CultNetRudpDeliveredFrame + { + ChannelId = packet.ChannelId, + Payload = packet.Payload ?? Array.Empty(), + Sequence = packet.Sequence + }; + + return new CultNetRudpReceiveResult + { + Delivered = packet.Ordered ? DeliverOrdered(frame) : new[] { frame } + }; + } + + /// + /// Creates a packet carrying the current acknowledgement state. + /// + public CultNetRudpPacket CreateAck() + { + return CreatePacket(CultNetRudpPacketType.Ack, "control", Array.Empty(), reliable: false, ordered: false, sequenced: false); + } + + /// + /// Returns reliable packets due for resend at the supplied logical time. + /// + public IReadOnlyList DueResends(long nowMs) + { + var due = new List(); + foreach (var pending in _pendingReliable.Values) + { + if (nowMs - pending.LastSentAtMs >= ResendDelayMs) + { + pending.LastSentAtMs = nowMs; + due.Add(ClonePacket(pending.Packet)); + } + } + + return due.OrderBy(packet => packet.Sequence).ToArray(); + } + + private CultNetRudpPacket CreatePacket( + CultNetRudpPacketType packetType, + string channelId, + byte[] payload, + bool reliable, + bool ordered, + bool sequenced) + { + var sequence = _nextSequence; + _nextSequence = checked(_nextSequence + 1); + var (ack, ackMask) = AckState(); + return new CultNetRudpPacket + { + PacketType = packetType, + ConnectionId = ConnectionId, + Sequence = sequence, + Ack = ack, + AckMask = ackMask, + ChannelId = channelId, + Reliable = reliable, + Ordered = ordered, + Sequenced = sequenced, + Payload = payload ?? Array.Empty() + }; + } + + private void TrackReliable(CultNetRudpPacket packet, long nowMs) + { + _pendingReliable[packet.Sequence] = new PendingReliablePacket + { + Packet = ClonePacket(packet), + LastSentAtMs = nowMs + }; + } + + private void ApplyAcknowledgements(CultNetRudpPacket packet) + { + _pendingReliable.Remove(packet.Ack); + for (var bit = 0; bit < 32; bit++) + { + if ((packet.AckMask & (1u << bit)) != 0 && packet.Ack > bit) + { + _pendingReliable.Remove(packet.Ack - (uint)bit - 1); + } + } + } + + private void RememberReceived(uint sequence) + { + _receivedSequences.Add(sequence); + if (!_highestReceivedSequence.HasValue || sequence > _highestReceivedSequence.Value) + { + _highestReceivedSequence = sequence; + } + } + + private (uint Ack, uint AckMask) AckState() + { + var ack = _highestReceivedSequence ?? 0; + uint ackMask = 0; + for (var bit = 0; bit < 32; bit++) + { + if (ack > bit && _receivedSequences.Contains(ack - (uint)bit - 1)) + { + ackMask |= 1u << bit; + } + } + + return (ack, ackMask); + } + + private IReadOnlyList DeliverOrdered(CultNetRudpDeliveredFrame frame) + { + if (!_orderedNextSequenceByChannel.TryGetValue(frame.ChannelId, out var next)) + { + _orderedNextSequenceByChannel[frame.ChannelId] = frame.Sequence + 1; + return new[] { frame }.Concat(DrainOrdered(frame.ChannelId)).ToArray(); + } + + if (frame.Sequence < next) + { + return Array.Empty(); + } + + if (frame.Sequence > next) + { + if (!_orderedBuffers.TryGetValue(frame.ChannelId, out var buffer)) + { + buffer = new SortedDictionary(); + _orderedBuffers[frame.ChannelId] = buffer; + } + + buffer[frame.Sequence] = frame; + return Array.Empty(); + } + + _orderedNextSequenceByChannel[frame.ChannelId] = next + 1; + return new[] { frame }.Concat(DrainOrdered(frame.ChannelId)).ToArray(); + } + + private IReadOnlyList DrainOrdered(string channelId) + { + var delivered = new List(); + if (!_orderedBuffers.TryGetValue(channelId, out var buffer)) + { + return delivered; + } + + while (_orderedNextSequenceByChannel.TryGetValue(channelId, out var next) && buffer.TryGetValue(next, out var frame)) + { + buffer.Remove(next); + delivered.Add(frame); + _orderedNextSequenceByChannel[channelId] = next + 1; + } + + return delivered; + } + + private void RequireConnection(CultNetRudpPacket packet) + { + if (packet.ConnectionId != ConnectionId) + { + throw new InvalidOperationException($"RUDP packet connection id {packet.ConnectionId} does not match {ConnectionId}."); + } + } + + private static CultNetRudpPacket ClonePacket(CultNetRudpPacket packet) + { + return new CultNetRudpPacket + { + PacketType = packet.PacketType, + ConnectionId = packet.ConnectionId, + Sequence = packet.Sequence, + Ack = packet.Ack, + AckMask = packet.AckMask, + ChannelId = packet.ChannelId, + Reliable = packet.Reliable, + Ordered = packet.Ordered, + Sequenced = packet.Sequenced, + FragmentId = packet.FragmentId, + FragmentIndex = packet.FragmentIndex, + FragmentCount = packet.FragmentCount, + Payload = packet.Payload?.ToArray() ?? Array.Empty() + }; + } + } + /// /// Transport connection for the current TCP length-prefixed CultNet schema lane. /// diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 70c69fc6..ba673e34 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -321,6 +321,132 @@ namespace GameCult.Networking.Tests })); } + [Test] + public void RudpSession_HandshakeAcksReliableConnectAndAcceptPackets() + { + var client = new CultNetRudpSession(new CultNetRudpSessionOptions + { + ConnectionId = 0x0a0b0c0d, + InitialSequence = 1, + ResendDelayMs = 50 + }); + var server = new CultNetRudpSession(new CultNetRudpSessionOptions + { + ConnectionId = 0x0a0b0c0d, + InitialSequence = 100, + ResendDelayMs = 50 + }); + + var connect = client.CreateConnect(0, Encoding.UTF8.GetBytes("join")); + Assert.That(connect.PacketType, Is.EqualTo(CultNetRudpPacketType.Connect)); + Assert.That(connect.Sequence, Is.EqualTo(1)); + Assert.That(client.PendingReliableSequences, Is.EqualTo(new uint[] { 1 })); + + var accept = server.AcceptConnect(connect, 10, Encoding.UTF8.GetBytes("ok")); + Assert.That(accept.PacketType, Is.EqualTo(CultNetRudpPacketType.Accept)); + Assert.That(accept.Ack, Is.EqualTo(1)); + Assert.That(server.Connected, Is.True); + Assert.That(server.PendingReliableSequences, Is.EqualTo(new uint[] { 100 })); + + client.Receive(accept, 20); + Assert.That(client.Connected, Is.True); + Assert.That(client.PendingReliableSequences, Is.Empty); + + var ack = client.CreateAck(); + Assert.That(ack.Ack, Is.EqualTo(100)); + server.Receive(ack, 30); + Assert.That(server.PendingReliableSequences, Is.Empty); + } + + [Test] + public void RudpSession_ComputesAckMasksAndClearsPendingReliablePackets() + { + var sender = new CultNetRudpSession(new CultNetRudpSessionOptions + { + ConnectionId = 7, + InitialSequence = 10, + ResendDelayMs = 100 + }); + var receiver = new CultNetRudpSession(new CultNetRudpSessionOptions + { + ConnectionId = 7, + InitialSequence = 200, + ResendDelayMs = 100 + }); + sender.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 7, Sequence = 1, ChannelId = "control" }); + receiver.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 7, Sequence = 2, ChannelId = "control" }); + + var first = sender.Send("schema", Encoding.UTF8.GetBytes("first"), new CultNetRudpSendOptions { Reliable = true, Ordered = true }); + var second = sender.Send("schema", Encoding.UTF8.GetBytes("second"), new CultNetRudpSendOptions { Reliable = true, Ordered = true }); + var third = sender.Send("schema", Encoding.UTF8.GetBytes("third"), new CultNetRudpSendOptions { Reliable = true, Ordered = true }); + Assert.That(sender.PendingReliableSequences, Is.EqualTo(new uint[] { 10, 11, 12 })); + + receiver.Receive(first); + receiver.Receive(third); + var ackWithGap = receiver.CreateAck(); + Assert.That(ackWithGap.Ack, Is.EqualTo(12)); + Assert.That(ackWithGap.AckMask, Is.EqualTo(0b10u | (1u << 9))); + sender.Receive(ackWithGap); + Assert.That(sender.PendingReliableSequences, Is.EqualTo(new uint[] { 11 })); + + receiver.Receive(second); + var fullAck = receiver.CreateAck(); + Assert.That(fullAck.Ack, Is.EqualTo(12)); + Assert.That(fullAck.AckMask, Is.EqualTo(0b11u | (1u << 9))); + sender.Receive(fullAck); + Assert.That(sender.PendingReliableSequences, Is.Empty); + } + + [Test] + public void RudpSession_SchedulesReliableResendsUntilAcked() + { + var session = new CultNetRudpSession(new CultNetRudpSessionOptions + { + ConnectionId = 99, + InitialSequence = 1, + ResendDelayMs = 100 + }); + session.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 99, Sequence = 50, ChannelId = "control" }); + var sent = session.Send("schema", Encoding.UTF8.GetBytes("payload"), new CultNetRudpSendOptions { Reliable = true, Ordered = true, NowMs = 10 }); + + Assert.That(session.DueResends(90), Is.Empty); + Assert.That(session.DueResends(110).Select(packet => packet.Sequence).ToArray(), Is.EqualTo(new[] { sent.Sequence })); + Assert.That(session.DueResends(150), Is.Empty); + + session.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Ack, ConnectionId = 99, Sequence = 51, Ack = sent.Sequence, ChannelId = "control" }); + Assert.That(session.DueResends(250), Is.Empty); + } + + [Test] + public void RudpSession_SuppressesDuplicatesAndDeliversReliableOrderedPayloadsInSequence() + { + var sender = new CultNetRudpSession(new CultNetRudpSessionOptions + { + ConnectionId = 123, + InitialSequence = 1 + }); + var receiver = new CultNetRudpSession(new CultNetRudpSessionOptions + { + ConnectionId = 123, + InitialSequence = 100 + }); + sender.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 123, Sequence = 90, ChannelId = "control" }); + receiver.Receive(new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Accept, ConnectionId = 123, Sequence = 91, ChannelId = "control" }); + + var first = sender.Send("schema", Encoding.UTF8.GetBytes("first"), new CultNetRudpSendOptions { Reliable = true, Ordered = true }); + var second = sender.Send("schema", Encoding.UTF8.GetBytes("second"), new CultNetRudpSendOptions { Reliable = true, Ordered = true }); + var third = sender.Send("schema", Encoding.UTF8.GetBytes("third"), new CultNetRudpSendOptions { Reliable = true, Ordered = true }); + + Assert.That( + receiver.Receive(first).Delivered.Select(frame => Encoding.UTF8.GetString(frame.Payload)).ToArray(), + Is.EqualTo(new[] { "first" })); + Assert.That(receiver.Receive(third).Delivered, Is.Empty); + Assert.That(receiver.Receive(first).Delivered, Is.Empty); + Assert.That( + receiver.Receive(second).Delivered.Select(frame => Encoding.UTF8.GetString(frame.Payload)).ToArray(), + Is.EqualTo(new[] { "second", "third" })); + } + [Test] public void CultNetSchemaMessageSerialization_RoundTrips_RawSnapshotResponse() {