Serialize concurrent RUDP session publication

Directory.Build.props

@@ -14,7 +14,7 @@
     <PackageLicenseExpression>MIT</PackageLicenseExpression>
     <Copyright>© Meta Vens and contributors</Copyright>
     <PublishRepositoryUrl>true</PublishRepositoryUrl>
-	<Version>1.0.25</Version>
+	<Version>1.0.26</Version>
 
     <!-- Symbols/debugging -->
     <EmbedUntrackedSources>true</EmbedUntrackedSources>

src/GameCult.Mesh/docs/reliability-migration-plan.md

@@ -964,6 +964,37 @@ body path when remote demand exists; Phase 6 must replace that expiry-sensitive
 traffic with a stream/datagram body plane rather than enlarging snapshot or CDN
 timeouts.
 
+### RUDP session mutation body map
+
+**Owner:** each socket transport binding serializes access to its own RUDP
+reliability session. The server owns one serialization gate per peer; a client
+connection owns one gate for its session.
+
+**Inputs:** concurrent typed publication, command and receipt sends, inbound
+ACK/data processing, handshake state, timeout checks, and resend polling.
+
+**Outputs:** one coherent packet sequence and fragment stream per peer, with
+reliable-packet bookkeeping updated in the same critical section as packet
+creation and transmission.
+
+**Derived state:** schema messages, database subscriptions, and Eve reactive
+variables are consumers of the reliable control plane. They do not own packet
+ordering and must not serialize their own publication loops.
+
+**Forbidden writers:** daemon schedulers, renderers, and provider sessions
+cannot mutate `CultNetRudpSession` directly or add retries, replacement
+snapshots, and artificial publication delays to compensate for concurrent
+transport calls.
+
+**Shared paths:** application send, receive/ACK, resend, connect, disconnect,
+ping, and timeout inspection all enter through the same per-session gate.
+Fragments belonging to one logical send are emitted before another thread may
+allocate sequence numbers for that peer.
+
+**Cut line:** all socket-backed `CultNetRudpSession` mutation is behind the
+transport-owned gate. Product code no longer has to choose between concurrent
+simulation publication and reliable delivery.
+
 Reliable control messages larger than the RUDP receive mask explicitly
 acknowledge the packet just received. A retransmit older than the rolling
 32-packet summary can therefore always leave the sender queue; the rolling mask

src/GameCult.Networking/CultNetTransport.cs

@@ -1542,6 +1542,7 @@ namespace GameCult.Networking
         private const int WireBurstPackets = 32;
         private readonly Socket _socket;
         private readonly CultNetRudpSession _session;
+        private readonly object _sessionGate = new object();
         private readonly CultNetRudpSocketMode _mode;
         private readonly int? _maxFragmentBytes;
         private readonly CultNetTransportStats _stats = new CultNetTransportStats();
@@ -1594,7 +1595,7 @@ namespace GameCult.Networking
         /// <summary>
         /// Gets whether the RUDP handshake has completed.
         /// </summary>
-        public bool Connected => _session.Connected;
+        public bool Connected { get { lock (_sessionGate) return _session.Connected; } }
 
         /// <summary>
         /// Gets a snapshot of the current transfer counters.
@@ -1630,7 +1631,8 @@ namespace GameCult.Networking
                 throw new InvalidOperationException("Only a client RUDP socket transport can initiate connect.");
             }
 
-            SendPacket(_session.CreateConnect(NowMs(), payload ?? Array.Empty<byte>()));
+            lock (_sessionGate)
+                SendPacket(_session.CreateConnect(NowMs(), payload ?? Array.Empty<byte>()));
         }
 
         /// <summary>
@@ -1691,7 +1693,8 @@ namespace GameCult.Networking
         /// </summary>
         public void Send(string channelId, byte[] payload)
         {
-            SendPackets(_session.SendMany(channelId, payload, ChannelSendOptions(channelId), _maxFragmentBytes));
+            lock (_sessionGate)
+                SendPackets(_session.SendMany(channelId, payload, ChannelSendOptions(channelId), _maxFragmentBytes));
             _stats.FramesSent++;
         }
 
@@ -1844,7 +1847,8 @@ namespace GameCult.Networking
         /// </summary>
         public void Disconnect(byte[]? reason = null)
         {
-            SendPacket(_session.CreateDisconnect(reason ?? Array.Empty<byte>()));
+            lock (_sessionGate)
+                SendPacket(_session.CreateDisconnect(reason ?? Array.Empty<byte>()));
         }
 
         /// <summary>
@@ -1852,7 +1856,8 @@ namespace GameCult.Networking
         /// </summary>
         public void Ping(byte[]? payload = null)
         {
-            SendPacket(_session.CreatePing(payload ?? Array.Empty<byte>()));
+            lock (_sessionGate)
+                SendPacket(_session.CreatePing(payload ?? Array.Empty<byte>()));
         }
 
         /// <summary>
