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

Add MsQuic realtime transport package

Browse source
This commit is contained in:
Meta Vens 2026-07-19 18:36:50 +01:00
commit 5e1633698e
7 changed files with 1020 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

30
CultLib.sln
View file

@ -31,6 +31,10 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "GameCult.Geometry.Tests", "
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "GameCult.Mesh.Tests", "tests\GameCult.Mesh.Tests\GameCult.Mesh.Tests.csproj", "{F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "GameCult.Mesh.Quic", "src\GameCult.Mesh.Quic\GameCult.Mesh.Quic.csproj", "{B67042B8-1135-4E54-B1EB-D45342C720D7}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "GameCult.Mesh.Quic.Tests", "tests\GameCult.Mesh.Quic.Tests\GameCult.Mesh.Quic.Tests.csproj", "{21F26329-2119-4037-93CC-DD1801A1AB78}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@ -185,6 +189,30 @@ Global
{F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}.Release|x64.Build.0 = Release|Any CPU
{F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}.Release|x86.ActiveCfg = Release|Any CPU
{F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}.Release|x86.Build.0 = Release|Any CPU
{B67042B8-1135-4E54-B1EB-D45342C720D7}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{B67042B8-1135-4E54-B1EB-D45342C720D7}.Debug|Any CPU.Build.0 = Debug|Any CPU
{B67042B8-1135-4E54-B1EB-D45342C720D7}.Debug|x64.ActiveCfg = Debug|Any CPU
{B67042B8-1135-4E54-B1EB-D45342C720D7}.Debug|x64.Build.0 = Debug|Any CPU
{B67042B8-1135-4E54-B1EB-D45342C720D7}.Debug|x86.ActiveCfg = Debug|Any CPU
{B67042B8-1135-4E54-B1EB-D45342C720D7}.Debug|x86.Build.0 = Debug|Any CPU
{B67042B8-1135-4E54-B1EB-D45342C720D7}.Release|Any CPU.ActiveCfg = Release|Any CPU
{B67042B8-1135-4E54-B1EB-D45342C720D7}.Release|Any CPU.Build.0 = Release|Any CPU
{B67042B8-1135-4E54-B1EB-D45342C720D7}.Release|x64.ActiveCfg = Release|Any CPU
{B67042B8-1135-4E54-B1EB-D45342C720D7}.Release|x64.Build.0 = Release|Any CPU
{B67042B8-1135-4E54-B1EB-D45342C720D7}.Release|x86.ActiveCfg = Release|Any CPU
{B67042B8-1135-4E54-B1EB-D45342C720D7}.Release|x86.Build.0 = Release|Any CPU
{21F26329-2119-4037-93CC-DD1801A1AB78}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{21F26329-2119-4037-93CC-DD1801A1AB78}.Debug|Any CPU.Build.0 = Debug|Any CPU
{21F26329-2119-4037-93CC-DD1801A1AB78}.Debug|x64.ActiveCfg = Debug|Any CPU
{21F26329-2119-4037-93CC-DD1801A1AB78}.Debug|x64.Build.0 = Debug|Any CPU
{21F26329-2119-4037-93CC-DD1801A1AB78}.Debug|x86.ActiveCfg = Debug|Any CPU
{21F26329-2119-4037-93CC-DD1801A1AB78}.Debug|x86.Build.0 = Debug|Any CPU
{21F26329-2119-4037-93CC-DD1801A1AB78}.Release|Any CPU.ActiveCfg = Release|Any CPU
{21F26329-2119-4037-93CC-DD1801A1AB78}.Release|Any CPU.Build.0 = Release|Any CPU
{21F26329-2119-4037-93CC-DD1801A1AB78}.Release|x64.ActiveCfg = Release|Any CPU
{21F26329-2119-4037-93CC-DD1801A1AB78}.Release|x64.Build.0 = Release|Any CPU
{21F26329-2119-4037-93CC-DD1801A1AB78}.Release|x86.ActiveCfg = Release|Any CPU
{21F26329-2119-4037-93CC-DD1801A1AB78}.Release|x86.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
@ -202,5 +230,7 @@ Global
{94C086DA-3359-424B-9688-29A198806481} = {E733CBFF-42AE-4B6A-B5B9-0071ECFB45F1}
{C95A9A21-34FA-4D82-BE5B-88FBDA836D31} = {39D316AB-ABB5-49A2-8E83-CAF4C75B48F4}
{F8A5060A-FB0B-4B8B-AE06-5147BB4CA127} = {39D316AB-ABB5-49A2-8E83-CAF4C75B48F4}
{B67042B8-1135-4E54-B1EB-D45342C720D7} = {E733CBFF-42AE-4B6A-B5B9-0071ECFB45F1}
{21F26329-2119-4037-93CC-DD1801A1AB78} = {39D316AB-ABB5-49A2-8E83-CAF4C75B48F4}
EndGlobalSection
EndGlobal

626
src/GameCult.Mesh.Quic/CultMeshQuicRealtimeTransport.cs Normal file
View file

