Port RUDP reliability core to C#

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
 

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
         }
     }
 
+    /// <summary>
+    /// Payload delivered by the CultNet RUDP reliability state machine.
+    /// </summary>
+    public sealed class CultNetRudpDeliveredFrame
+    {
+        /// <summary>
+        /// Gets or sets the logical channel id.
+        /// </summary>
+        public string ChannelId { get; set; } = string.Empty;
+        /// <summary>
+        /// Gets or sets the delivered payload bytes.
+        /// </summary>
+        public byte[] Payload { get; set; } = Array.Empty<byte>();
+        /// <summary>
+        /// Gets or sets the packet sequence that delivered the frame.
+        /// </summary>
+        public uint Sequence { get; set; }
+    }
+
+    /// <summary>
+    /// Result of feeding one packet into a CultNet RUDP session.
+    /// </summary>
+    public sealed class CultNetRudpReceiveResult
+    {
+        /// <summary>
+        /// Gets or sets delivered frames.
+        /// </summary>
+        public IReadOnlyList<CultNetRudpDeliveredFrame> Delivered { get; set; } = Array.Empty<CultNetRudpDeliveredFrame>();
+        /// <summary>
+        /// Gets or sets an optional immediate reply packet.
+        /// </summary>
+        public CultNetRudpPacket? Reply { get; set; }
+    }
+
+    /// <summary>
+    /// Options for the in-memory CultNet RUDP reliability state machine.
+    /// </summary>
+    public sealed class CultNetRudpSessionOptions
+    {
+        /// <summary>
+        /// Gets or sets the connection/session binding id.
+        /// </summary>
+        public uint ConnectionId { get; set; }
+        /// <summary>
+        /// Gets or sets the first local packet sequence.
+        /// </summary>
+        public uint InitialSequence { get; set; } = 1;
+        /// <summary>
+        /// Gets or sets the resend delay in milliseconds.
+        /// </summary>
+        public long ResendDelayMs { get; set; } = 250;
+    }
+
+    /// <summary>
+    /// Send options for RUDP data packets.
+    /// </summary>
+    public sealed class CultNetRudpSendOptions
+    {
+        /// <summary>
+        /// Gets or sets whether the packet participates in reliable delivery.
+        /// </summary>
+        public bool Reliable { get; set; }
+        /// <summary>
+        /// Gets or sets whether the packet participates in ordered delivery.
+        /// </summary>
+        public bool Ordered { get; set; }
+        /// <summary>
+        /// Gets or sets whether the packet is latest-state sequenced.
+        /// </summary>
+        public bool Sequenced { get; set; }
+        /// <summary>
+        /// Gets or sets the current logical time in milliseconds.
+        /// </summary>
+        public long NowMs { get; set; }
+    }
+
+    /// <summary>
+    /// Socket-free reliability state machine for CultNet RUDP.
+    /// </summary>
+    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<uint> _receivedSequences = new HashSet<uint>();
+        private readonly Dictionary<uint, PendingReliablePacket> _pendingReliable = new Dictionary<uint, PendingReliablePacket>();
+        private readonly Dictionary<string, uint> _orderedNextSequenceByChannel = new Dictionary<string, uint>(StringComparer.Ordinal);
+        private readonly Dictionary<string, SortedDictionary<uint, CultNetRudpDeliveredFrame>> _orderedBuffers =
+            new Dictionary<string, SortedDictionary<uint, CultNetRudpDeliveredFrame>>(StringComparer.Ordinal);
+
+        /// <summary>
+        /// Initializes a RUDP reliability session.
+        /// </summary>
+        public CultNetRudpSession(CultNetRudpSessionOptions options)
+        {
+            if (options == null) throw new ArgumentNullException(nameof(options));
+            ConnectionId = options.ConnectionId;
+            _nextSequence = options.InitialSequence;
+            ResendDelayMs = options.ResendDelayMs;
+        }
+
+        /// <summary>
+        /// Gets the connection/session binding id.
+        /// </summary>
+        public uint ConnectionId { get; }
+        /// <summary>
+        /// Gets the resend delay in milliseconds.
+        /// </summary>
+        public long ResendDelayMs { get; }
+        /// <summary>
+        /// Gets whether the session has completed the connect/accept handshake.
+        /// </summary>
+        public bool Connected => _connected;
+        /// <summary>
+        /// Gets reliable packet sequences awaiting acknowledgement.
+        /// </summary>
+        public IReadOnlyList<uint> PendingReliableSequences => _pendingReliable.Keys.OrderBy(value => value).ToArray();
+
+        /// <summary>
+        /// Creates a reliable ordered connect packet.
+        /// </summary>
+        public CultNetRudpPacket CreateConnect(long nowMs = 0, byte[]? payload = null)
+        {
+            var packet = CreatePacket(CultNetRudpPacketType.Connect, "control", payload ?? Array.Empty<byte>(), reliable: true, ordered: true, sequenced: false);
+            TrackReliable(packet, nowMs);
+            return packet;
+        }
+
+        /// <summary>
+        /// Accepts a connect packet and returns a reliable ordered accept packet.
+        /// </summary>
+        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<byte>(), reliable: true, ordered: true, sequenced: false);
+            TrackReliable(response, nowMs);
+            return response;
+        }
+
+        /// <summary>
+        /// Creates a data packet.
+        /// </summary>
+        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;
+        }
+
+        /// <summary>
+        /// Applies a remote packet to the session.
+        /// </summary>
+        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<byte>(),
+                        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<byte>(),
+                Sequence = packet.Sequence
+            };
+
+            return new CultNetRudpReceiveResult
+            {
+                Delivered = packet.Ordered ? DeliverOrdered(frame) : new[] { frame }
+            };
+        }
+
+        /// <summary>
+        /// Creates a packet carrying the current acknowledgement state.
+        /// </summary>
+        public CultNetRudpPacket CreateAck()
+        {
+            return CreatePacket(CultNetRudpPacketType.Ack, "control", Array.Empty<byte>(), reliable: false, ordered: false, sequenced: false);
+        }
+
+        /// <summary>
+        /// Returns reliable packets due for resend at the supplied logical time.
+        /// </summary>
+        public IReadOnlyList<CultNetRudpPacket> DueResends(long nowMs)
+        {
+            var due = new List<CultNetRudpPacket>();
+            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<byte>()
+            };
+        }
+
+        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<CultNetRudpDeliveredFrame> 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<CultNetRudpDeliveredFrame>();
+            }
+
+            if (frame.Sequence > next)
+            {
+                if (!_orderedBuffers.TryGetValue(frame.ChannelId, out var buffer))
+                {
+                    buffer = new SortedDictionary<uint, CultNetRudpDeliveredFrame>();
+                    _orderedBuffers[frame.ChannelId] = buffer;
+                }
+
+                buffer[frame.Sequence] = frame;
+                return Array.Empty<CultNetRudpDeliveredFrame>();
+            }
+
+            _orderedNextSequenceByChannel[frame.ChannelId] = next + 1;
+            return new[] { frame }.Concat(DrainOrdered(frame.ChannelId)).ToArray();
+        }
+
+        private IReadOnlyList<CultNetRudpDeliveredFrame> DrainOrdered(string channelId)
+        {
+            var delivered = new List<CultNetRudpDeliveredFrame>();
+            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<byte>()
+            };
+        }
+    }
+
     /// <summary>
     /// Transport connection for the current TCP length-prefixed CultNet schema lane.
     /// </summary>

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