@@ -1860,7 +1865,8 @@ namespace GameCult.Networking
         /// </summary>
         public bool CheckTimeout(long timeoutMs)
         {
-            return _session.CheckTimeout(NowMs(), timeoutMs);
+            lock (_sessionGate)
+                return _session.CheckTimeout(NowMs(), timeoutMs);
         }
 
         /// <summary>
@@ -1929,14 +1935,21 @@ namespace GameCult.Networking
             TracePacket("rx", packet, received, remote);
             if (_mode == CultNetRudpSocketMode.Server && packet.PacketType == CultNetRudpPacketType.Connect)
             {
-                SendPacket(_session.AcceptConnect(packet, NowMs()));
+                lock (_sessionGate)
+                    SendPacket(_session.AcceptConnect(packet, NowMs()));
                 return true;
             }
 
-            var result = _session.Receive(packet, NowMs());
-            if (result.Reply != null)
+            CultNetRudpReceiveResult result;
+            CultNetRudpPacket? acknowledgement = null;
+            lock (_sessionGate)
             {
-                SendPacket(result.Reply);
+                result = _session.Receive(packet, NowMs());
+                if (result.Reply != null)
+                    SendPacket(result.Reply);
+                if (packet.PacketType == CultNetRudpPacketType.Accept ||
+                    packet.PacketType == CultNetRudpPacketType.Data)
+                    acknowledgement = _session.CreateAck(packet.Sequence);
             }
             if (result.Pong)
             {
@@ -1959,11 +1972,8 @@ namespace GameCult.Networking
             }
 
             delivered = _deliveredFrames.Count > 0 ? _deliveredFrames.Dequeue() : null;
-            if (packet.PacketType == CultNetRudpPacketType.Accept ||
-                packet.PacketType == CultNetRudpPacketType.Data)
-            {
-                SendPacket(_session.CreateAck(packet.Sequence));
-            }
+            if (acknowledgement != null)
+                SendPacket(acknowledgement);
 
             return true;
         }
@@ -1975,7 +1985,8 @@ namespace GameCult.Networking
         {
             if (_socket.Poll(0, SelectMode.SelectRead))
                 return;
-            SendPackets(_session.DueResends(NowMs()));
+            lock (_sessionGate)
+                SendPackets(_session.DueResends(NowMs()));
         }
 
         private void SendPackets(IReadOnlyList<CultNetRudpPacket> packets)
@@ -2102,6 +2113,7 @@ namespace GameCult.Networking
         }
 
         internal CultNetRudpSession Session { get; }
+        internal object SessionGate { get; } = new object();
 
         /// <summary>
         /// Gets the UDP endpoint for this peer.
@@ -2110,11 +2122,11 @@ namespace GameCult.Networking
         /// <summary>
         /// Gets whether the peer RUDP handshake has completed.
         /// </summary>
-        public bool Connected => Session.Connected;
+        public bool Connected { get { lock (SessionGate) return Session.Connected; } }
         /// <summary>
         /// Gets the number of reliable packets still awaiting acknowledgement.
         /// </summary>
-        public int PendingReliablePacketCount => Session.PendingReliableSequences.Count;
+        public int PendingReliablePacketCount { get { lock (SessionGate) return Session.PendingReliableSequences.Count; } }
         /// <summary>
         /// Gets the last transport-level remote disconnect reason, if one was received.
         /// </summary>
@@ -2215,7 +2227,8 @@ namespace GameCult.Networking
         public void Send(CultNetRudpSocketServerPeer peer, string channelId, byte[] payload)
         {
             if (peer == null) throw new ArgumentNullException(nameof(peer));
-            SendPackets(peer, peer.Session.SendMany(channelId, payload, ChannelSendOptions(channelId), _maxFragmentBytes));
+            lock (peer.SessionGate)
+                SendPackets(peer, peer.Session.SendMany(channelId, payload, ChannelSendOptions(channelId), _maxFragmentBytes));
             _stats.FramesSent++;
         }
 
@@ -2326,7 +2339,8 @@ namespace GameCult.Networking
                         MaxPendingReliablePackets = _maxPendingReliablePackets
                     }));
                 _peers[peerKey] = peer;
-                SendPacket(peer.RemoteEndPoint, peer.Session.AcceptConnect(packet, NowMs(), _acceptPayload));
+                lock (peer.SessionGate)
+                    SendPacket(peer.RemoteEndPoint, peer.Session.AcceptConnect(packet, NowMs(), _acceptPayload));
                 return true;
             }
 
@@ -2335,10 +2349,15 @@ namespace GameCult.Networking
                 return true;
             }
 
-            var result = existingPeer.Session.Receive(packet, NowMs());
-            if (result.Reply != null)
+            CultNetRudpReceiveResult result;
+            CultNetRudpPacket? acknowledgement = null;
+            lock (existingPeer.SessionGate)
             {
-                SendPacket(existingPeer.RemoteEndPoint, result.Reply);
+                result = existingPeer.Session.Receive(packet, NowMs());
+                if (result.Reply != null)
+                    SendPacket(existingPeer.RemoteEndPoint, result.Reply);
+                if (packet.PacketType == CultNetRudpPacketType.Data)
+                    acknowledgement = existingPeer.Session.CreateAck(packet.Sequence);
             }
             if (result.Disconnected)
             {
@@ -2363,10 +2382,8 @@ namespace GameCult.Networking
             }
 
             delivered = _deliveredFrames.Count > 0 ? _deliveredFrames.Dequeue() : null;