@ -0,0 +1,626 @@
using System.Buffers.Binary;
using System.Collections.Concurrent;
using System.Net;
using System.Net.Quic;
using System.Net.Security;
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using System.Threading.Channels;
namespace GameCult.Mesh.Quic;
/// <summary>Configures identity-bound CultMesh QUIC client connections.</summary>
public sealed class CultMeshQuicRealtimeConnectorOptions
{
/// <summary>Application protocol negotiated during the TLS handshake.</summary>
public SslApplicationProtocol ApplicationProtocol { get; set; } = CultMeshQuicRealtimeProtocol.ApplicationProtocol;
/// <summary>Maximum time permitted for connection establishment.</summary>
public TimeSpan HandshakeTimeout { get; set; } = TimeSpan.FromSeconds(10);
/// <summary>
/// Optional provider-aware certificate validator. Standard TLS validation is used when omitted.
/// </summary>
public Func<CultMeshEndpointId, X509Certificate2?, X509Chain?, SslPolicyErrors, bool>?
ValidateProviderCertificate { get; set; }
}
/// <summary>Creates MsQuic-backed realtime state connections for .NET runtimes.</summary>
public sealed class CultMeshQuicRealtimeTransportConnector : ICultMeshRealtimeTransportConnector
{
/// <summary>Discovery scheme owned by this connector.</summary>
public const string Scheme = "cultmesh-state+quic";
private readonly CultMeshQuicRealtimeConnectorOptions _options;
/// <summary>Creates a QUIC connector.</summary>
public CultMeshQuicRealtimeTransportConnector(CultMeshQuicRealtimeConnectorOptions? options = null)
{
_options = options ?? new CultMeshQuicRealtimeConnectorOptions();
if (_options.HandshakeTimeout <= TimeSpan.Zero)
throw new ArgumentOutOfRangeException(nameof(options), "QUIC handshake timeout must be positive.");
}
/// <inheritdoc />
public string ConnectorId => "msquic-realtime";
/// <inheritdoc />
public int Priority => 0;
/// <inheritdoc />
public bool CanConnect(CultMeshTransportCandidate candidate) =>
candidate != null && TryParseEndpoint(candidate.Endpoint, out _, out _);
/// <inheritdoc />
public async Task<ICultMeshRealtimeTransport> ConnectAsync(
CultMeshTransportCandidate candidate,
CultMeshEndpointId endpointId,
CancellationToken cancellationToken = default)
{
if (candidate == null) throw new ArgumentNullException(nameof(candidate));
if (endpointId == null) throw new ArgumentNullException(nameof(endpointId));
if (!QuicConnection.IsSupported)
throw new PlatformNotSupportedException("QUIC is unavailable. Install a supported MsQuic runtime.");
if (!TryParseEndpoint(candidate.Endpoint, out var host, out var port))
throw new NotSupportedException($"QUIC realtime connector does not support '{candidate.Endpoint}'.");
var validator = _options.ValidateProviderCertificate;
EndPoint remoteEndPoint = IPAddress.TryParse(host, out var address)
? new IPEndPoint(address, port)
: new DnsEndPoint(host, port);
var connection = await QuicConnection.ConnectAsync(new QuicClientConnectionOptions
{
RemoteEndPoint = remoteEndPoint,
DefaultCloseErrorCode = CultMeshQuicRealtimeProtocol.ConnectionCloseCode,
DefaultStreamErrorCode = CultMeshQuicRealtimeProtocol.StreamAbortCode,
MaxInboundUnidirectionalStreams = 1024,
HandshakeTimeout = _options.HandshakeTimeout,
ClientAuthenticationOptions = new SslClientAuthenticationOptions
{
ApplicationProtocols = new List<SslApplicationProtocol> { _options.ApplicationProtocol },
EnabledSslProtocols = SslProtocols.Tls13,
RemoteCertificateValidationCallback = (_, certificate, chain, errors) =>
validator?.Invoke(endpointId, certificate as X509Certificate2, chain, errors) ??
errors == SslPolicyErrors.None
}
}, cancellationToken).ConfigureAwait(false);
return new CultMeshQuicRealtimeTransport(candidate.Endpoint, connection);
}
private static bool TryParseEndpoint(string? endpoint, out string host, out int port)
{
host = string.Empty;
port = 0;
if (!Uri.TryCreate(endpoint, UriKind.Absolute, out var uri) ||
!string.Equals(uri.Scheme, Scheme, StringComparison.OrdinalIgnoreCase) ||
string.IsNullOrWhiteSpace(uri.Host) || uri.Port <= 0)
return false;
host = uri.Host;
port = uri.Port;
return true;
}
}
/// <summary>Configures an authenticated CultMesh QUIC realtime listener.</summary>
public sealed class CultMeshQuicRealtimeServerOptions
{
/// <summary>Local endpoint to bind.</summary>
public IPEndPoint ListenEndPoint { get; set; } = new(IPAddress.Loopback, 0);
/// <summary>Server certificate used by QUIC TLS 1.3.</summary>
public X509Certificate2 ServerCertificate { get; set; } = null!;
/// <summary>Application protocol negotiated during the TLS handshake.</summary>
public SslApplicationProtocol ApplicationProtocol { get; set; } = CultMeshQuicRealtimeProtocol.ApplicationProtocol;
/// <summary>Maximum time permitted for connection establishment.</summary>
public TimeSpan HandshakeTimeout { get; set; } = TimeSpan.FromSeconds(10);
}
/// <summary>
/// Accepts .NET QUIC realtime sessions. It owns physical connections only; provider
/// authorization and the meaning of frames remain outside this adapter.
/// </summary>
public sealed class CultMeshQuicRealtimeServer : IAsyncDisposable
{
private readonly QuicListener _listener;
private readonly CancellationTokenSource _shutdown = new();
private readonly ConcurrentDictionary<CultMeshQuicRealtimeTransport, byte> _clients = new();
private readonly CultMeshRealtimeInbox _received = new();
private readonly TaskCompletionSource<Exception> _backgroundFailure = new(
TaskCreationOptions.RunContinuationsAsynchronously);
private readonly Task _acceptLoop;
private bool _disposed;
private CultMeshQuicRealtimeServer(QuicListener listener)
{
_listener = listener;
_acceptLoop = AcceptLoopAsync(_shutdown.Token);
}
/// <summary>Bound listener endpoint.</summary>
public IPEndPoint LocalEndPoint => (IPEndPoint)_listener.LocalEndPoint;
/// <summary>Number of active physical client connections.</summary>
public int ConnectionCount => _clients.Count;
/// <summary>Completes if the listener's background accept loop fails.</summary>
public Task<Exception> BackgroundFailure => _backgroundFailure.Task;
/// <summary>Creates and starts an authenticated QUIC listener.</summary>
public static async Task<CultMeshQuicRealtimeServer> ListenAsync(
CultMeshQuicRealtimeServerOptions options,
CancellationToken cancellationToken = default)
{
if (options == null) throw new ArgumentNullException(nameof(options));
if (options.ServerCertificate == null)
throw new ArgumentException("A QUIC server certificate is required.", nameof(options));
if (options.HandshakeTimeout <= TimeSpan.Zero)
throw new ArgumentOutOfRangeException(nameof(options), "QUIC handshake timeout must be positive.");
if (!QuicListener.IsSupported)
throw new PlatformNotSupportedException("QUIC is unavailable. Install a supported MsQuic runtime.");
var listener = await QuicListener.ListenAsync(new QuicListenerOptions
{
ListenEndPoint = options.ListenEndPoint,
ApplicationProtocols = new List<SslApplicationProtocol> { options.ApplicationProtocol },
ConnectionOptionsCallback = (_, _, _) => ValueTask.FromResult(new QuicServerConnectionOptions
{
DefaultCloseErrorCode = CultMeshQuicRealtimeProtocol.ConnectionCloseCode,
DefaultStreamErrorCode = CultMeshQuicRealtimeProtocol.StreamAbortCode,
MaxInboundUnidirectionalStreams = 1024,
HandshakeTimeout = options.HandshakeTimeout,
ServerAuthenticationOptions = new SslServerAuthenticationOptions
{
ApplicationProtocols = new List<SslApplicationProtocol> { options.ApplicationProtocol },
EnabledSslProtocols = SslProtocols.Tls13,
ServerCertificate = options.ServerCertificate
}
})
}, cancellationToken).ConfigureAwait(false);
return new CultMeshQuicRealtimeServer(listener);
}
/// <summary>Broadcasts one state frame to the currently connected clients.</summary>
public async Task BroadcastAsync(
CultMeshRealtimeFrame frame,
CancellationToken cancellationToken = default)
{
if (frame == null) throw new ArgumentNullException(nameof(frame));
var sends = _clients.Keys.Select(client => client.SendAsync(frame, cancellationToken)).ToArray();
if (sends.Length > 0) await Task.WhenAll(sends).ConfigureAwait(false);
}
/// <summary>Receives the next client-originated state frame.</summary>
public Task<CultMeshRealtimeFrame> ReceiveAsync(CancellationToken cancellationToken = default) =>
_received.ReceiveAsync(cancellationToken);
/// <inheritdoc />
public async ValueTask DisposeAsync()
{
if (_disposed) return;
_disposed = true;
_shutdown.Cancel();
await _listener.DisposeAsync().ConfigureAwait(false);
foreach (var client in _clients.Keys) client.Dispose();
_clients.Clear();
_received.Complete();
try { await _acceptLoop.ConfigureAwait(false); }
catch (OperationCanceledException) { }
_shutdown.Dispose();
}
private async Task AcceptLoopAsync(CancellationToken cancellationToken)
{
try
{
while (!cancellationToken.IsCancellationRequested)
{
var connection = await _listener.AcceptConnectionAsync(cancellationToken).ConfigureAwait(false);
var transport = new CultMeshQuicRealtimeTransport(
connection.RemoteEndPoint?.ToString() ?? "quic-peer",
connection,
frame => _received.Publish(frame));
_clients.TryAdd(transport, 0);
_ = transport.Completion.ContinueWith(
completed =>
{
_clients.TryRemove(transport, out _);
transport.Dispose();
},
CancellationToken.None,
TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default);
}
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
}
catch (ObjectDisposedException) when (_disposed)
{
}
catch (Exception error)
{
_backgroundFailure.TrySetResult(error);
}
}
}
internal sealed class CultMeshQuicRealtimeTransport : ICultMeshRealtimeTransport
{
private readonly QuicConnection _connection;
private readonly CancellationTokenSource _shutdown = new();
private readonly CultMeshRealtimeInbox _received = new();
private readonly ConcurrentDictionary<string, CultMeshGeneration> _latestGenerations = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<int, Task> _streamReaders = new();
private readonly SemaphoreSlim _reliableSendGate = new(1, 1);
private readonly Action<CultMeshRealtimeFrame>? _observer;
private readonly Task _acceptLoop;
private QuicStream? _reliableOutbound;
private int _nextReaderId;
private bool _disposed;
public CultMeshQuicRealtimeTransport(
string endpoint,
QuicConnection connection,
Action<CultMeshRealtimeFrame>? observer = null)
{
Endpoint = endpoint;
_connection = connection;
_observer = observer;
_acceptLoop = AcceptStreamsAsync(_shutdown.Token);
}
public string TransportId => "msquic-realtime";
public string Endpoint { get; }
public Task Completion => _acceptLoop;
public async Task SendAsync(CultMeshRealtimeFrame frame, CancellationToken cancellationToken = default)
{
if (_disposed) throw new ObjectDisposedException(nameof(CultMeshQuicRealtimeTransport));
if (frame == null) throw new ArgumentNullException(nameof(frame));
if (frame.Delivery == CultMeshRealtimeDelivery.Unreliable)
throw new NotSupportedException(
"System.Net.Quic exposes streams but not QUIC datagrams; unreliable delivery requires a native MsQuic connector.");
var payload = CultMeshQuicRealtimeProtocol.EncodeFrame(frame);
if (frame.Delivery == CultMeshRealtimeDelivery.ReliableOrdered)
{
await _reliableSendGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
_reliableOutbound ??= await OpenStreamAsync(
CultMeshQuicRealtimeProtocol.ReliableStream,
cancellationToken).ConfigureAwait(false);
await CultMeshQuicRealtimeProtocol.WriteFramedAsync(
_reliableOutbound,
payload,
cancellationToken).ConfigureAwait(false);
}
finally
{
_reliableSendGate.Release();
}
return;
}
await using var stream = await OpenStreamAsync(
CultMeshQuicRealtimeProtocol.LatestOnlyStream,
cancellationToken).ConfigureAwait(false);
await CultMeshQuicRealtimeProtocol.WriteFramedAsync(stream, payload, cancellationToken).ConfigureAwait(false);
stream.CompleteWrites();
}
public Task<CultMeshRealtimeFrame> ReceiveAsync(CancellationToken cancellationToken = default) =>
_received.ReceiveAsync(cancellationToken);
public void Dispose()
{
if (_disposed) return;
_disposed = true;
_shutdown.Cancel();
_reliableOutbound?.DisposeAsync().AsTask().GetAwaiter().GetResult();
_connection.DisposeAsync().AsTask().GetAwaiter().GetResult();
_received.Complete();
_reliableSendGate.Dispose();
_shutdown.Dispose();
}
private async Task<QuicStream> OpenStreamAsync(byte kind, CancellationToken cancellationToken)
{
var stream = await _connection.OpenOutboundStreamAsync(
QuicStreamType.Unidirectional,
cancellationToken).ConfigureAwait(false);
await stream.WriteAsync(new[] { kind }, cancellationToken).ConfigureAwait(false);
return stream;
}
private async Task AcceptStreamsAsync(CancellationToken cancellationToken)
{
try
{
while (!cancellationToken.IsCancellationRequested)
{
var stream = await _connection.AcceptInboundStreamAsync(cancellationToken).ConfigureAwait(false);
var readerId = Interlocked.Increment(ref _nextReaderId);
var reader = ReadStreamOwnedAsync(readerId, stream, cancellationToken);
_streamReaders.TryAdd(readerId, reader);
}
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
}
catch (QuicException) when (_disposed)
{
}
finally
{
try
{
await Task.WhenAll(_streamReaders.Values).ConfigureAwait(false);
_received.Complete();
}
catch (Exception error)
{
_received.Complete(error);
throw;
}
}
}
private async Task ReadStreamOwnedAsync(
int readerId,
QuicStream stream,
CancellationToken cancellationToken)
{
try
{
await ReadStreamAsync(stream, cancellationToken).ConfigureAwait(false);
}
catch (Exception error) when (!_disposed && error is not OperationCanceledException)
{
_received.Complete(error);
_shutdown.Cancel();
}
finally
{
_streamReaders.TryRemove(readerId, out _);
}
}
private async Task ReadStreamAsync(QuicStream stream, CancellationToken cancellationToken)
{
await using (stream.ConfigureAwait(false))
{
try
{
var kind = await CultMeshQuicRealtimeProtocol.ReadByteAsync(stream, cancellationToken).ConfigureAwait(false);
if (kind != CultMeshQuicRealtimeProtocol.ReliableStream &&
kind != CultMeshQuicRealtimeProtocol.LatestOnlyStream)
throw new InvalidDataException($"Unknown CultMesh QUIC stream kind {kind}.");
do
{
var payload = await CultMeshQuicRealtimeProtocol.ReadFramedAsync(stream, cancellationToken).ConfigureAwait(false);
if (payload == null) break;
var frame = CultMeshQuicRealtimeProtocol.DecodeFrame(payload);
if (kind == CultMeshQuicRealtimeProtocol.ReliableStream &&
frame.Delivery != CultMeshRealtimeDelivery.ReliableOrdered)
throw new InvalidDataException("Reliable QUIC stream carried incompatible delivery semantics.");
if (kind == CultMeshQuicRealtimeProtocol.LatestOnlyStream &&
frame.Delivery != CultMeshRealtimeDelivery.LatestOnly)
throw new InvalidDataException("Latest-only QUIC stream carried incompatible delivery semantics.");
PublishReceived(frame);
}
while (kind == CultMeshQuicRealtimeProtocol.ReliableStream);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
}
catch (QuicException) when (_disposed)
{
}
}
}
private void PublishReceived(CultMeshRealtimeFrame frame)
{
if (frame.Delivery == CultMeshRealtimeDelivery.LatestOnly)
{
var key = frame.ChannelId + "\u001f" + frame.BodyId;
var generation = new CultMeshGeneration(frame.ProducerEpoch, frame.Sequence);
while (true)
{
if (_latestGenerations.TryGetValue(key, out var current))
{
if (generation.CompareTo(current) <= 0) return;
if (!_latestGenerations.TryUpdate(key, generation, current)) continue;
}
else if (!_latestGenerations.TryAdd(key, generation))
{
continue;
}
break;
}
}
_observer?.Invoke(frame);
_received.Publish(frame);
}
private readonly record struct CultMeshGeneration(long ProducerEpoch, long Sequence) : IComparable<CultMeshGeneration>
{
public int CompareTo(CultMeshGeneration other)
{
var epochOrder = ProducerEpoch.CompareTo(other.ProducerEpoch);
return epochOrder != 0 ? epochOrder : Sequence.CompareTo(other.Sequence);
}
}
}
internal sealed class CultMeshRealtimeInbox
{
private readonly object _gate = new();
private readonly Channel<InboxToken> _ready = Channel.CreateUnbounded<InboxToken>(
new UnboundedChannelOptions { SingleReader = false, SingleWriter = false });
private readonly Dictionary<string, CultMeshRealtimeFrame> _latest = new(StringComparer.Ordinal);
private bool _completed;
public void Publish(CultMeshRealtimeFrame frame)
{
if (frame.Delivery != CultMeshRealtimeDelivery.LatestOnly)
{
_ready.Writer.TryWrite(new InboxToken(frame, null));
return;
}
var key = frame.ChannelId + "\u001f" + frame.BodyId;
lock (_gate)
{
if (_completed) return;
var alreadyPending = _latest.ContainsKey(key);
_latest[key] = frame;
if (!alreadyPending) _ready.Writer.TryWrite(new InboxToken(null, key));
}
}
public async Task<CultMeshRealtimeFrame> ReceiveAsync(CancellationToken cancellationToken)
{
while (true)
{
var token = await _ready.Reader.ReadAsync(cancellationToken).ConfigureAwait(false);
if (token.Frame != null) return token.Frame;
lock (_gate)
{
if (token.LatestKey != null && _latest.Remove(token.LatestKey, out var latest))
return latest;
}
}
}
public void Complete(Exception? error = null)
{
lock (_gate) _completed = true;
_ready.Writer.TryComplete(error);
}
private sealed record InboxToken(CultMeshRealtimeFrame? Frame, string? LatestKey);
}
internal static class CultMeshQuicRealtimeProtocol
{
public static readonly SslApplicationProtocol ApplicationProtocol = new("cultmesh-state-v1");
public const long ConnectionCloseCode = 0x43554c54;
public const long StreamAbortCode = 0x53544154;
public const byte ReliableStream = 1;
public const byte LatestOnlyStream = 2;
private const uint Magic = 0x31545343;
private const int FixedHeaderBytes = 37;
private const int MaximumFrameBytes = 64 * 1024 * 1024;
public static byte[] EncodeFrame(CultMeshRealtimeFrame frame)
{
var channel = Encoding.UTF8.GetBytes(frame.ChannelId);
var schema = Encoding.UTF8.GetBytes(frame.SchemaId);
var body = Encoding.UTF8.GetBytes(frame.BodyId);
if (channel.Length > ushort.MaxValue || schema.Length > ushort.MaxValue || body.Length > ushort.MaxValue)
throw new InvalidDataException("Realtime frame identity exceeds the QUIC wire limit.");
if (frame.Payload.Length > MaximumFrameBytes)
throw new InvalidDataException("Realtime frame payload exceeds the QUIC wire limit.");
var result = new byte[checked(FixedHeaderBytes + channel.Length + schema.Length + body.Length + frame.Payload.Length)];
var span = result.AsSpan();
BinaryPrimitives.WriteUInt32LittleEndian(span, Magic);
span[4] = (byte)frame.Delivery;
BinaryPrimitives.WriteInt64LittleEndian(span[5..], frame.ProducerEpoch);
BinaryPrimitives.WriteInt64LittleEndian(span[13..], frame.Sequence);
BinaryPrimitives.WriteUInt16LittleEndian(span[21..], (ushort)channel.Length);
BinaryPrimitives.WriteUInt16LittleEndian(span[23..], (ushort)schema.Length);
BinaryPrimitives.WriteUInt16LittleEndian(span[25..], (ushort)body.Length);
BinaryPrimitives.WriteInt32LittleEndian(span[27..], frame.Payload.Length);
BinaryPrimitives.WriteInt32LittleEndian(span[31..], FixedHeaderBytes);
BinaryPrimitives.WriteUInt16LittleEndian(span[35..], 1);
var offset = FixedHeaderBytes;
channel.CopyTo(span[offset..]); offset += channel.Length;
schema.CopyTo(span[offset..]); offset += schema.Length;
body.CopyTo(span[offset..]); offset += body.Length;
frame.Payload.Span.CopyTo(span[offset..]);
return result;
}
public static CultMeshRealtimeFrame DecodeFrame(byte[] bytes)
{
if (bytes.Length < FixedHeaderBytes) throw new InvalidDataException("Realtime frame header is truncated.");
var span = bytes.AsSpan();
if (BinaryPrimitives.ReadUInt32LittleEndian(span) != Magic)
throw new InvalidDataException("Realtime frame magic is invalid.");
if (BinaryPrimitives.ReadUInt16LittleEndian(span[35..]) != 1 ||
BinaryPrimitives.ReadInt32LittleEndian(span[31..]) != FixedHeaderBytes)
throw new InvalidDataException("Realtime frame wire version is unsupported.");
var delivery = (CultMeshRealtimeDelivery)span[4];
if (delivery is < CultMeshRealtimeDelivery.ReliableOrdered or > CultMeshRealtimeDelivery.Unreliable)
throw new InvalidDataException("Realtime frame delivery mode is invalid.");
var channelLength = BinaryPrimitives.ReadUInt16LittleEndian(span[21..]);
var schemaLength = BinaryPrimitives.ReadUInt16LittleEndian(span[23..]);
var bodyLength = BinaryPrimitives.ReadUInt16LittleEndian(span[25..]);
var payloadLength = BinaryPrimitives.ReadInt32LittleEndian(span[27..]);
var expected = checked(FixedHeaderBytes + channelLength + schemaLength + bodyLength + payloadLength);
if (payloadLength < 0 || expected != bytes.Length)
throw new InvalidDataException("Realtime frame length is invalid.");
var offset = FixedHeaderBytes;
var channel = Encoding.UTF8.GetString(span.Slice(offset, channelLength)); offset += channelLength;
var schema = Encoding.UTF8.GetString(span.Slice(offset, schemaLength)); offset += schemaLength;
var body = Encoding.UTF8.GetString(span.Slice(offset, bodyLength)); offset += bodyLength;
return new CultMeshRealtimeFrame
{
ChannelId = channel,
SchemaId = schema,
BodyId = body,
ProducerEpoch = BinaryPrimitives.ReadInt64LittleEndian(span[5..]),
Sequence = BinaryPrimitives.ReadInt64LittleEndian(span[13..]),
Delivery = delivery,
Payload = bytes.AsMemory(offset, payloadLength)
};
}
public static async Task WriteFramedAsync(Stream stream, byte[] payload, CancellationToken cancellationToken)
{
var header = new byte[4];
BinaryPrimitives.WriteInt32LittleEndian(header, payload.Length);
await stream.WriteAsync(header, cancellationToken).ConfigureAwait(false);
await stream.WriteAsync(payload, cancellationToken).ConfigureAwait(false);
}
public static async Task<byte[]?> ReadFramedAsync(Stream stream, CancellationToken cancellationToken)
{
var header = new byte[4];
var first = await stream.ReadAsync(header.AsMemory(0, 1), cancellationToken).ConfigureAwait(false);
if (first == 0) return null;
await ReadExactlyAsync(stream, header.AsMemory(1), cancellationToken).ConfigureAwait(false);
var length = BinaryPrimitives.ReadInt32LittleEndian(header);
if (length <= 0 || length > MaximumFrameBytes + FixedHeaderBytes + (3 * ushort.MaxValue))
throw new InvalidDataException("Realtime QUIC frame length is invalid.");
var payload = new byte[length];
await ReadExactlyAsync(stream, payload, cancellationToken).ConfigureAwait(false);
return payload;
}
public static async Task<byte> ReadByteAsync(Stream stream, CancellationToken cancellationToken)
{
var value = new byte[1];
await ReadExactlyAsync(stream, value, cancellationToken).ConfigureAwait(false);
return value[0];
}
private static async Task ReadExactlyAsync(Stream stream, Memory<byte> destination, CancellationToken cancellationToken)
{
var offset = 0;
while (offset < destination.Length)
{
var read = await stream.ReadAsync(destination[offset..], cancellationToken).ConfigureAwait(false);
if (read == 0) throw new EndOfStreamException("QUIC stream ended before its frame was complete.");
offset += read;
}
}
}

