diff --git a/src/GameCult.Mesh.Quic/CultMeshQuicRealtimeTransport.cs b/src/GameCult.Mesh.Quic/CultMeshQuicRealtimeTransport.cs index 2bf691a6..a0198a32 100644 --- a/src/GameCult.Mesh.Quic/CultMeshQuicRealtimeTransport.cs +++ b/src/GameCult.Mesh.Quic/CultMeshQuicRealtimeTransport.cs @@ -217,6 +217,18 @@ public sealed class CultMeshQuicRealtimeServer : IAsyncDisposable CancellationToken cancellationToken = default) { if (frame == null) throw new ArgumentNullException(nameof(frame)); + cancellationToken.ThrowIfCancellationRequested(); + + if (frame.Delivery == CultMeshRealtimeDelivery.LatestOnly) + { + foreach (var client in _clients.Keys) + { + if (client.TryPublishLatest(frame)) continue; + if (_clients.TryRemove(client, out _)) client.Dispose(); + } + return; + } + var sends = _clients.Keys.Select(client => SendToConnectedClientAsync(client, frame, cancellationToken)).ToArray(); if (sends.Length > 0) await Task.WhenAll(sends).ConfigureAwait(false); } @@ -304,8 +316,11 @@ internal sealed class CultMeshQuicRealtimeTransport : ICultMeshRealtimeTransport private readonly ConcurrentDictionary _latestGenerations = new(StringComparer.Ordinal); private readonly ConcurrentDictionary _streamReaders = new(); private readonly SemaphoreSlim _reliableSendGate = new(1, 1); + private readonly CultMeshRealtimeInbox _outbound = new(); private readonly Action? _observer; private readonly Task _acceptLoop; + private readonly Task _outboundLoop; + private readonly Task _completion; private QuicStream? _reliableOutbound; private int _nextReaderId; private bool _disposed; @@ -319,11 +334,19 @@ internal sealed class CultMeshQuicRealtimeTransport : ICultMeshRealtimeTransport _connection = connection; _observer = observer; _acceptLoop = AcceptStreamsAsync(_shutdown.Token); + _outboundLoop = SendPublishedFramesAsync(_shutdown.Token); + _completion = Task.WhenAll(_acceptLoop, _outboundLoop); } public string TransportId => "msquic-realtime"; public string Endpoint { get; } - public Task Completion => _acceptLoop; + public Task Completion => _completion; + + internal bool TryPublishLatest(CultMeshRealtimeFrame frame) + { + if (_disposed || frame.Delivery != CultMeshRealtimeDelivery.LatestOnly) return false; + return _outbound.Publish(frame); + } public async Task SendAsync(CultMeshRealtimeFrame frame, CancellationToken cancellationToken = default) { @@ -368,6 +391,7 @@ internal sealed class CultMeshQuicRealtimeTransport : ICultMeshRealtimeTransport { if (_disposed) return; _disposed = true; + _outbound.Complete(); _shutdown.Cancel(); _reliableOutbound?.DisposeAsync().AsTask().GetAwaiter().GetResult(); _connection.DisposeAsync().AsTask().GetAwaiter().GetResult(); @@ -376,6 +400,26 @@ internal sealed class CultMeshQuicRealtimeTransport : ICultMeshRealtimeTransport _shutdown.Dispose(); } + private async Task SendPublishedFramesAsync(CancellationToken cancellationToken) + { + try + { + while (!cancellationToken.IsCancellationRequested) + { + var frame = await _outbound.ReceiveAsync(cancellationToken).ConfigureAwait(false); + await SendAsync(frame, cancellationToken).ConfigureAwait(false); + } + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + } + catch + { + _shutdown.Cancel(); + throw; + } + } + private async Task OpenStreamAsync(byte kind, CancellationToken cancellationToken) { var stream = await _connection.OpenOutboundStreamAsync( @@ -515,21 +559,20 @@ internal sealed class CultMeshRealtimeInbox private readonly Dictionary _latest = new(StringComparer.Ordinal); private bool _completed; - public void Publish(CultMeshRealtimeFrame frame) + public bool Publish(CultMeshRealtimeFrame frame) { if (frame.Delivery != CultMeshRealtimeDelivery.LatestOnly) { - _ready.Writer.TryWrite(new InboxToken(frame, null)); - return; + return _ready.Writer.TryWrite(new InboxToken(frame, null)); } var key = frame.ChannelId + "\u001f" + frame.BodyId; lock (_gate) { - if (_completed) return; + if (_completed) return false; var alreadyPending = _latest.ContainsKey(key); _latest[key] = frame; - if (!alreadyPending) _ready.Writer.TryWrite(new InboxToken(null, key)); + return alreadyPending || _ready.Writer.TryWrite(new InboxToken(null, key)); } } diff --git a/src/GameCult.Mesh/docs/transport-planes.md b/src/GameCult.Mesh/docs/transport-planes.md index 2d529593..c1372105 100644 --- a/src/GameCult.Mesh/docs/transport-planes.md +++ b/src/GameCult.Mesh/docs/transport-planes.md @@ -49,9 +49,15 @@ advertised mapping when it is actually reachable; otherwise it connects to the advertised QUIC route. Latest-only receivers must replace pending generations per body rather than queueing render debt. -Provider broadcast isolates physical peers. A client disconnect or send failure -removes that session without failing the publication tick for healthy clients; -only caller cancellation stops the broadcast operation. +Provider simulation owns state generation, not socket completion. Latest-only +broadcast enqueues into a keyed, coalescing outbox owned by each physical peer +and returns without waiting for network delivery. The peer's transport worker +owns stream writes and eviction after disconnect or send failure. Thus a slow, +stalled, or departed client cannot backpressure a simulation tick, while a +healthy peer still converges on the newest generation. Reliable-ordered +broadcast remains completion-bearing because dropping or replacing it would +violate its declared delivery contract; caller cancellation can stop that +operation. **Derived state:** connector choice, physical endpoint, connection health, and transport diagnostics. None of these grant provider authority or become content diff --git a/tests/GameCult.Mesh.Quic.Native.Tests/CultMeshNativeQuicRealtimeTransportTests.cs b/tests/GameCult.Mesh.Quic.Native.Tests/CultMeshNativeQuicRealtimeTransportTests.cs index b491d912..c12cae03 100644 --- a/tests/GameCult.Mesh.Quic.Native.Tests/CultMeshNativeQuicRealtimeTransportTests.cs +++ b/tests/GameCult.Mesh.Quic.Native.Tests/CultMeshNativeQuicRealtimeTransportTests.cs @@ -92,19 +92,32 @@ public sealed class CultMeshNativeQuicRealtimeTransportTests var pin = Convert.ToHexString(SHA256.HashData(certificate.RawData)); var endpoint = $"cultmesh-state+quic://127.0.0.1:{server.LocalEndPoint.Port}?cert-sha256={pin}"; var connector = new CultMeshNativeQuicRealtimeTransportConnector(); - var client = await connector.ConnectAsync( + var departedClient = await connector.ConnectAsync( new CultMeshTransportCandidate(endpoint), CultMeshEndpointId.Parse("aetheria.local")); - await WaitUntilAsync(() => server.ConnectionCount == 1); + using var healthyClient = await connector.ConnectAsync( + new CultMeshTransportCandidate(endpoint), + CultMeshEndpointId.Parse("aetheria.local")); + await WaitUntilAsync(() => server.ConnectionCount == 2); await server.BroadcastAsync(Frame(sequence: 1)); using var receiveTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(5)); - await client.ReceiveAsync(receiveTimeout.Token); + await departedClient.ReceiveAsync(receiveTimeout.Token); + await healthyClient.ReceiveAsync(receiveTimeout.Token); - client.Dispose(); + departedClient.Dispose(); - Func broadcastAfterDeparture = () => server.BroadcastAsync(Frame(sequence: 2)); - await broadcastAfterDeparture.Should().NotThrowAsync( - "a short-lived realtime observer cannot terminate provider publication"); + for (var sequence = 2; sequence <= 100; sequence++) + await server.BroadcastAsync(Frame(sequence)).WaitAsync(TimeSpan.FromSeconds(1)); + + CultMeshRealtimeFrame received; + do + { + received = await healthyClient.ReceiveAsync(receiveTimeout.Token); + } + while (received.Sequence < 100); + + received.Sequence.Should().Be(100, + "a departed observer cannot stall provider publication or starve healthy peers"); } private static CultMeshRealtimeFrame Frame(long sequence) => new() diff --git a/unity/org.gamecult.cultlib/README.md b/unity/org.gamecult.cultlib/README.md index 136c38ab..fe294920 100644 --- a/unity/org.gamecult.cultlib/README.md +++ b/unity/org.gamecult.cultlib/README.md @@ -9,10 +9,10 @@ runtime; the Microsoft license is included under `Third Party Notices`. Build it with `scripts/build-unity-package.ps1`. Do not edit generated plugin assemblies in the package output. -Install version `1.0.37` through Unity Package Manager with: +Install version `1.0.38` through Unity Package Manager with: ```text -https://github.com/GameCult/CultLib.git?path=/unity/org.gamecult.cultlib#cultlib-unity-v1.0.37 +https://github.com/GameCult/CultLib.git?path=/unity/org.gamecult.cultlib#cultlib-unity-v1.0.38 ``` CultMesh chooses mapped memory for reachable same-machine bodies and advertised diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll index 9512f6b3..e6dbf77d 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb index d932026c..afbf39e5 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll index 575b4d98..addc7625 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb index 9cd13875..68fba322 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll index 6dbc721f..cd07fedc 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb index 2c47c069..03c2e6c1 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.dll index 288cbd05..2d5589bb 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.pdb index 515a7472..7354db8c 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll index 3aac9d60..1c38ba77 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb index 54429321..bd294f53 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll index a27f022d..6405385c 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb index 3bef8c59..ad23e4fb 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb differ diff --git a/unity/org.gamecult.cultlib/package.json b/unity/org.gamecult.cultlib/package.json index 08930273..1fdcdc7f 100644 --- a/unity/org.gamecult.cultlib/package.json +++ b/unity/org.gamecult.cultlib/package.json @@ -1,7 +1,7 @@ { "name": "org.gamecult.cultlib", "displayName": "CultLib", - "version": "1.0.37", + "version": "1.0.38", "unity": "2021.3", "description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.", "license": "MIT",