-            if (packet.PacketType == CultNetRudpPacketType.Data)
-            {
-                SendPacket(existingPeer.RemoteEndPoint, existingPeer.Session.CreateAck(packet.Sequence));
-            }
+            if (acknowledgement != null)
+                SendPacket(existingPeer.RemoteEndPoint, acknowledgement);
 
             return true;
         }
@@ -2380,7 +2397,8 @@ namespace GameCult.Networking
                 return;
             foreach (var peer in _peers.Values.ToArray())
             {
-                SendPackets(peer, peer.Session.DueResends(NowMs()));
+                lock (peer.SessionGate)
+                    SendPackets(peer, peer.Session.DueResends(NowMs()));
             }
         }
 

tests/GameCult.Networking.Tests/NetworkingTests.cs

@@ -1864,6 +1864,84 @@ namespace GameCult.Networking.Tests
                 "A lossless large response must not remain trapped in the reliable resend queue.");
         }
 
+        [Test]
+        public async Task RudpCultNetSchemaServer_SerializesConcurrentFragmentedSendsPerPeer()
+        {
+            const int publisherCount = 4;
+            const int messagesPerPublisher = 4;
+            const int expectedMessages = publisherCount * messagesPerPublisher;
+            using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions
+            {
+                RuntimeId = "csharp-rudp-concurrent-publish-host",
+                Socket = BindUdpSocket(),
+                MaxFragmentBytes = 512,
+                MaxPendingReliablePackets = 1024
+            });
+            var receivedIds = new ConcurrentDictionary<string, byte>();
+            var receivedAll = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var serverFailure = new TaskCompletionSource<Exception>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+            using var cancellation = new CancellationTokenSource();
+            var serverThread = new Thread(() =>
+            {
+                try
+                {
+                    while (!cancellation.IsCancellationRequested)
+                    {
+                        _ = server.PollOnceAsync().GetAwaiter().GetResult();
+                        Thread.Sleep(1);
+                    }
+                }
+                catch (Exception error) { serverFailure.TrySetResult(error); }
+            }) { IsBackground = true };
+            serverThread.Start();
+
+            using var client = CultNetSchemaClients.CreateRudp(
+                runtimeId: "csharp-rudp-concurrent-publish-client",
+                maxFragmentBytes: 512);
+            client.OnCultNet<CultNetSnapshotResponseRawMessage>(response =>
+            {
+                if (receivedIds.TryAdd(response.MessageId, 0) && receivedIds.Count == expectedMessages)
+                    receivedAll.TrySetResult(true);
+            });
+            client.Connect("127.0.0.1", server.LocalEndPoint.Port);
+            await WaitUntilAsync(() => client.Connected && server.Peers.Count == 1, TimeSpan.FromSeconds(2));
+            var peer = server.Peers.Single();
+            using var sendBarrier = new Barrier(publisherCount);
+
+            await Task.WhenAll(Enumerable.Range(0, publisherCount).Select(publisher => Task.Run(() =>
+            {
+                sendBarrier.SignalAndWait();
+                for (var message = 0; message < messagesPerPublisher; message++)
+                {
+                    var messageId = $"publisher-{publisher}-message-{message}";
+                    server.SendCultNet(peer, new CultNetSnapshotResponseRawMessage
+                    {
+                        MessageId = messageId,
+                        Documents =
+                        [
+                            new CultNetRawDocumentRecord
+                            {
+                                SchemaId = "test.concurrent.publish.v1",
+                                RecordKey = $"test:concurrent:{messageId}",
+                                PayloadEncoding = "messagepack",
+                                Payload = Enumerable.Range(0, 4 * 1024)
+                                    .Select(index => (byte)((index + publisher + message) % 251)).ToArray()
+                            }
+                        ]
+                    });
+                }
+            })));
+
+            var completed = await Task.WhenAny(receivedAll.Task, serverFailure.Task, Task.Delay(5000));
+            if (completed == serverFailure.Task) throw await serverFailure.Task;
+            if (completed != receivedAll.Task)
+                Assert.Fail($"Concurrent publish timed out after receiving {receivedIds.Count} of {expectedMessages} logical messages.");
+            cancellation.Cancel();
+
+            Assert.That(receivedIds.Count, Is.EqualTo(expectedMessages));
+        }
+
         [Test]
         public void RudpSocketTransport_ErgonomicHelpersCarryNamedChannels()
         {

unity/org.gamecult.cultlib/package.json

@@ -1,7 +1,7 @@
 {
   "name": "org.gamecult.cultlib",
   "displayName": "CultLib",
-  "version": "1.0.25",
+  "version": "1.0.26",
   "unity": "2021.3",
   "description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.",
   "license": "MIT",