16
src/GameCult.Mesh.Quic/GameCult.Mesh.Quic.csproj Normal file
View file

@ -0,0 +1,16 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<Description>MsQuic-backed realtime state transport for CultMesh</Description>
<PackageId>GameCult.Mesh.Quic</PackageId>
<PackageTags>game;distributed;realtime;mesh;quic;msquic</PackageTags>
<PackageReadmeFile>README.md</PackageReadmeFile>
<NoWarn>$(NoWarn);CA1416</NoWarn>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\GameCult.Mesh\GameCult.Mesh.csproj" />
<None Include="README.md" Pack="true" PackagePath="\" />
</ItemGroup>
</Project>

72
src/GameCult.Mesh.Quic/README.md Normal file
View file

@ -0,0 +1,72 @@
# GameCult.Mesh.Quic
`GameCult.Mesh.Quic` is the optional MsQuic-backed realtime state plane for
CultMesh daemons and .NET tools. It implements the transport-neutral contracts
from `GameCult.Mesh`; it does not own discovery, provider authorization, state
meaning, commands, receipts, schemas, or immutable content.
The package targets .NET 10. It uses `System.Net.Quic`, which requires a
supported MsQuic runtime and TLS 1.3. Unity runtimes do not consume this package;
their generic CultMesh lowerer needs a native MsQuic connector implementing the
same `ICultMeshRealtimeTransportConnector` contract.
## Provider
```csharp
using GameCult.Mesh.Quic;
await using var server = await CultMeshQuicRealtimeServer.ListenAsync(
new CultMeshQuicRealtimeServerOptions
{
ListenEndPoint = new IPEndPoint(IPAddress.Any, 7443),
ServerCertificate = providerCertificate
});
await server.BroadcastAsync(new CultMeshRealtimeFrame
{
ChannelId = "aetheria.entities",
SchemaId = "eve.entity_soa.v1",
BodyId = "body:aetheria:entities",
ProducerEpoch = epoch,
Sequence = sequence,
Delivery = CultMeshRealtimeDelivery.LatestOnly,
Payload = encodedSoa
});
```
Advertise the listener as `cultmesh-state+quic://host:port`. Certificate
identity and provider identity must be bound by the application trust policy;
accepting every certificate is appropriate only in isolated tests.
## Consumer
```csharp
var connector = new CultMeshQuicRealtimeTransportConnector(
new CultMeshQuicRealtimeConnectorOptions
{
ValidateProviderCertificate = ValidateAdvertisedProvider
});
using var sessions = new CultMeshSessionManager(
discovery,
schemaConnectors,
contentConnectors,
new ICultMeshRealtimeTransportConnector[] { connector });
var session = await sessions.ConnectRealtimeAsync(providerId);
var frame = await session.ReceiveAsync(cancellationToken);
```
## Delivery modes
- `ReliableOrdered` uses one ordered QUIC stream.
- `LatestOnly` uses independent streams and keeps at most one pending frame per
channel and body. A newer `(producer epoch, sequence)` generation replaces
queued state.
- `Unreliable` fails closed in this adapter because `System.Net.Quic` does not
expose QUIC datagrams. Use a native MsQuic connector when datagram semantics
are required.
Schemas, commands, receipts, manifests, and immutable content remain on their
TCP planes. Same-machine SoA should negotiate mapped/shared memory instead of
crossing QUIC.

