diff --git a/CultLib.sln b/CultLib.sln index f6ae25e9..7bd7ae93 100644 --- a/CultLib.sln +++ b/CultLib.sln @@ -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 diff --git a/src/GameCult.Mesh.Quic/CultMeshQuicRealtimeTransport.cs b/src/GameCult.Mesh.Quic/CultMeshQuicRealtimeTransport.cs new file mode 100644 index 00000000..81a66d9d --- /dev/null +++ b/src/GameCult.Mesh.Quic/CultMeshQuicRealtimeTransport.cs @@ -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; + +/// Configures identity-bound CultMesh QUIC client connections. +public sealed class CultMeshQuicRealtimeConnectorOptions +{ + /// Application protocol negotiated during the TLS handshake. + public SslApplicationProtocol ApplicationProtocol { get; set; } = CultMeshQuicRealtimeProtocol.ApplicationProtocol; + + /// Maximum time permitted for connection establishment. + public TimeSpan HandshakeTimeout { get; set; } = TimeSpan.FromSeconds(10); + + /// + /// Optional provider-aware certificate validator. Standard TLS validation is used when omitted. + /// + public Func? + ValidateProviderCertificate { get; set; } +} + +/// Creates MsQuic-backed realtime state connections for .NET runtimes. +public sealed class CultMeshQuicRealtimeTransportConnector : ICultMeshRealtimeTransportConnector +{ + /// Discovery scheme owned by this connector. + public const string Scheme = "cultmesh-state+quic"; + + private readonly CultMeshQuicRealtimeConnectorOptions _options; + + /// Creates a QUIC connector. + 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."); + } + + /// + public string ConnectorId => "msquic-realtime"; + + /// + public int Priority => 0; + + /// + public bool CanConnect(CultMeshTransportCandidate candidate) => + candidate != null && TryParseEndpoint(candidate.Endpoint, out _, out _); + + /// + public async Task 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 { _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; + } +} + +/// Configures an authenticated CultMesh QUIC realtime listener. +public sealed class CultMeshQuicRealtimeServerOptions +{ + /// Local endpoint to bind. + public IPEndPoint ListenEndPoint { get; set; } = new(IPAddress.Loopback, 0); + + /// Server certificate used by QUIC TLS 1.3. + public X509Certificate2 ServerCertificate { get; set; } = null!; + + /// Application protocol negotiated during the TLS handshake. + public SslApplicationProtocol ApplicationProtocol { get; set; } = CultMeshQuicRealtimeProtocol.ApplicationProtocol; + + /// Maximum time permitted for connection establishment. + public TimeSpan HandshakeTimeout { get; set; } = TimeSpan.FromSeconds(10); +} + +/// +/// Accepts .NET QUIC realtime sessions. It owns physical connections only; provider +/// authorization and the meaning of frames remain outside this adapter. +/// +public sealed class CultMeshQuicRealtimeServer : IAsyncDisposable +{ + private readonly QuicListener _listener; + private readonly CancellationTokenSource _shutdown = new(); + private readonly ConcurrentDictionary _clients = new(); + private readonly CultMeshRealtimeInbox _received = new(); + private readonly TaskCompletionSource _backgroundFailure = new( + TaskCreationOptions.RunContinuationsAsynchronously); + private readonly Task _acceptLoop; + private bool _disposed; + + private CultMeshQuicRealtimeServer(QuicListener listener) + { + _listener = listener; + _acceptLoop = AcceptLoopAsync(_shutdown.Token); + } + + /// Bound listener endpoint. + public IPEndPoint LocalEndPoint => (IPEndPoint)_listener.LocalEndPoint; + + /// Number of active physical client connections. + public int ConnectionCount => _clients.Count; + + /// Completes if the listener's background accept loop fails. + public Task BackgroundFailure => _backgroundFailure.Task; + + /// Creates and starts an authenticated QUIC listener. + public static async Task 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 { options.ApplicationProtocol }, + ConnectionOptionsCallback = (_, _, _) => ValueTask.FromResult(new QuicServerConnectionOptions + { + DefaultCloseErrorCode = CultMeshQuicRealtimeProtocol.ConnectionCloseCode, + DefaultStreamErrorCode = CultMeshQuicRealtimeProtocol.StreamAbortCode, + MaxInboundUnidirectionalStreams = 1024, + HandshakeTimeout = options.HandshakeTimeout, + ServerAuthenticationOptions = new SslServerAuthenticationOptions + { + ApplicationProtocols = new List { options.ApplicationProtocol }, + EnabledSslProtocols = SslProtocols.Tls13, + ServerCertificate = options.ServerCertificate + } + }) + }, cancellationToken).ConfigureAwait(false); + return new CultMeshQuicRealtimeServer(listener); + } + + /// Broadcasts one state frame to the currently connected clients. + 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); + } + + /// Receives the next client-originated state frame. + public Task ReceiveAsync(CancellationToken cancellationToken = default) => + _received.ReceiveAsync(cancellationToken); + + /// + 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 _latestGenerations = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _streamReaders = new(); + private readonly SemaphoreSlim _reliableSendGate = new(1, 1); + private readonly Action? _observer; + private readonly Task _acceptLoop; + private QuicStream? _reliableOutbound; + private int _nextReaderId; + private bool _disposed; + + public CultMeshQuicRealtimeTransport( + string endpoint, + QuicConnection connection, + Action? 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 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 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 + { + 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 _ready = Channel.CreateUnbounded( + new UnboundedChannelOptions { SingleReader = false, SingleWriter = false }); + private readonly Dictionary _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 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 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 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 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; + } + } +} diff --git a/src/GameCult.Mesh.Quic/GameCult.Mesh.Quic.csproj b/src/GameCult.Mesh.Quic/GameCult.Mesh.Quic.csproj new file mode 100644 index 00000000..2685220a --- /dev/null +++ b/src/GameCult.Mesh.Quic/GameCult.Mesh.Quic.csproj @@ -0,0 +1,16 @@ + + + net10.0 + enable + enable + MsQuic-backed realtime state transport for CultMesh + GameCult.Mesh.Quic + game;distributed;realtime;mesh;quic;msquic + README.md + $(NoWarn);CA1416 + + + + + + diff --git a/src/GameCult.Mesh.Quic/README.md b/src/GameCult.Mesh.Quic/README.md new file mode 100644 index 00000000..e0745154 --- /dev/null +++ b/src/GameCult.Mesh.Quic/README.md @@ -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. diff --git a/src/GameCult.Mesh/docs/transport-planes.md b/src/GameCult.Mesh/docs/transport-planes.md index 879bd5c6..0a86ccab 100644 --- a/src/GameCult.Mesh/docs/transport-planes.md +++ b/src/GameCult.Mesh/docs/transport-planes.md @@ -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 diff --git a/tests/GameCult.Mesh.Quic.Tests/CultMeshQuicRealtimeTransportTests.cs b/tests/GameCult.Mesh.Quic.Tests/CultMeshQuicRealtimeTransportTests.cs new file mode 100644 index 00000000..56fbae5e --- /dev/null +++ b/tests/GameCult.Mesh.Quic.Tests/CultMeshQuicRealtimeTransportTests.cs @@ -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 { protocol }, + ConnectionOptionsCallback = (_, _, _) => ValueTask.FromResult(new QuicServerConnectionOptions + { + DefaultCloseErrorCode = 0, + DefaultStreamErrorCode = 0, + ServerAuthenticationOptions = new System.Net.Security.SslServerAuthenticationOptions + { + ApplicationProtocols = new List { 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 receiveStale = () => session.ReceiveAsync(staleDeadline.Token); + + latest.ProducerEpoch.Should().Be(8); + latest.Sequence.Should().Be(0); + await receiveStale.Should().ThrowAsync( + "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 send = () => session.SendAsync(Frame(1, CultMeshRealtimeDelivery.Unreliable)); + + await send.Should().ThrowAsync().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(), + Array.Empty(), + 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 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> LookupAsync( + CultMeshDiscoveryQuery query, + CancellationToken cancellationToken = default) + { + var now = DateTimeOffset.UtcNow; + IReadOnlyList 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; } + } +} diff --git a/tests/GameCult.Mesh.Quic.Tests/GameCult.Mesh.Quic.Tests.csproj b/tests/GameCult.Mesh.Quic.Tests/GameCult.Mesh.Quic.Tests.csproj new file mode 100644 index 00000000..44a808f3 --- /dev/null +++ b/tests/GameCult.Mesh.Quic.Tests/GameCult.Mesh.Quic.Tests.csproj @@ -0,0 +1,23 @@ + + + net10.0 + false + GameCult.Mesh.Quic.Tests + enable + enable + $(NoWarn);CA1416 + + + + + + + runtime; build; native; contentfiles; analyzers; buildtransitive + all + + + + + + +