16
src/GameCult.Mesh/docs/transport-planes.md
View file

@ -30,8 +30,15 @@ and atomic promotion regardless of the selected content connector.
Passing explicit connector lists replaces those defaults.
Realtime frames declare whether they require reliable ordered delivery,
latest-only delivery, or unreliable delivery. The contract is ready for a QUIC
connector; no RUDP implementation is installed implicitly in its place.
latest-only delivery, or unreliable delivery. The optional
`GameCult.Mesh.Quic` package provides the .NET 10/MsQuic connector and server.
It maps reliable ordered frames to one persistent stream and latest-only frames
to independent streams with keyed `(producer epoch, sequence)` pending-state
replacement. Its
`System.Net.Quic` adapter fails closed for unreliable frames because that API
does not expose QUIC datagrams. A native MsQuic connector must own that mode and
the generic Unity runtime boundary. No RUDP implementation is installed
implicitly in its place.
**Derived state:** connector choice, physical endpoint, connection health, and
transport diagnostics. None of these grant provider authority or become content
@ -63,6 +70,11 @@ peer to the advertised provider identity. Adding TLS must not move discovery,
authorization, content verification, or application semantics into the TCP
adapter.
The QUIC adapter always uses TLS 1.3. Its certificate validation callback is
given the advertised endpoint identity so the consumer can bind transport
authentication to provider authorization. Test-only trust-all callbacks are not
a release configuration.
## Local parity evidence
On 2026-07-19, the parity harness moved a 13 MiB payload on loopback using

239
tests/GameCult.Mesh.Quic.Tests/CultMeshQuicRealtimeTransportTests.cs Normal file
View file

@ -0,0 +1,239 @@
using System.Net;
using System.Net.Quic;
using System.Security.Cryptography;
using System.Security.Cryptography.X509Certificates;
using FluentAssertions;
using NUnit.Framework;
namespace GameCult.Mesh.Quic.Tests;
[TestFixture]
public sealed class CultMeshQuicRealtimeTransportTests
{
[Test]
public async Task ConnectorNegotiatesAgainstAPlainMsQuicListener()
{
if (!QuicListener.IsSupported) Assert.Ignore("MsQuic is unavailable on this test host.");
var protocol = new System.Net.Security.SslApplicationProtocol("cultmesh-state-v1");
using var certificate = CreateCertificate();
await using var listener = await QuicListener.ListenAsync(new QuicListenerOptions
{
ListenEndPoint = new IPEndPoint(IPAddress.Loopback, 0),
ApplicationProtocols = new List<System.Net.Security.SslApplicationProtocol> { protocol },
ConnectionOptionsCallback = (_, _, _) => ValueTask.FromResult(new QuicServerConnectionOptions
{
DefaultCloseErrorCode = 0,
DefaultStreamErrorCode = 0,
ServerAuthenticationOptions = new System.Net.Security.SslServerAuthenticationOptions
{
ApplicationProtocols = new List<System.Net.Security.SslApplicationProtocol> { protocol },
EnabledSslProtocols = System.Security.Authentication.SslProtocols.Tls13,
ServerCertificate = certificate
}
})
});
var accept = listener.AcceptConnectionAsync().AsTask();
var connector = new CultMeshQuicRealtimeTransportConnector(new CultMeshQuicRealtimeConnectorOptions
{
ApplicationProtocol = protocol,
ValidateProviderCertificate = (_, _, _, _) => true
});
using var client = await connector.ConnectAsync(
new CultMeshTransportCandidate($"cultmesh-state+quic://127.0.0.1:{listener.LocalEndPoint.Port}"),
CultMeshEndpointId.Parse("service:aetheria.daemon"));
await using var serverConnection = await accept;
client.TransportId.Should().Be("msquic-realtime");
}
[Test]
public async Task ReliableOrderedFramesShareAnOrderedStreamInBothDirections()
{
if (!QuicListener.IsSupported) Assert.Ignore("MsQuic is unavailable on this test host.");
using var certificate = CreateCertificate();
await using var server = await CultMeshQuicRealtimeServer.ListenAsync(new CultMeshQuicRealtimeServerOptions
{
ListenEndPoint = new IPEndPoint(IPAddress.Loopback, 0),
ServerCertificate = certificate
});
var endpointId = CultMeshEndpointId.Parse("service:aetheria.daemon");
using var sessions = CreateSessions(server, out var validatedIdentity);
CultMeshRealtimeSession session;
try
{
session = await sessions.ConnectRealtimeAsync(endpointId);
}
catch when (server.BackgroundFailure.IsCompleted)
{
throw new InvalidOperationException("QUIC server accept loop failed.", await server.BackgroundFailure);
}
await WaitUntilAsync(() => server.ConnectionCount == 1);
await session.SendAsync(Frame(1, CultMeshRealtimeDelivery.ReliableOrdered));
await session.SendAsync(Frame(2, CultMeshRealtimeDelivery.ReliableOrdered));
var firstAtServer = await server.ReceiveAsync().WaitAsync(TimeSpan.FromSeconds(5));
var secondAtServer = await server.ReceiveAsync().WaitAsync(TimeSpan.FromSeconds(5));
await server.BroadcastAsync(Frame(3, CultMeshRealtimeDelivery.ReliableOrdered));
await server.BroadcastAsync(Frame(4, CultMeshRealtimeDelivery.ReliableOrdered));
var firstAtClient = await session.ReceiveAsync().WaitAsync(TimeSpan.FromSeconds(5));
var secondAtClient = await session.ReceiveAsync().WaitAsync(TimeSpan.FromSeconds(5));
new[] { firstAtServer.Sequence, secondAtServer.Sequence }.Should().Equal(1, 2);
new[] { firstAtClient.Sequence, secondAtClient.Sequence }.Should().Equal(3, 4);
firstAtClient.Payload.ToArray().Should().Equal(3, 4, 5);
validatedIdentity.Value.Should().Be(endpointId.Value);
}
[Test]
public async Task LatestOnlyCoalescesPendingStateAndDropsOlderGenerations()
{
if (!QuicListener.IsSupported) Assert.Ignore("MsQuic is unavailable on this test host.");
using var certificate = CreateCertificate();
await using var server = await CultMeshQuicRealtimeServer.ListenAsync(new CultMeshQuicRealtimeServerOptions
{
ListenEndPoint = new IPEndPoint(IPAddress.Loopback, 0),
ServerCertificate = certificate
});
var endpointId = CultMeshEndpointId.Parse("service:aetheria.daemon");
using var sessions = CreateSessions(server, out _);
var session = await sessions.ConnectRealtimeAsync(endpointId);
await WaitUntilAsync(() => server.ConnectionCount == 1);
await server.BroadcastAsync(Frame(100, CultMeshRealtimeDelivery.LatestOnly, producerEpoch: 7));
await server.BroadcastAsync(Frame(0, CultMeshRealtimeDelivery.LatestOnly, producerEpoch: 8));
await Task.Delay(100);
var latest = await session.ReceiveAsync().WaitAsync(TimeSpan.FromSeconds(5));
await server.BroadcastAsync(Frame(999, CultMeshRealtimeDelivery.LatestOnly, producerEpoch: 7));
using var staleDeadline = new CancellationTokenSource(TimeSpan.FromMilliseconds(300));
Func<Task> receiveStale = () => session.ReceiveAsync(staleDeadline.Token);
latest.ProducerEpoch.Should().Be(8);
latest.Sequence.Should().Be(0);
await receiveStale.Should().ThrowAsync<OperationCanceledException>(
"a late older generation must not twitch the consumer backwards");
}
[Test]
public async Task UnreliableDeliveryFailsClosedUntilANativeDatagramConnectorExists()
{
if (!QuicListener.IsSupported) Assert.Ignore("MsQuic is unavailable on this test host.");
using var certificate = CreateCertificate();
await using var server = await CultMeshQuicRealtimeServer.ListenAsync(new CultMeshQuicRealtimeServerOptions
{
ListenEndPoint = new IPEndPoint(IPAddress.Loopback, 0),
ServerCertificate = certificate
});
var endpointId = CultMeshEndpointId.Parse("service:aetheria.daemon");
using var sessions = CreateSessions(server, out _);
var session = await sessions.ConnectRealtimeAsync(endpointId);
Func<Task> send = () => session.SendAsync(Frame(1, CultMeshRealtimeDelivery.Unreliable));
await send.Should().ThrowAsync<NotSupportedException>().WithMessage("*datagrams*");
session.State.Status.Should().Be(CultMeshSessionStatus.Offline);
}
private static CultMeshSessionManager CreateSessions(
CultMeshQuicRealtimeServer server,
out IdentityObservation validatedIdentity)
{
validatedIdentity = new IdentityObservation();
var observation = validatedIdentity;
var endpoint = $"{CultMeshQuicRealtimeTransportConnector.Scheme}://127.0.0.1:{server.LocalEndPoint.Port}";
var discovery = new CultMeshDiscoveryService(new[] { new RouteSource(endpoint) });
var connector = new CultMeshQuicRealtimeTransportConnector(new CultMeshQuicRealtimeConnectorOptions
{
ValidateProviderCertificate = (identity, _, _, _) =>
{
observation.Value = identity.Value;
return true;
}
});
return new CultMeshSessionManager(
discovery,
Array.Empty<ICultMeshTransportConnector>(),
Array.Empty<ICultMeshContentTransportConnector>(),
new ICultMeshRealtimeTransportConnector[] { connector });
}
private static CultMeshRealtimeFrame Frame(
long sequence,
CultMeshRealtimeDelivery delivery,
long producerEpoch = 7) => new()
{
ChannelId = "aetheria.entities",
SchemaId = "eve.entity_soa.v1",
BodyId = "body:aetheria:entities",
ProducerEpoch = producerEpoch,
Sequence = sequence,
Delivery = delivery,
Payload = new byte[] { 3, 4, 5 }
};
private static X509Certificate2 CreateCertificate()
{
using var key = RSA.Create(2048);
var request = new CertificateRequest(
"CN=localhost",
key,
HashAlgorithmName.SHA256,
RSASignaturePadding.Pkcs1);
request.CertificateExtensions.Add(new X509BasicConstraintsExtension(false, false, 0, false));
request.CertificateExtensions.Add(new X509KeyUsageExtension(X509KeyUsageFlags.DigitalSignature, false));
request.CertificateExtensions.Add(new X509EnhancedKeyUsageExtension(
new OidCollection { new("1.3.6.1.5.5.7.3.1") },
false));
var names = new SubjectAlternativeNameBuilder();
names.AddDnsName("localhost");
names.AddIpAddress(IPAddress.Loopback);
request.CertificateExtensions.Add(names.Build());
request.CertificateExtensions.Add(new X509SubjectKeyIdentifierExtension(request.PublicKey, false));
using var certificate = request.CreateSelfSigned(
DateTimeOffset.UtcNow.AddMinutes(-1),
DateTimeOffset.UtcNow.AddHours(1));
return X509CertificateLoader.LoadPkcs12(certificate.Export(X509ContentType.Pfx), null);
}
private static async Task WaitUntilAsync(Func<bool> predicate)
{
var deadline = DateTime.UtcNow.AddSeconds(5);
while (!predicate())
{
if (DateTime.UtcNow >= deadline) throw new TimeoutException("Timed out waiting for QUIC connection state.");
await Task.Delay(10);
}
}
private sealed class RouteSource(string endpoint) : ICultMeshLookupSource
{
public string SourceId => "quic-test";
public Task<IReadOnlyList<CultMeshDiscoveryObservation>> LookupAsync(
CultMeshDiscoveryQuery query,
CancellationToken cancellationToken = default)
{
var now = DateTimeOffset.UtcNow;
IReadOnlyList<CultMeshDiscoveryObservation> result = new[]
{
new CultMeshDiscoveryObservation(
new CultMeshVerseDescriptor(
"aetheria",
"Aetheria",
CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v1", "test"),
new[] { endpoint }),
SourceId,
now,
now.AddMinutes(1),
CultMeshDiscoveryTrust.Signed)
};
return Task.FromResult(result);
}
}
private sealed class IdentityObservation
{
public string? Value { get; set; }
}
}

23
tests/GameCult.Mesh.Quic.Tests/GameCult.Mesh.Quic.Tests.csproj Normal file
View file

@ -0,0 +1,23 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<IsPackable>false</IsPackable>
<RootNamespace>GameCult.Mesh.Quic.Tests</RootNamespace>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<NoWarn>$(NoWarn);CA1416</NoWarn>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\src\GameCult.Mesh.Quic\GameCult.Mesh.Quic.csproj" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="coverlet.msbuild" Version="10.0.0">
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
<PrivateAssets>all</PrivateAssets>
</PackageReference>
<PackageReference Include="NUnit" Version="4.*" />
<PackageReference Include="NUnit3TestAdapter" Version="6.2.0" />
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="18.4.0" />
<PackageReference Include="FluentAssertions" Version="8.9.0" />
</ItemGroup>
</Project>