diff --git a/src/GameCult.Mesh/CultMeshClient.cs b/src/GameCult.Mesh/CultMeshClient.cs index 2be64288..12ccd470 100644 --- a/src/GameCult.Mesh/CultMeshClient.cs +++ b/src/GameCult.Mesh/CultMeshClient.cs @@ -22,9 +22,22 @@ namespace GameCult.Mesh /// Gets or sets session path and diagnostic policy. public CultMeshSessionManagerOptions Sessions { get; set; } = new(); - /// Gets or sets optional transport connectors. The CultNet schema connector is used by default. + /// Gets or sets schema connectors. Length-prefixed TCP is used by default. public IReadOnlyList? Connectors { get; set; } + /// + /// Gets or sets streaming content connectors. TCP content delivery is used by default; + /// applications may replace it or explicitly add legacy RUDP. + /// + public IReadOnlyList? ContentConnectors { get; set; } + + /// + /// Gets or sets realtime state connectors. No connector is installed implicitly; + /// promoted runtimes should register QUIC and may explicitly add a legacy fallback. + /// + public IReadOnlyList RealtimeConnectors { get; set; } = + Array.Empty(); + /// Gets or sets the deadline before an unanswered subscription intent is replayed. public TimeSpan SubscriptionResponseTimeout { get; set; } = TimeSpan.FromSeconds(2); } @@ -68,8 +81,13 @@ namespace GameCult.Mesh _discovery, options.Connectors ?? new ICultMeshTransportConnector[] { - new CultMeshSchemaTransportConnector(clock: options.Sessions.Clock) + new CultMeshTcpSchemaTransportConnector(clock: options.Sessions.Clock) }, + options.ContentConnectors ?? new ICultMeshContentTransportConnector[] + { + new CultMeshTcpContentTransportConnector() + }, + options.RealtimeConnectors, options.Sessions); _subscriptionResponseTimeout = options.SubscriptionResponseTimeout; } @@ -84,6 +102,15 @@ namespace GameCult.Mesh return _sessions.ConnectAsync(CultMeshEndpointId.Parse(endpointId), protocol, cancellationToken); } + /// Connects the reusable realtime state plane by stable endpoint identity. + public Task ConnectRealtimeAsync( + string endpointId, + CancellationToken cancellationToken = default) + { + ThrowIfDisposed(); + return _sessions.ConnectRealtimeAsync(CultMeshEndpointId.Parse(endpointId), cancellationToken); + } + /// Connects to a stable endpoint identity using one application protocol. public Task ConnectAsync( CultMeshEndpointId endpointId, @@ -97,15 +124,13 @@ namespace GameCult.Mesh /// Creates a verified-transfer provider over this client's reusable content session owner. public CultMeshSessionContentProvider ContentProvider( string providerId, - string endpointId, - CultMeshSessionContentProviderOptions? options = null) + string endpointId) { ThrowIfDisposed(); return new CultMeshSessionContentProvider( providerId, _sessions, - CultMeshEndpointId.Parse(endpointId), - options); + CultMeshEndpointId.Parse(endpointId)); } /// Creates a direct body provider over this client's reusable negotiated session owner. diff --git a/src/GameCult.Mesh/CultMeshContentSessions.cs b/src/GameCult.Mesh/CultMeshContentSessions.cs index 99687358..51570c54 100644 --- a/src/GameCult.Mesh/CultMeshContentSessions.cs +++ b/src/GameCult.Mesh/CultMeshContentSessions.cs @@ -11,7 +11,7 @@ namespace GameCult.Mesh /// Serves bounded, content-addressed CDN chunks over the CultMesh content protocol. /// Manifests and descriptors remain typed document state; payload bytes do not. /// - public sealed class CultMeshContentServer : IDisposable + public sealed class CultMeshLegacyRudpContentServer : IDisposable { private readonly ICultNetSchemaServer _server; private readonly CultCache _content; @@ -20,7 +20,7 @@ namespace GameCult.Mesh private bool _disposed; /// Attaches content serving to an existing schema host and provider cache. - public CultMeshContentServer( + public CultMeshLegacyRudpContentServer( ICultNetSchemaServer server, CultCache content, Func? canServeHash = null) @@ -85,8 +85,8 @@ namespace GameCult.Mesh } } - /// Configures bounded response waiting for session-backed content requests. - public sealed class CultMeshSessionContentProviderOptions + /// Configures the explicit compatibility path for content over schema-message RUDP. + public sealed class CultMeshLegacyRudpContentTransportOptions { /// Gets or sets the injected reliability clock. public ICultMeshClock Clock { get; set; } = CultMeshSystemClock.Instance; @@ -95,97 +95,148 @@ namespace GameCult.Mesh } /// - /// Retrieves verified CDN chunks through one reusable identity-first CultMesh session. - /// Transfer verification and final cache promotion remain owned by CultMeshContentTransferService. + /// Explicit low-priority compatibility connector for the retired content-over-RUDP path. + /// It is never installed by default. /// - public sealed class CultMeshSessionContentProvider : ICultMeshContentProvider + public sealed class CultMeshLegacyRudpContentTransportConnector : ICultMeshContentTransportConnector { - private readonly CultMeshSessionManager _sessions; - private readonly CultMeshEndpointId _endpointId; - private readonly CultMeshSessionContentProviderOptions _options; + public const int LegacyPriority = 10_000; + private readonly CultMeshSchemaTransportConnector _schemaConnector; + private readonly CultMeshLegacyRudpContentTransportOptions _options; - /// Creates a content provider backed by a reusable identity-first session. - public CultMeshSessionContentProvider( - string providerId, - CultMeshSessionManager sessions, - CultMeshEndpointId endpointId, - CultMeshSessionContentProviderOptions? options = null) + public CultMeshLegacyRudpContentTransportConnector( + CultMeshLegacyRudpContentTransportOptions? options = null, + Func? createClient = null) { - ProviderId = string.IsNullOrWhiteSpace(providerId) - ? throw new ArgumentException("Provider identity is required.", nameof(providerId)) - : providerId; - _sessions = sessions ?? throw new ArgumentNullException(nameof(sessions)); - _endpointId = endpointId ?? throw new ArgumentNullException(nameof(endpointId)); - _options = options ?? new CultMeshSessionContentProviderOptions(); + _options = options ?? new CultMeshLegacyRudpContentTransportOptions(); if (_options.ResponseTimeout <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(options), "Content response timeout must be positive."); + _schemaConnector = new CultMeshSchemaTransportConnector(createClient, _options.Clock, _options.ResponseTimeout); } - /// - public string ProviderId { get; } + public string ConnectorId => "legacy-rudp-content"; + public int Priority => LegacyPriority; + public bool CanConnect(CultMeshTransportCandidate candidate) => + Uri.TryCreate(candidate.Endpoint, UriKind.Absolute, out var uri) && + string.Equals(uri.Scheme, "rudp", StringComparison.OrdinalIgnoreCase); - /// - public async Task GetChunkAsync( - CultMeshCdnChunkRef chunk, + public async Task ConnectAsync( + CultMeshTransportCandidate candidate, + CultMeshEndpointId endpointId, CancellationToken cancellationToken = default) { - if (chunk == null) throw new ArgumentNullException(nameof(chunk)); - var hash = CultMeshCdn.NormalizeHash(chunk.ChunkHash, nameof(chunk.ChunkHash)); - var session = await _sessions.ConnectAsync( - _endpointId, - CultMeshProtocols.Content, - cancellationToken).ConfigureAwait(false); - var messageId = Guid.NewGuid().ToString("N"); - var completion = new TaskCompletionSource( - TaskCreationOptions.RunContinuationsAsynchronously); - using var responseSubscription = session.OnCultNet(response => - { - if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) - completion.TrySetResult(response); - }); - using var errorSubscription = session.OnCultNet(error => - completion.TrySetException(new IOException(error.Error))); - session.SendCultNet(new CultMeshContentChunkRequestMessage - { - MessageId = messageId, - ChunkHash = hash, - RecordKey = string.IsNullOrWhiteSpace(chunk.RecordKey) - ? CultMeshCdnArtifactChunk.CreateRecordKey(hash).Value - : chunk.RecordKey, - ExpectedSizeBytes = chunk.SizeBytes - }); - - var response = await WaitForResponseAsync(completion.Task, cancellationToken).ConfigureAwait(false); - if (!response.Found) - throw new FileNotFoundException( - "Content provider '" + ProviderId + "' rejected chunk '" + hash + "': " + response.Error, - chunk.RecordKey); - var resolved = new CultMeshCdnArtifactChunk - { - ChunkHash = response.ChunkHash, - SizeBytes = response.SizeBytes, - Payload = response.Payload ?? Array.Empty() - }; - CultMeshCdn.ValidateChunkPayload(chunk, resolved); - return resolved; + if (!CanConnect(candidate)) + throw new NotSupportedException($"Legacy RUDP content connector does not support '{candidate.Endpoint}'."); + var client = await _schemaConnector.ConnectAsync(candidate, CultMeshProtocols.Content, cancellationToken) + .ConfigureAwait(false); + return new LegacyRudpContentTransport(candidate.Endpoint, client, _options); } - private async Task WaitForResponseAsync( - Task response, - CancellationToken cancellationToken) + private sealed class LegacyRudpContentTransport : ICultMeshContentTransport { - using var deadlineCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - var deadline = _options.Clock.DelayAsync(_options.ResponseTimeout, deadlineCancellation.Token); - var completed = await Task.WhenAny(response, deadline).ConfigureAwait(false); - if (completed == response) + private readonly ICultNetSchemaClient _client; + private readonly CultMeshLegacyRudpContentTransportOptions _options; + private readonly System.Collections.Concurrent.ConcurrentDictionary> _pending = new(StringComparer.Ordinal); + private bool _disposed; + + public LegacyRudpContentTransport( + string endpoint, + ICultNetSchemaClient client, + CultMeshLegacyRudpContentTransportOptions options) { - deadlineCancellation.Cancel(); - return await response.ConfigureAwait(false); + Endpoint = endpoint; + _client = client; + _options = options; + _client.OnCultNet(OnResponse); + _client.OnCultNet(OnError); + } + + public string TransportId => "legacy-rudp-content"; + public string Endpoint { get; } + + public async Task CopyChunkToAsync( + CultMeshCdnChunkRef chunk, + Stream destination, + CancellationToken cancellationToken = default) + { + if (_disposed) throw new ObjectDisposedException(nameof(LegacyRudpContentTransport)); + if (chunk == null) throw new ArgumentNullException(nameof(chunk)); + if (destination == null) throw new ArgumentNullException(nameof(destination)); + var hash = CultMeshCdn.NormalizeHash(chunk.ChunkHash, nameof(chunk.ChunkHash)); + var messageId = Guid.NewGuid().ToString("N"); + var completion = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + if (!_pending.TryAdd(messageId, completion)) + throw new InvalidOperationException("Duplicate legacy content request identity."); + try + { + _client.SendCultNet(new CultMeshContentChunkRequestMessage + { + MessageId = messageId, + ChunkHash = hash, + RecordKey = string.IsNullOrWhiteSpace(chunk.RecordKey) + ? CultMeshCdnArtifactChunk.CreateRecordKey(hash).Value + : chunk.RecordKey, + ExpectedSizeBytes = chunk.SizeBytes + }); + var response = await WaitForResponseAsync(completion.Task, cancellationToken).ConfigureAwait(false); + if (!response.Found) + throw new FileNotFoundException( + "Legacy RUDP content provider rejected chunk '" + hash + "': " + response.Error, + chunk.RecordKey); + var resolved = new CultMeshCdnArtifactChunk + { + ChunkHash = response.ChunkHash, + SizeBytes = response.SizeBytes, + Payload = response.Payload ?? Array.Empty() + }; + CultMeshCdn.ValidateChunkPayload(chunk, resolved); + await destination.WriteAsync(resolved.Payload, 0, resolved.Payload.Length, cancellationToken) + .ConfigureAwait(false); + } + finally + { + _pending.TryRemove(messageId, out _); + } + } + + public void Dispose() + { + if (_disposed) return; + _disposed = true; + foreach (var completion in _pending.Values) + completion.TrySetException(new ObjectDisposedException(nameof(LegacyRudpContentTransport))); + _pending.Clear(); + _client.Dispose(); + } + + private void OnResponse(CultMeshContentChunkResponseMessage response) + { + if (_pending.TryGetValue(response.MessageId ?? string.Empty, out var completion)) + completion.TrySetResult(response); + } + + private void OnError(CultNetErrorMessage error) + { + var exception = new IOException(error.Error); + foreach (var completion in _pending.Values) completion.TrySetException(exception); + } + + private async Task WaitForResponseAsync( + Task response, + CancellationToken cancellationToken) + { + using var deadlineCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + var deadline = _options.Clock.DelayAsync(_options.ResponseTimeout, deadlineCancellation.Token); + var completed = await Task.WhenAny(response, deadline).ConfigureAwait(false); + if (completed == response) + { + deadlineCancellation.Cancel(); + return await response.ConfigureAwait(false); + } + cancellationToken.ThrowIfCancellationRequested(); + throw new TimeoutException("Timed out fetching content through the legacy RUDP connector."); } - cancellationToken.ThrowIfCancellationRequested(); - throw new TimeoutException( - "Timed out fetching content from provider '" + ProviderId + "' through endpoint identity '" + - _endpointId.Value + "'."); } } } diff --git a/src/GameCult.Mesh/CultMeshContentTransfer.cs b/src/GameCult.Mesh/CultMeshContentTransfer.cs index 59f89a15..f6a426b8 100644 --- a/src/GameCult.Mesh/CultMeshContentTransfer.cs +++ b/src/GameCult.Mesh/CultMeshContentTransfer.cs @@ -32,7 +32,10 @@ namespace GameCult.Mesh public interface ICultMeshContentProvider { string ProviderId { get; } - Task GetChunkAsync(CultMeshCdnChunkRef chunk, CancellationToken cancellationToken = default); + Task CopyChunkToAsync( + CultMeshCdnChunkRef chunk, + Stream destination, + CancellationToken cancellationToken = default); } public sealed class CultMeshDatabaseContentProvider : ICultMeshContentProvider @@ -47,15 +50,23 @@ namespace GameCult.Mesh public string ProviderId { get; } - public async Task GetChunkAsync(CultMeshCdnChunkRef chunk, CancellationToken cancellationToken = default) + public async Task CopyChunkToAsync( + CultMeshCdnChunkRef chunk, + Stream destination, + CancellationToken cancellationToken = default) { + if (chunk == null) throw new ArgumentNullException(nameof(chunk)); + if (destination == null) throw new ArgumentNullException(nameof(destination)); cancellationToken.ThrowIfCancellationRequested(); var key = string.IsNullOrWhiteSpace(chunk.RecordKey) ? CultMeshCdnArtifactChunk.CreateRecordKey(chunk.ChunkHash) : new CultRecordKey(chunk.RecordKey); var result = await _database.GetAsync(key).ConfigureAwait(false); cancellationToken.ThrowIfCancellationRequested(); - return result; + if (result == null) + throw new FileNotFoundException("CDN artifact chunk is missing from the provider.", key.Value); + CultMeshCdn.ValidateChunkPayload(chunk, result); + await destination.WriteAsync(result.Payload, 0, result.Payload.Length, cancellationToken).ConfigureAwait(false); } } @@ -218,44 +229,51 @@ namespace GameCult.Mesh var state = _stateCache.Get(stateKey); var verified = RestoreVerifiedState(state, manifest, fingerprint, partialPath); - using (var stream = new FileStream(partialPath, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.Read, 81920, FileOptions.Asynchronous)) + using (var stream = new FileStream( + partialPath, + FileMode.OpenOrCreate, + FileAccess.ReadWrite, + FileShare.ReadWrite, + 81920, + FileOptions.Asynchronous)) { stream.SetLength(manifest.SizeBytes); - var ordered = manifest.Chunks - .Select((chunk, index) => new { Chunk = chunk, Index = index }) - .OrderBy(item => item.Chunk.Offset) - .ToArray(); + await stream.FlushAsync(cancellationToken).ConfigureAwait(false); + } - var remaining = ordered.Where(item => !verified.Contains(item.Index)).ToArray(); - for (var windowStart = 0; windowStart < remaining.Length; windowStart += _maxConcurrentChunkRequests) + var ordered = manifest.Chunks + .Select((chunk, index) => new { Chunk = chunk, Index = index }) + .OrderBy(item => item.Chunk.Offset) + .ToArray(); + var remaining = ordered.Where(item => !verified.Contains(item.Index)).ToArray(); + for (var windowStart = 0; windowStart < remaining.Length; windowStart += _maxConcurrentChunkRequests) + { + cancellationToken.ThrowIfCancellationRequested(); + var window = remaining + .Skip(windowStart) + .Take(_maxConcurrentChunkRequests) + .Select(item => (Item: item, Copy: CopyVerifiedChunkToPartialAsync( + item.Chunk, + partialPath, + cancellationToken))) + .ToArray(); + try { - cancellationToken.ThrowIfCancellationRequested(); - var window = remaining - .Skip(windowStart) - .Take(_maxConcurrentChunkRequests) - .Select(item => (Item: item, Fetch: FetchVerifiedChunkAsync(item.Chunk, cancellationToken))) - .ToArray(); - try + foreach (var pending in window) { - foreach (var pending in window) - { - var chunk = await pending.Fetch.ConfigureAwait(false); - stream.Position = pending.Item.Chunk.Offset; - await stream.WriteAsync(chunk.Payload, 0, chunk.Payload.Length, cancellationToken).ConfigureAwait(false); - await stream.FlushAsync(cancellationToken).ConfigureAwait(false); - verified.Add(pending.Item.Index); - await SaveStateAsync(contentHash, fingerprint, manifest.SizeBytes, verified, stateKey).ConfigureAwait(false); - } + await pending.Copy.ConfigureAwait(false); + verified.Add(pending.Item.Index); + await SaveStateAsync(contentHash, fingerprint, manifest.SizeBytes, verified, stateKey).ConfigureAwait(false); } - catch + } + catch + { + foreach (var pending in window) { - foreach (var pending in window) - { - try { await pending.Fetch.ConfigureAwait(false); } - catch { } - } - throw; + try { await pending.Copy.ConfigureAwait(false); } + catch { } } + throw; } } @@ -307,8 +325,9 @@ namespace GameCult.Mesh return verified; } - private async Task FetchVerifiedChunkAsync( + private async Task CopyVerifiedChunkToPartialAsync( CultMeshCdnChunkRef reference, + string partialPath, CancellationToken cancellationToken) { var failures = new List(); @@ -317,10 +336,22 @@ namespace GameCult.Mesh cancellationToken.ThrowIfCancellationRequested(); try { - var chunk = await provider.GetChunkAsync(reference, cancellationToken).ConfigureAwait(false) - ?? throw new FileNotFoundException("CDN artifact chunk is missing from the provider.", reference.RecordKey); - CultMeshCdn.ValidateChunkPayload(reference, chunk); - return chunk; + using var partial = new FileStream( + partialPath, + FileMode.Open, + FileAccess.Write, + FileShare.ReadWrite, + 81920, + FileOptions.Asynchronous | FileOptions.RandomAccess); + partial.Position = reference.Offset; + using var destination = new BoundedHashingWriteStream(partial, reference.SizeBytes); + await provider.CopyChunkToAsync(reference, destination, cancellationToken).ConfigureAwait(false); + await destination.FlushAsync(cancellationToken).ConfigureAwait(false); + var actualHash = destination.CompleteHash(); + var expectedHash = CultMeshCdn.NormalizeHash(reference.ChunkHash, nameof(reference.ChunkHash)); + if (!string.Equals(actualHash, expectedHash, StringComparison.Ordinal)) + throw new InvalidDataException("CDN artifact chunk hash does not match its manifest."); + return; } catch (Exception error) when (!(error is OperationCanceledException && cancellationToken.IsCancellationRequested)) { @@ -331,6 +362,89 @@ namespace GameCult.Mesh throw new AggregateException("No content provider supplied a valid CDN chunk.", failures); } + private sealed class BoundedHashingWriteStream : Stream + { + private readonly Stream _destination; + private readonly long _expectedBytes; + private readonly SHA256 _sha = SHA256.Create(); + private long _written; + private bool _completed; + + public BoundedHashingWriteStream(Stream destination, long expectedBytes) + { + _destination = destination ?? throw new ArgumentNullException(nameof(destination)); + if (!destination.CanWrite) throw new ArgumentException("Content destination must be writable.", nameof(destination)); + if (expectedBytes < 0) throw new ArgumentOutOfRangeException(nameof(expectedBytes)); + _expectedBytes = expectedBytes; + } + + public override bool CanRead => false; + public override bool CanSeek => false; + public override bool CanWrite => true; + public override long Length => throw new NotSupportedException(); + public override long Position + { + get => _written; + set => throw new NotSupportedException(); + } + + public override void Flush() => _destination.Flush(); + public override Task FlushAsync(CancellationToken cancellationToken) => + _destination.FlushAsync(cancellationToken); + + public override void Write(byte[] buffer, int offset, int count) + { + ValidateWrite(buffer, offset, count); + _destination.Write(buffer, offset, count); + _sha.TransformBlock(buffer, offset, count, null, 0); + _written += count; + } + + public override async Task WriteAsync( + byte[] buffer, + int offset, + int count, + CancellationToken cancellationToken) + { + ValidateWrite(buffer, offset, count); + await _destination.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false); + _sha.TransformBlock(buffer, offset, count, null, 0); + _written += count; + } + + public string CompleteHash() + { + if (_completed) throw new InvalidOperationException("Content destination hash is already complete."); + if (_written != _expectedBytes) + throw new EndOfStreamException( + $"Content transport wrote {_written} bytes; expected {_expectedBytes} bytes."); + _sha.TransformFinalBlock(Array.Empty(), 0, 0); + _completed = true; + return string.Concat(_sha.Hash!.Select(value => value.ToString("x2", CultureInfo.InvariantCulture))); + } + + protected override void Dispose(bool disposing) + { + if (disposing) _sha.Dispose(); + base.Dispose(disposing); + } + + public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + + private void ValidateWrite(byte[] buffer, int offset, int count) + { + if (_completed) throw new InvalidOperationException("Content destination is already complete."); + if (buffer == null) throw new ArgumentNullException(nameof(buffer)); + if (offset < 0 || count < 0 || offset > buffer.Length - count) + throw new ArgumentOutOfRangeException(nameof(offset)); + if (_written + count > _expectedBytes) + throw new InvalidDataException( + $"Content transport exceeded its advertised {_expectedBytes}-byte chunk boundary."); + } + } + private async Task SaveStateAsync( string contentHash, string fingerprint, diff --git a/src/GameCult.Mesh/CultMeshContentTransports.cs b/src/GameCult.Mesh/CultMeshContentTransports.cs new file mode 100644 index 00000000..f1814b2c --- /dev/null +++ b/src/GameCult.Mesh/CultMeshContentTransports.cs @@ -0,0 +1,166 @@ +using System; +using System.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace GameCult.Mesh +{ + /// + /// Moves immutable content bytes for one selected physical path. Content identity, + /// verification, checkpoints, and promotion remain owned by + /// . + /// + public interface ICultMeshContentTransport : IDisposable + { + string TransportId { get; } + string Endpoint { get; } + + /// + /// Writes exactly one advertised chunk into the caller-owned destination. + /// The transport must honor destination backpressure and must not close it. + /// + Task CopyChunkToAsync( + CultMeshCdnChunkRef chunk, + Stream destination, + CancellationToken cancellationToken = default); + } + + /// + /// Creates physical content transports. Lower priority values are attempted first; + /// a fallback tier is not raced against a preferred tier. + /// + public interface ICultMeshContentTransportConnector + { + string ConnectorId { get; } + int Priority { get; } + bool CanConnect(CultMeshTransportCandidate candidate); + Task ConnectAsync( + CultMeshTransportCandidate candidate, + CultMeshEndpointId endpointId, + CancellationToken cancellationToken = default); + } + + /// + /// Reusable identity-bound content session selected and owned by + /// . + /// + public sealed class CultMeshContentSession : IDisposable + { + private readonly object _gate = new(); + private readonly Action _onTransportFailure; + private ICultMeshContentTransport _transport; + private bool _disposed; + + internal CultMeshContentSession( + CultMeshEndpointId endpointId, + ICultMeshContentTransport transport, + CultMeshSessionState state, + Action onTransportFailure) + { + EndpointId = endpointId ?? throw new ArgumentNullException(nameof(endpointId)); + _transport = transport ?? throw new ArgumentNullException(nameof(transport)); + State = state ?? throw new ArgumentNullException(nameof(state)); + _onTransportFailure = onTransportFailure ?? throw new ArgumentNullException(nameof(onTransportFailure)); + } + + public CultMeshEndpointId EndpointId { get; } + public CultMeshSessionState State { get; private set; } + public string TransportId { get { lock (_gate) return _transport.TransportId; } } + + public async Task CopyChunkToAsync( + CultMeshCdnChunkRef chunk, + Stream destination, + CancellationToken cancellationToken = default) + { + if (chunk == null) throw new ArgumentNullException(nameof(chunk)); + if (destination == null) throw new ArgumentNullException(nameof(destination)); + ICultMeshContentTransport transport; + lock (_gate) + { + ThrowIfDisposed(); + transport = _transport; + } + try + { + await transport.CopyChunkToAsync(chunk, destination, cancellationToken).ConfigureAwait(false); + } + catch (Exception) when (!cancellationToken.IsCancellationRequested) + { + _onTransportFailure(this); + throw; + } + } + + internal void Replace(ICultMeshContentTransport transport, CultMeshSessionState state) + { + if (transport == null) throw new ArgumentNullException(nameof(transport)); + ICultMeshContentTransport previous; + lock (_gate) + { + ThrowIfDisposed(); + previous = _transport; + _transport = transport; + State = state; + } + previous.Dispose(); + } + + internal void MarkOffline(CultMeshSessionState state) + { + lock (_gate) + { + if (_disposed) return; + State = state; + } + } + + public void Dispose() + { + ICultMeshContentTransport transport; + lock (_gate) + { + if (_disposed) return; + _disposed = true; + transport = _transport; + } + transport.Dispose(); + } + + private void ThrowIfDisposed() + { + if (_disposed) throw new ObjectDisposedException(nameof(CultMeshContentSession)); + } + } + + /// + /// Adapts a selected streaming transport to the verified content-transfer source port. + /// + public sealed class CultMeshSessionContentProvider : ICultMeshContentProvider + { + private readonly CultMeshSessionManager _sessions; + private readonly CultMeshEndpointId _endpointId; + + public CultMeshSessionContentProvider( + string providerId, + CultMeshSessionManager sessions, + CultMeshEndpointId endpointId) + { + ProviderId = string.IsNullOrWhiteSpace(providerId) + ? throw new ArgumentException("Provider identity is required.", nameof(providerId)) + : providerId; + _sessions = sessions ?? throw new ArgumentNullException(nameof(sessions)); + _endpointId = endpointId ?? throw new ArgumentNullException(nameof(endpointId)); + } + + public string ProviderId { get; } + + public async Task CopyChunkToAsync( + CultMeshCdnChunkRef chunk, + Stream destination, + CancellationToken cancellationToken = default) + { + var session = await _sessions.ConnectContentAsync(_endpointId, cancellationToken).ConfigureAwait(false); + await session.CopyChunkToAsync(chunk, destination, cancellationToken).ConfigureAwait(false); + } + } +} diff --git a/src/GameCult.Mesh/CultMeshRealtimeTransports.cs b/src/GameCult.Mesh/CultMeshRealtimeTransports.cs new file mode 100644 index 00000000..d087b304 --- /dev/null +++ b/src/GameCult.Mesh/CultMeshRealtimeTransports.cs @@ -0,0 +1,154 @@ +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace GameCult.Mesh +{ + /// Delivery semantics requested from a realtime state transport. + public enum CultMeshRealtimeDelivery + { + ReliableOrdered, + LatestOnly, + Unreliable + } + + /// + /// Transport-neutral realtime state frame. Schema identity and logical generation + /// remain stable when the physical connector changes. + /// + public sealed class CultMeshRealtimeFrame + { + public string ChannelId { get; set; } = string.Empty; + public string SchemaId { get; set; } = string.Empty; + public string BodyId { get; set; } = string.Empty; + public long ProducerEpoch { get; set; } + public long Sequence { get; set; } + public CultMeshRealtimeDelivery Delivery { get; set; } + public ReadOnlyMemory Payload { get; set; } + + internal void Validate() + { + if (string.IsNullOrWhiteSpace(ChannelId)) + throw new InvalidOperationException("Realtime channel identity is required."); + if (string.IsNullOrWhiteSpace(SchemaId)) + throw new InvalidOperationException("Realtime schema identity is required."); + if (string.IsNullOrWhiteSpace(BodyId)) + throw new InvalidOperationException("Realtime body identity is required."); + if (ProducerEpoch < 0 || Sequence < 0) + throw new InvalidOperationException("Realtime epoch and sequence must be non-negative."); + } + } + + /// + /// Bidirectional realtime state plane. Commands, receipts, manifests, and other + /// registered schema documents do not travel through this port. + /// + public interface ICultMeshRealtimeTransport : IDisposable + { + string TransportId { get; } + string Endpoint { get; } + Task SendAsync(CultMeshRealtimeFrame frame, CancellationToken cancellationToken = default); + Task ReceiveAsync(CancellationToken cancellationToken = default); + } + + /// Creates a realtime state transport, normally QUIC streams/datagrams. + public interface ICultMeshRealtimeTransportConnector + { + string ConnectorId { get; } + int Priority { get; } + bool CanConnect(CultMeshTransportCandidate candidate); + Task ConnectAsync( + CultMeshTransportCandidate candidate, + CultMeshEndpointId endpointId, + CancellationToken cancellationToken = default); + } + + /// Reusable identity-bound realtime state session. + public sealed class CultMeshRealtimeSession : IDisposable + { + private readonly object _gate = new(); + private readonly Action _onTransportFailure; + private ICultMeshRealtimeTransport _transport; + private bool _disposed; + + internal CultMeshRealtimeSession( + CultMeshEndpointId endpointId, + ICultMeshRealtimeTransport transport, + CultMeshSessionState state, + Action onTransportFailure) + { + EndpointId = endpointId ?? throw new ArgumentNullException(nameof(endpointId)); + _transport = transport ?? throw new ArgumentNullException(nameof(transport)); + State = state ?? throw new ArgumentNullException(nameof(state)); + _onTransportFailure = onTransportFailure ?? throw new ArgumentNullException(nameof(onTransportFailure)); + } + + public CultMeshEndpointId EndpointId { get; } + public CultMeshSessionState State { get; private set; } + public string TransportId { get { lock (_gate) return _transport.TransportId; } } + + public async Task SendAsync( + CultMeshRealtimeFrame frame, + CancellationToken cancellationToken = default) + { + if (frame == null) throw new ArgumentNullException(nameof(frame)); + frame.Validate(); + var transport = GetTransport(); + try + { + await transport.SendAsync(frame, cancellationToken).ConfigureAwait(false); + } + catch (Exception) when (!cancellationToken.IsCancellationRequested) + { + _onTransportFailure(this); + throw; + } + } + + public async Task ReceiveAsync(CancellationToken cancellationToken = default) + { + var transport = GetTransport(); + try + { + var frame = await transport.ReceiveAsync(cancellationToken).ConfigureAwait(false); + frame.Validate(); + return frame; + } + catch (Exception) when (!cancellationToken.IsCancellationRequested) + { + _onTransportFailure(this); + throw; + } + } + + internal void MarkOffline(CultMeshSessionState state) + { + lock (_gate) + { + if (_disposed) return; + State = state; + } + } + + public void Dispose() + { + ICultMeshRealtimeTransport transport; + lock (_gate) + { + if (_disposed) return; + _disposed = true; + transport = _transport; + } + transport.Dispose(); + } + + private ICultMeshRealtimeTransport GetTransport() + { + lock (_gate) + { + if (_disposed) throw new ObjectDisposedException(nameof(CultMeshRealtimeSession)); + return _transport; + } + } + } +} diff --git a/src/GameCult.Mesh/CultMeshSessions.cs b/src/GameCult.Mesh/CultMeshSessions.cs index 6c10e263..87f18158 100644 --- a/src/GameCult.Mesh/CultMeshSessions.cs +++ b/src/GameCult.Mesh/CultMeshSessions.cs @@ -39,6 +39,7 @@ namespace GameCult.Mesh public static CultMeshProtocolId Discovery { get; } = new CultMeshProtocolId("cultmesh.discovery.v1"); public static CultMeshProtocolId PeerExchange { get; } = new CultMeshProtocolId("cultmesh.peer_exchange.v1"); public static CultMeshProtocolId Subscriptions { get; } = new CultMeshProtocolId("cultmesh.subscriptions.v1"); + public static CultMeshProtocolId RealtimeState { get; } = new CultMeshProtocolId("cultmesh.realtime_state.v1"); } public enum CultMeshTransportPathKind { Direct, Relay, Tunnel } @@ -101,6 +102,7 @@ namespace GameCult.Mesh public interface ICultMeshTransportConnector { string ConnectorId { get; } + int Priority { get; } bool CanConnect(CultMeshTransportCandidate candidate); Task ConnectAsync( CultMeshTransportCandidate candidate, @@ -108,6 +110,10 @@ namespace GameCult.Mesh CancellationToken cancellationToken = default); } + /// + /// Explicit compatibility connector for the previous RUDP/LiteNetLib schema lanes. + /// Prefer for registered schemas. + /// public sealed class CultMeshSchemaTransportConnector : ICultMeshTransportConnector { private readonly Func _createClient; @@ -124,7 +130,8 @@ namespace GameCult.Mesh _connectTimeout = connectTimeout ?? TimeSpan.FromSeconds(5); } - public string ConnectorId => "cultnet-schema"; + public string ConnectorId => "legacy-datagram-schema"; + public int Priority => 10_000; public bool CanConnect(CultMeshTransportCandidate candidate) => Uri.TryCreate(candidate.Endpoint, UriKind.Absolute, out var uri) && (string.Equals(uri.Scheme, "rudp", StringComparison.OrdinalIgnoreCase) || @@ -159,6 +166,59 @@ namespace GameCult.Mesh } } + /// Preferred connector for registered schema, command, receipt, and manifest traffic. + public sealed class CultMeshTcpSchemaTransportConnector : ICultMeshTransportConnector + { + private readonly Func _createClient; + private readonly ICultMeshClock _clock; + private readonly TimeSpan _connectTimeout; + + public CultMeshTcpSchemaTransportConnector( + Func? createClient = null, + ICultMeshClock? clock = null, + TimeSpan? connectTimeout = null) + { + _createClient = createClient ?? (_ => CultNetSchemaClients.CreateTcpFramed()); + _clock = clock ?? CultMeshSystemClock.Instance; + _connectTimeout = connectTimeout ?? TimeSpan.FromSeconds(5); + } + + public string ConnectorId => "tcp-schema"; + public int Priority => 0; + public bool CanConnect(CultMeshTransportCandidate candidate) => + Uri.TryCreate(candidate.Endpoint, UriKind.Absolute, out var uri) && + string.Equals(uri.Scheme, "cultnet+tcp", StringComparison.OrdinalIgnoreCase); + + public async Task ConnectAsync( + CultMeshTransportCandidate candidate, + CultMeshProtocolId protocol, + CancellationToken cancellationToken = default) + { + if (!CanConnect(candidate)) + throw new NotSupportedException($"TCP schema connector does not support '{candidate.Endpoint}'."); + var client = _createClient(candidate.Endpoint); + try + { + var (host, port) = CultNetSchemaWriteForwarder.ParseEndpoint(candidate.Endpoint); + client.Connect(host, port); + var deadline = _clock.UtcNow + _connectTimeout; + while (!client.Connected) + { + cancellationToken.ThrowIfCancellationRequested(); + if (_clock.UtcNow >= deadline) + throw new TimeoutException($"Timed out connecting to '{candidate.Endpoint}'."); + await _clock.DelayAsync(TimeSpan.FromMilliseconds(25), cancellationToken).ConfigureAwait(false); + } + return client; + } + catch + { + client.Dispose(); + throw; + } + } + } + public sealed class CultMeshSession : IDisposable { private readonly Subject _states = new(); @@ -328,8 +388,12 @@ namespace GameCult.Mesh { private readonly CultMeshDiscoveryService _discovery; private readonly ICultMeshTransportConnector[] _connectors; + private readonly ICultMeshContentTransportConnector[] _contentConnectors; + private readonly ICultMeshRealtimeTransportConnector[] _realtimeConnectors; private readonly CultMeshSessionManagerOptions _options; private readonly ConcurrentDictionary>> _sessions = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary>> _contentSessions = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary>> _realtimeSessions = new(StringComparer.Ordinal); private long _diagnosticSequence; private bool _disposed; @@ -337,9 +401,44 @@ namespace GameCult.Mesh CultMeshDiscoveryService discovery, IEnumerable connectors, CultMeshSessionManagerOptions? options = null) + : this( + discovery, + connectors, + Array.Empty(), + Array.Empty(), + options) + { + } + + public CultMeshSessionManager( + CultMeshDiscoveryService discovery, + IEnumerable connectors, + IEnumerable contentConnectors, + CultMeshSessionManagerOptions? options = null) + : this( + discovery, + connectors, + contentConnectors, + Array.Empty(), + options) + { + } + + public CultMeshSessionManager( + CultMeshDiscoveryService discovery, + IEnumerable connectors, + IEnumerable contentConnectors, + IEnumerable realtimeConnectors, + CultMeshSessionManagerOptions? options = null) { _discovery = discovery ?? throw new ArgumentNullException(nameof(discovery)); _connectors = connectors?.ToArray() ?? throw new ArgumentNullException(nameof(connectors)); + _contentConnectors = contentConnectors?.ToArray() ?? throw new ArgumentNullException(nameof(contentConnectors)); + _realtimeConnectors = realtimeConnectors?.ToArray() ?? throw new ArgumentNullException(nameof(realtimeConnectors)); + if (_contentConnectors.Any(connector => connector == null)) + throw new ArgumentException("Content transport connectors cannot contain null entries.", nameof(contentConnectors)); + if (_realtimeConnectors.Any(connector => connector == null)) + throw new ArgumentException("Realtime transport connectors cannot contain null entries.", nameof(realtimeConnectors)); _options = options ?? new CultMeshSessionManagerOptions(); if (_options.MaxRacedCandidates <= 0) throw new ArgumentOutOfRangeException(nameof(options)); } @@ -371,6 +470,257 @@ namespace GameCult.Mesh } } + /// + /// Connects one reusable streaming content plane by stable endpoint identity. + /// Preferred connector tiers are exhausted before legacy fallback tiers are attempted. + /// + public async Task ConnectContentAsync( + CultMeshEndpointId endpointId, + CancellationToken cancellationToken = default) + { + if (endpointId == null) throw new ArgumentNullException(nameof(endpointId)); + ThrowIfDisposed(); + var key = endpointId.Value; + while (true) + { + var lazy = _contentSessions.GetOrAdd(key, _ => new Lazy>( + () => ConnectContentOwnedAsync(key, endpointId), LazyThreadSafetyMode.ExecutionAndPublication)); + try + { + var session = await AwaitForCallerAsync(lazy.Value, cancellationToken).ConfigureAwait(false); + if (session.State.Status != CultMeshSessionStatus.Offline) return session; + _contentSessions.TryRemove(key, out _); + } + catch + { + _contentSessions.TryRemove(key, out _); + throw; + } + } + } + + private async Task ConnectContentOwnedAsync( + string key, + CultMeshEndpointId endpointId) + { + var result = await ResolveContentPathAsync(endpointId).ConfigureAwait(false); + var session = new CultMeshContentSession( + endpointId, + result.Transport, + new CultMeshSessionState(CultMeshSessionStatus.Online, _options.Clock.UtcNow, result.Candidate), + failed => InvalidateContentSession(key, failed)); + Emit(endpointId, CultMeshProtocols.Content, "online", result.Candidate.Endpoint); + return session; + } + + private async Task ResolveContentPathAsync(CultMeshEndpointId endpointId) + { + if (_contentConnectors.Length == 0) + throw Failure( + CultMeshSessionFailureReason.UnsupportedPath, + "No streaming content connectors are configured. Register the TCP content connector or the explicit legacy RUDP connector."); + + var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery(endpointId.Value)).ConfigureAwait(false); + var candidates = discovery.Candidates + .SelectMany(candidate => candidate.Descriptor.DiscoveryEndpoints) + .Distinct(StringComparer.Ordinal) + .Select(endpoint => new CultMeshTransportCandidate(endpoint)) + .ToArray(); + var routes = candidates + .SelectMany(candidate => _contentConnectors + .Where(connector => connector.CanConnect(candidate)) + .Select(connector => new ContentRoute(candidate, connector))) + .OrderBy(route => route.Connector.Priority) + .ThenBy(route => route.Candidate.Priority) + .ToArray(); + if (routes.Length == 0) + throw Failure( + CultMeshSessionFailureReason.UnsupportedPath, + $"No streaming content connector supports an advertised route for '{endpointId}'."); + + var failures = new List(); + foreach (var tier in routes.GroupBy(route => route.Connector.Priority).OrderBy(group => group.Key)) + { + var attempts = tier.Take(_options.MaxRacedCandidates) + .Select(route => ConnectContentCandidateAsync(endpointId, route)) + .ToList(); + while (attempts.Count > 0) + { + var completed = await Task.WhenAny(attempts).ConfigureAwait(false); + attempts.Remove(completed); + try + { + var result = await completed.ConfigureAwait(false); + foreach (var loser in attempts) + _ = loser.ContinueWith(task => + { + if (task.Status == TaskStatus.RanToCompletion) task.Result.Transport.Dispose(); + }, TaskScheduler.Default); + return result; + } + catch (Exception error) + { + failures.Add(error); + } + } + } + + var last = failures.LastOrDefault(); + throw Failure( + last is TimeoutException ? CultMeshSessionFailureReason.Timeout : CultMeshSessionFailureReason.Transport, + $"No streaming content path connected for '{endpointId}'.", + last); + } + + private static async Task ConnectContentCandidateAsync( + CultMeshEndpointId endpointId, + ContentRoute route) + { + var transport = await route.Connector.ConnectAsync(route.Candidate, endpointId).ConfigureAwait(false); + return new ConnectedContentPath(route.Candidate, transport); + } + + private void InvalidateContentSession(string key, CultMeshContentSession session) + { + session.MarkOffline(new CultMeshSessionState( + CultMeshSessionStatus.Offline, + _options.Clock.UtcNow, + session.State.Path, + new CultMeshSessionFailure( + CultMeshSessionFailureReason.Transport, + "The selected streaming content transport failed.", + session.State.Path?.Endpoint ?? string.Empty))); + if (_contentSessions.TryRemove(key, out var removed) && + removed.IsValueCreated && removed.Value.Status == TaskStatus.RanToCompletion) + removed.Value.Result.Dispose(); + Emit(session.EndpointId, CultMeshProtocols.Content, "offline", session.State.Path?.Endpoint ?? string.Empty); + } + + /// + /// Connects the realtime state plane. No connector is implied: promoted runtimes + /// register QUIC explicitly, while RUDP compatibility remains opt-in. + /// + public async Task ConnectRealtimeAsync( + CultMeshEndpointId endpointId, + CancellationToken cancellationToken = default) + { + if (endpointId == null) throw new ArgumentNullException(nameof(endpointId)); + ThrowIfDisposed(); + var key = endpointId.Value; + while (true) + { + var lazy = _realtimeSessions.GetOrAdd(key, _ => new Lazy>( + () => ConnectRealtimeOwnedAsync(key, endpointId), LazyThreadSafetyMode.ExecutionAndPublication)); + try + { + var session = await AwaitForCallerAsync(lazy.Value, cancellationToken).ConfigureAwait(false); + if (session.State.Status != CultMeshSessionStatus.Offline) return session; + _realtimeSessions.TryRemove(key, out _); + } + catch + { + _realtimeSessions.TryRemove(key, out _); + throw; + } + } + } + + private async Task ConnectRealtimeOwnedAsync( + string key, + CultMeshEndpointId endpointId) + { + var result = await ResolveRealtimePathAsync(endpointId).ConfigureAwait(false); + var session = new CultMeshRealtimeSession( + endpointId, + result.Transport, + new CultMeshSessionState(CultMeshSessionStatus.Online, _options.Clock.UtcNow, result.Candidate), + failed => InvalidateRealtimeSession(key, failed)); + Emit(endpointId, CultMeshProtocols.RealtimeState, "online", result.Candidate.Endpoint); + return session; + } + + private async Task ResolveRealtimePathAsync(CultMeshEndpointId endpointId) + { + if (_realtimeConnectors.Length == 0) + throw Failure( + CultMeshSessionFailureReason.UnsupportedPath, + "No realtime state connectors are configured. Register a QUIC connector explicitly."); + + var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery(endpointId.Value)).ConfigureAwait(false); + var routes = discovery.Candidates + .SelectMany(candidate => candidate.Descriptor.DiscoveryEndpoints) + .Distinct(StringComparer.Ordinal) + .Select(endpoint => new CultMeshTransportCandidate(endpoint)) + .SelectMany(candidate => _realtimeConnectors + .Where(connector => connector.CanConnect(candidate)) + .Select(connector => new RealtimeRoute(candidate, connector))) + .OrderBy(route => route.Connector.Priority) + .ThenBy(route => route.Candidate.Priority) + .ToArray(); + if (routes.Length == 0) + throw Failure( + CultMeshSessionFailureReason.UnsupportedPath, + $"No realtime connector supports an advertised route for '{endpointId}'."); + + var failures = new List(); + foreach (var tier in routes.GroupBy(route => route.Connector.Priority).OrderBy(group => group.Key)) + { + var attempts = tier.Take(_options.MaxRacedCandidates) + .Select(route => ConnectRealtimeCandidateAsync(endpointId, route)) + .ToList(); + while (attempts.Count > 0) + { + var completed = await Task.WhenAny(attempts).ConfigureAwait(false); + attempts.Remove(completed); + try + { + var result = await completed.ConfigureAwait(false); + foreach (var loser in attempts) + _ = loser.ContinueWith(task => + { + if (task.Status == TaskStatus.RanToCompletion) task.Result.Transport.Dispose(); + }, TaskScheduler.Default); + return result; + } + catch (Exception error) + { + failures.Add(error); + } + } + } + + var last = failures.LastOrDefault(); + throw Failure( + last is TimeoutException ? CultMeshSessionFailureReason.Timeout : CultMeshSessionFailureReason.Transport, + $"No realtime state path connected for '{endpointId}'.", + last); + } + + private static async Task ConnectRealtimeCandidateAsync( + CultMeshEndpointId endpointId, + RealtimeRoute route) + { + var transport = await route.Connector.ConnectAsync(route.Candidate, endpointId).ConfigureAwait(false); + return new ConnectedRealtimePath(route.Candidate, transport); + } + + private void InvalidateRealtimeSession(string key, CultMeshRealtimeSession session) + { + var endpoint = session.State.Path?.Endpoint ?? string.Empty; + session.MarkOffline(new CultMeshSessionState( + CultMeshSessionStatus.Offline, + _options.Clock.UtcNow, + session.State.Path, + new CultMeshSessionFailure( + CultMeshSessionFailureReason.Transport, + "The selected realtime state transport failed.", + endpoint))); + if (_realtimeSessions.TryRemove(key, out var removed) && + removed.IsValueCreated && removed.Value.Status == TaskStatus.RanToCompletion) + removed.Value.Result.Dispose(); + Emit(session.EndpointId, CultMeshProtocols.RealtimeState, "offline", endpoint); + } + private async Task ConnectOwnedAsync(string key, CultMeshEndpointId endpointId, CultMeshProtocolId protocol) { var result = await ResolveConnectedPathAsync(endpointId, protocol).ConfigureAwait(false); @@ -387,26 +737,43 @@ namespace GameCult.Mesh var candidates = discovery.Candidates.SelectMany(candidate => candidate.Descriptor.DiscoveryEndpoints) .Distinct(StringComparer.Ordinal) .Select(endpoint => new CultMeshTransportCandidate(endpoint)) - .OrderBy(candidate => candidate.Priority) - .Take(_options.MaxRacedCandidates) + .ToArray(); + var routes = candidates + .SelectMany(candidate => _connectors + .Where(connector => connector.CanConnect(candidate)) + .Select(connector => new SchemaRoute(candidate, connector))) + .OrderBy(route => route.Connector.Priority) + .ThenBy(route => route.Candidate.Priority) .ToArray(); if (candidates.Length == 0) throw Failure(CultMeshSessionFailureReason.Resolution, $"No route candidates for '{endpointId}'."); + if (routes.Length == 0) + throw Failure( + CultMeshSessionFailureReason.Transport, + $"No transport path connected for '{endpointId}' and protocol '{protocol}'.", + Failure( + CultMeshSessionFailureReason.UnsupportedPath, + $"No registered connector supports an advertised route for '{endpointId}'.")); - var attempts = candidates.Select(candidate => ConnectCandidateAsync(candidate, protocol)).ToList(); var failures = new List(); - while (attempts.Count > 0) + foreach (var tier in routes.GroupBy(route => route.Connector.Priority).OrderBy(group => group.Key)) { - var completed = await Task.WhenAny(attempts).ConfigureAwait(false); - attempts.Remove(completed); - try + var attempts = tier.Take(_options.MaxRacedCandidates) + .Select(route => ConnectCandidateAsync(route, protocol)) + .ToList(); + while (attempts.Count > 0) { - var result = await completed.ConfigureAwait(false); - foreach (var loser in attempts) - _ = loser.ContinueWith(task => { if (task.Status == TaskStatus.RanToCompletion) task.Result.Client.Dispose(); }, TaskScheduler.Default); - return result; + var completed = await Task.WhenAny(attempts).ConfigureAwait(false); + attempts.Remove(completed); + try + { + var result = await completed.ConfigureAwait(false); + foreach (var loser in attempts) + _ = loser.ContinueWith(task => { if (task.Status == TaskStatus.RanToCompletion) task.Result.Client.Dispose(); }, TaskScheduler.Default); + return result; + } + catch (Exception error) { failures.Add(error); } } - catch (Exception error) { failures.Add(error); } } var last = failures.LastOrDefault(); throw Failure(last is TimeoutException ? CultMeshSessionFailureReason.Timeout : CultMeshSessionFailureReason.Transport, @@ -467,12 +834,10 @@ namespace GameCult.Mesh } } - private async Task ConnectCandidateAsync(CultMeshTransportCandidate candidate, CultMeshProtocolId protocol) + private static async Task ConnectCandidateAsync(SchemaRoute route, CultMeshProtocolId protocol) { - var connector = _connectors.FirstOrDefault(item => item.CanConnect(candidate)); - if (connector == null) throw Failure(CultMeshSessionFailureReason.UnsupportedPath, $"No connector supports '{candidate.Endpoint}'."); - var client = await connector.ConnectAsync(candidate, protocol).ConfigureAwait(false); - return new ConnectedPath(candidate, client); + var client = await route.Connector.ConnectAsync(route.Candidate, protocol).ConfigureAwait(false); + return new ConnectedPath(route.Candidate, client); } private static async Task AwaitForCallerAsync(Task shared, CancellationToken cancellationToken) @@ -506,6 +871,12 @@ namespace GameCult.Mesh foreach (var lazy in _sessions.Values) if (lazy.IsValueCreated && lazy.Value.Status == TaskStatus.RanToCompletion) lazy.Value.Result.Dispose(); _sessions.Clear(); + foreach (var lazy in _contentSessions.Values) + if (lazy.IsValueCreated && lazy.Value.Status == TaskStatus.RanToCompletion) lazy.Value.Result.Dispose(); + _contentSessions.Clear(); + foreach (var lazy in _realtimeSessions.Values) + if (lazy.IsValueCreated && lazy.Value.Status == TaskStatus.RanToCompletion) lazy.Value.Result.Dispose(); + _realtimeSessions.Clear(); } private void ThrowIfDisposed() { if (_disposed) throw new ObjectDisposedException(nameof(CultMeshSessionManager)); } @@ -516,5 +887,60 @@ namespace GameCult.Mesh public CultMeshTransportCandidate Candidate { get; } public ICultNetSchemaClient Client { get; } } + + private sealed class ConnectedContentPath + { + public ConnectedContentPath(CultMeshTransportCandidate candidate, ICultMeshContentTransport transport) + { + Candidate = candidate; + Transport = transport; + } + public CultMeshTransportCandidate Candidate { get; } + public ICultMeshContentTransport Transport { get; } + } + + private sealed class SchemaRoute + { + public SchemaRoute(CultMeshTransportCandidate candidate, ICultMeshTransportConnector connector) + { + Candidate = candidate; + Connector = connector; + } + public CultMeshTransportCandidate Candidate { get; } + public ICultMeshTransportConnector Connector { get; } + } + + private sealed class ContentRoute + { + public ContentRoute(CultMeshTransportCandidate candidate, ICultMeshContentTransportConnector connector) + { + Candidate = candidate; + Connector = connector; + } + public CultMeshTransportCandidate Candidate { get; } + public ICultMeshContentTransportConnector Connector { get; } + } + + private sealed class ConnectedRealtimePath + { + public ConnectedRealtimePath(CultMeshTransportCandidate candidate, ICultMeshRealtimeTransport transport) + { + Candidate = candidate; + Transport = transport; + } + public CultMeshTransportCandidate Candidate { get; } + public ICultMeshRealtimeTransport Transport { get; } + } + + private sealed class RealtimeRoute + { + public RealtimeRoute(CultMeshTransportCandidate candidate, ICultMeshRealtimeTransportConnector connector) + { + Candidate = candidate; + Connector = connector; + } + public CultMeshTransportCandidate Candidate { get; } + public ICultMeshRealtimeTransportConnector Connector { get; } + } } } diff --git a/src/GameCult.Mesh/CultMeshTcpContentTransport.cs b/src/GameCult.Mesh/CultMeshTcpContentTransport.cs new file mode 100644 index 00000000..f70f2a34 --- /dev/null +++ b/src/GameCult.Mesh/CultMeshTcpContentTransport.cs @@ -0,0 +1,421 @@ +using System; +using System.Buffers.Binary; +using System.Collections.Concurrent; +using System.IO; +using System.Net; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; +using GameCult.Caching; +using GameCult.Networking; + +namespace GameCult.Mesh +{ + /// Creates dedicated TCP streams for immutable CultMesh content. + public sealed class CultMeshTcpContentTransportConnector : ICultMeshContentTransportConnector + { + public const string Scheme = "cultmesh-content+tcp"; + private readonly TimeSpan _connectTimeout; + private int _connectCount; + + public CultMeshTcpContentTransportConnector(TimeSpan? connectTimeout = null) + { + _connectTimeout = connectTimeout ?? TimeSpan.FromSeconds(5); + if (_connectTimeout <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(connectTimeout)); + } + + public string ConnectorId => "tcp-content"; + public int Priority => 0; + public int ConnectCount => Volatile.Read(ref _connectCount); + + public bool CanConnect(CultMeshTransportCandidate candidate) => + 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 (!TryParseEndpoint(candidate.Endpoint, out var host, out var port)) + throw new NotSupportedException($"TCP content connector does not support '{candidate.Endpoint}'."); + + var client = new TcpClient(); + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + timeout.CancelAfter(_connectTimeout); + try + { + await CultMeshTcpContentWire.AwaitAsync(client.ConnectAsync(host, port), timeout.Token) + .ConfigureAwait(false); + Interlocked.Increment(ref _connectCount); + return new CultMeshTcpContentTransport(candidate.Endpoint, client); + } + catch + { + client.Dispose(); + throw; + } + } + + internal 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; + } + } + + internal sealed class CultMeshTcpContentTransport : ICultMeshContentTransport + { + private readonly TcpClient _client; + private readonly NetworkStream _stream; + private readonly SemaphoreSlim _sendLock = new(1, 1); + private readonly ConcurrentDictionary _pending = new(StringComparer.Ordinal); + private readonly CancellationTokenSource _shutdown = new(); + private readonly Task _receiveLoop; + private bool _disposed; + + public CultMeshTcpContentTransport(string endpoint, TcpClient client) + { + Endpoint = endpoint; + _client = client ?? throw new ArgumentNullException(nameof(client)); + _stream = client.GetStream(); + _receiveLoop = ReceiveLoopAsync(_shutdown.Token); + } + + public string TransportId => "tcp-content"; + public string Endpoint { get; } + + public async Task CopyChunkToAsync( + CultMeshCdnChunkRef chunk, + Stream destination, + CancellationToken cancellationToken = default) + { + if (_disposed) throw new ObjectDisposedException(nameof(CultMeshTcpContentTransport)); + if (chunk == null) throw new ArgumentNullException(nameof(chunk)); + if (destination == null) throw new ArgumentNullException(nameof(destination)); + if (!destination.CanWrite) throw new ArgumentException("Content destination must be writable.", nameof(destination)); + + var messageId = Guid.NewGuid().ToString("N"); + var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var pending = new PendingCopy(destination, completion, cancellationToken); + if (!_pending.TryAdd(messageId, pending)) + throw new InvalidOperationException("Duplicate TCP content request identity."); + try + { + var request = new CultMeshContentChunkRequestMessage + { + MessageId = messageId, + ChunkHash = CultMeshCdn.NormalizeHash(chunk.ChunkHash, nameof(chunk.ChunkHash)), + RecordKey = string.IsNullOrWhiteSpace(chunk.RecordKey) + ? CultMeshCdnArtifactChunk.CreateRecordKey(chunk.ChunkHash).Value + : chunk.RecordKey, + ExpectedSizeBytes = chunk.SizeBytes + }; + var payload = CultNetSchemaMessageSerialization.Serialize(request); + await _sendLock.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + await CultMeshTcpContentWire.WriteFrameAsync(_stream, payload, cancellationToken).ConfigureAwait(false); + } + finally + { + _sendLock.Release(); + } + await CultMeshTcpContentWire.AwaitAsync(completion.Task, cancellationToken).ConfigureAwait(false); + } + finally + { + _pending.TryRemove(messageId, out _); + pending.Dispose(); + } + } + + public void Dispose() + { + if (_disposed) return; + _disposed = true; + _shutdown.Cancel(); + _client.Dispose(); + var error = new ObjectDisposedException(nameof(CultMeshTcpContentTransport)); + foreach (var pending in _pending.Values) pending.Completion.TrySetException(error); + _pending.Clear(); + _sendLock.Dispose(); + _shutdown.Dispose(); + } + + private async Task ReceiveLoopAsync(CancellationToken cancellationToken) + { + try + { + while (!cancellationToken.IsCancellationRequested) + { + var headerBytes = await CultMeshTcpContentWire.ReadFrameAsync(_stream, cancellationToken).ConfigureAwait(false); + var response = CultNetSchemaMessageSerialization.Deserialize(headerBytes) as CultMeshContentChunkResponseMessage + ?? throw new InvalidDataException("TCP content response used an unexpected schema."); + if (!_pending.TryGetValue(response.MessageId ?? string.Empty, out var pending)) + throw new InvalidDataException("TCP content response did not match an active request."); + if (!response.Found) + { + pending.Completion.TrySetException(new FileNotFoundException(response.Error, response.ChunkHash)); + continue; + } + await CultMeshTcpContentWire.CopyExactlyAsync( + _stream, + pending.Destination, + response.SizeBytes, + pending.CancellationToken).ConfigureAwait(false); + pending.Completion.TrySetResult(true); + } + } + catch (Exception error) when (!_disposed) + { + foreach (var pending in _pending.Values) pending.Completion.TrySetException(error); + } + } + + private sealed class PendingCopy : IDisposable + { + private readonly CancellationTokenRegistration _registration; + + public PendingCopy( + Stream destination, + TaskCompletionSource completion, + CancellationToken cancellationToken) + { + Destination = destination; + Completion = completion; + CancellationToken = cancellationToken; + _registration = cancellationToken.Register(() => completion.TrySetCanceled(cancellationToken)); + } + + public Stream Destination { get; } + public TaskCompletionSource Completion { get; } + public CancellationToken CancellationToken { get; } + public void Dispose() => _registration.Dispose(); + } + } + + /// + /// Serves content-addressed chunks over a dedicated TCP byte stream. Typed headers + /// remain CultNet schema messages; chunk bytes are streamed outside those messages. + /// + public sealed class CultMeshTcpContentServer : IDisposable + { + private readonly TcpListener _listener; + private readonly CultCache _content; + private readonly Func _canServeHash; + private readonly CancellationTokenSource _shutdown = new(); + private readonly ConcurrentDictionary _clients = new(); + private readonly Task _acceptLoop; + private long _chunkRequestsServed; + private bool _disposed; + + public CultMeshTcpContentServer( + TcpListener listener, + CultCache content, + Func? canServeHash = null) + { + _listener = listener ?? throw new ArgumentNullException(nameof(listener)); + _content = content ?? throw new ArgumentNullException(nameof(content)); + _canServeHash = canServeHash ?? (_ => true); + _listener.Start(); + _acceptLoop = AcceptLoopAsync(_shutdown.Token); + } + + public IPEndPoint LocalEndPoint => (IPEndPoint)_listener.LocalEndpoint; + public long ChunkRequestsServed => Interlocked.Read(ref _chunkRequestsServed); + + public void Dispose() + { + if (_disposed) return; + _disposed = true; + _shutdown.Cancel(); + _listener.Stop(); + foreach (var client in _clients.Keys) client.Dispose(); + _clients.Clear(); + _shutdown.Dispose(); + } + + private async Task AcceptLoopAsync(CancellationToken cancellationToken) + { + try + { + while (!cancellationToken.IsCancellationRequested) + { + var client = await _listener.AcceptTcpClientAsync().ConfigureAwait(false); + _clients.TryAdd(client, 0); + _ = ServeClientAsync(client, cancellationToken); + } + } + catch (Exception) when (_disposed || cancellationToken.IsCancellationRequested) + { + } + } + + private async Task ServeClientAsync(TcpClient client, CancellationToken cancellationToken) + { + try + { + using var stream = client.GetStream(); + while (!cancellationToken.IsCancellationRequested) + { + var requestBytes = await CultMeshTcpContentWire.ReadFrameAsync(stream, cancellationToken).ConfigureAwait(false); + var request = CultNetSchemaMessageSerialization.Deserialize(requestBytes) as CultMeshContentChunkRequestMessage + ?? throw new InvalidDataException("TCP content request used an unexpected schema."); + await SendChunkAsync(stream, request, cancellationToken).ConfigureAwait(false); + } + } + catch (Exception) when (_disposed || cancellationToken.IsCancellationRequested || !client.Connected) + { + } + finally + { + _clients.TryRemove(client, out _); + client.Dispose(); + } + } + + private async Task SendChunkAsync( + Stream stream, + CultMeshContentChunkRequestMessage request, + CancellationToken cancellationToken) + { + var response = new CultMeshContentChunkResponseMessage + { + MessageId = request.MessageId, + ChunkHash = request.ChunkHash, + Payload = Array.Empty() + }; + CultMeshCdnArtifactChunk? chunk = null; + try + { + if (string.IsNullOrWhiteSpace(request.MessageId)) + throw new InvalidDataException("Content chunk request requires a message identity."); + var hash = CultMeshCdn.NormalizeHash(request.ChunkHash, nameof(request.ChunkHash)); + if (!_canServeHash(hash)) + throw new UnauthorizedAccessException("Content access policy rejected the requested hash."); + var key = CultMeshCdnArtifactChunk.CreateRecordKey(hash); + if (!string.IsNullOrWhiteSpace(request.RecordKey) && + !string.Equals(request.RecordKey, key.Value, StringComparison.Ordinal)) + throw new InvalidDataException("Content chunk record key disagrees with its content hash."); + chunk = _content.Get(key) + ?? throw new FileNotFoundException("Content chunk is not available.", key.Value); + CultMeshCdn.ValidateChunkPayload(new CultMeshCdnChunkRef + { + ChunkHash = hash, + RecordKey = key.Value, + SizeBytes = request.ExpectedSizeBytes + }, chunk); + response.Found = true; + response.ChunkHash = hash; + response.SizeBytes = chunk.SizeBytes; + Interlocked.Increment(ref _chunkRequestsServed); + } + catch (Exception error) + { + response.Found = false; + response.Error = error.GetType().Name + ": " + error.Message; + } + + await CultMeshTcpContentWire.WriteFrameAsync( + stream, + CultNetSchemaMessageSerialization.Serialize(response), + cancellationToken).ConfigureAwait(false); + if (chunk != null) + { + await stream.WriteAsync(chunk.Payload, 0, chunk.Payload.Length, cancellationToken).ConfigureAwait(false); + await stream.FlushAsync(cancellationToken).ConfigureAwait(false); + } + } + } + + internal static class CultMeshTcpContentWire + { + private const int HeaderBytes = 4; + private const int MaximumHeaderBytes = 1024 * 1024; + + public static async Task AwaitAsync(Task task, CancellationToken cancellationToken) + { + if (!cancellationToken.CanBeCanceled) + { + await task.ConfigureAwait(false); + return; + } + var cancelled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using (cancellationToken.Register(() => cancelled.TrySetResult(true))) + { + if (await Task.WhenAny(task, cancelled.Task).ConfigureAwait(false) != task) + throw new OperationCanceledException(cancellationToken); + } + await task.ConfigureAwait(false); + } + + public static async Task WriteFrameAsync( + Stream stream, + byte[] payload, + CancellationToken cancellationToken) + { + if (payload.Length > MaximumHeaderBytes) + throw new InvalidDataException("TCP content control frame exceeds its size bound."); + var header = new byte[HeaderBytes]; + BinaryPrimitives.WriteUInt32BigEndian(header, checked((uint)payload.Length)); + await stream.WriteAsync(header, 0, header.Length, cancellationToken).ConfigureAwait(false); + await stream.WriteAsync(payload, 0, payload.Length, cancellationToken).ConfigureAwait(false); + await stream.FlushAsync(cancellationToken).ConfigureAwait(false); + } + + public static async Task ReadFrameAsync(Stream stream, CancellationToken cancellationToken) + { + var header = new byte[HeaderBytes]; + await ReadExactlyAsync(stream, header, cancellationToken).ConfigureAwait(false); + var size = checked((int)BinaryPrimitives.ReadUInt32BigEndian(header)); + if (size <= 0 || size > MaximumHeaderBytes) + throw new InvalidDataException("TCP content control frame has an invalid size."); + var payload = new byte[size]; + await ReadExactlyAsync(stream, payload, cancellationToken).ConfigureAwait(false); + return payload; + } + + public static async Task CopyExactlyAsync( + Stream source, + Stream destination, + long bytes, + CancellationToken cancellationToken) + { + if (bytes < 0) throw new InvalidDataException("TCP content body size cannot be negative."); + var buffer = new byte[81920]; + var remaining = bytes; + while (remaining > 0) + { + var requested = (int)Math.Min(buffer.Length, remaining); + var read = await source.ReadAsync(buffer, 0, requested, cancellationToken).ConfigureAwait(false); + if (read == 0) throw new EndOfStreamException(); + await destination.WriteAsync(buffer, 0, read, cancellationToken).ConfigureAwait(false); + remaining -= read; + } + } + + private static async Task ReadExactlyAsync( + Stream stream, + byte[] buffer, + CancellationToken cancellationToken) + { + var offset = 0; + while (offset < buffer.Length) + { + var read = await stream.ReadAsync(buffer, offset, buffer.Length - offset, cancellationToken) + .ConfigureAwait(false); + if (read == 0) throw new EndOfStreamException(); + offset += read; + } + } + } +} diff --git a/src/GameCult.Mesh/docs/reliability-migration-plan.md b/src/GameCult.Mesh/docs/reliability-migration-plan.md index bcab4537..72e60ca9 100644 --- a/src/GameCult.Mesh/docs/reliability-migration-plan.md +++ b/src/GameCult.Mesh/docs/reliability-migration-plan.md @@ -2,6 +2,8 @@ Incident record: [CultMesh bulk content over custom RUDP, July 2026](postmortem-2026-07-cultmesh-bulk-rudp.md). +Physical-plane authority: [CultMesh transport planes](transport-planes.md). + ## Objective CultMesh should give an application one typed, watchable route to a Verse, @@ -676,16 +678,17 @@ body in process. **Owner:** CultMesh negotiation selects a body transport from provider-owned capabilities and consumer demand. The selected transport owns byte movement; `CultMeshContentTransferService` continues to own content identity, resumable -verification, and atomic cache promotion. CultNet RUDP does not own bulk-body -performance. +verification, and atomic cache promotion. TCP owns remote immutable content; +CultNet RUDP does not own bulk-body performance. **Inputs:** content hash and size, provider-advertised transport candidates, consumer-supported transports, same-machine evidence, and platform support. **Outputs:** an authorized mapped body for same-machine deployments, or a -dedicated remote byte stream feeding the existing verification/promotion owner. -Typed control state, reactive variables, commands, and receipts remain CultNet -messages and are not smuggled into the bulk stream. +dedicated TCP byte stream feeding the existing verification/promotion owner. +Schemas, commands, receipts, and manifests use framed TCP CultNet messages. +Realtime state uses a separately negotiated QUIC stream or datagram and is not +smuggled through either snapshots or the immutable-content stream. **Derived state:** selected plane, physical endpoint, mapped-file descriptor, stream progress, and transport telemetry. None can change content identity, diff --git a/src/GameCult.Mesh/docs/transport-planes.md b/src/GameCult.Mesh/docs/transport-planes.md new file mode 100644 index 00000000..879bd5c6 --- /dev/null +++ b/src/GameCult.Mesh/docs/transport-planes.md @@ -0,0 +1,74 @@ +# CultMesh Transport Planes + +CultMesh selects a transport per workload. It does not ask one wire protocol to +be equally good at files, realtime state, and same-machine memory access. + +## Authority map + +**Owner:** `CultMeshSessionManager` selects an advertised route through a +registered connector for each traffic class. A connector owns only its physical +connection and byte movement. + +**Inputs:** stable endpoint identity, advertised physical candidates, connector +support and priority, caller cancellation, and locality evidence for body +negotiation. + +**Outputs:** + +| Workload | Preferred plane | Contract | +| --- | --- | --- | +| Schemas, commands, receipts, manifests | TCP | `cultnet+tcp://` through `ICultMeshTransportConnector` | +| Immutable chunks and other file-shaped bodies | TCP byte stream | `cultmesh-content+tcp://` through `ICultMeshContentTransportConnector` | +| Realtime state | QUIC streams or datagrams | `ICultMeshRealtimeTransportConnector` with an explicit delivery mode | +| Same-machine SoA and large bodies | mapped/shared memory | body transport negotiation | + +The TCP content plane sends typed CultNet request and response headers, then +streams raw chunk bytes outside the schema envelope. The transfer service owns +range hashing, resume checkpoints, provider failover, full-body verification, +and atomic promotion regardless of the selected content connector. +`CultMeshClient` installs the TCP schema and TCP content connectors by default. +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. + +**Derived state:** connector choice, physical endpoint, connection health, and +transport diagnostics. None of these grant provider authority or become content +identity. + +**Forbidden writers:** applications and renderers do not choose unadvertised +endpoints, implement reconnect loops, or bypass verification. Schema messages +must not carry bulk bodies. RUDP/LiteNetLib cannot become an implicit default +for any traffic class. QUIC does not own schemas or immutable files merely +because it can transport bytes. + +**Shared paths:** discovery and identity resolution remain common. Each traffic +class has its own connector set, connection cache, priority tiers, and failure +state. All content connectors feed the same transfer and promotion owner. + +**Cut line:** the former RUDP content server is +`CultMeshLegacyRudpContentServer`, and its connector must be registered +explicitly. It has basement priority and exists for compatibility and parity +measurement. The legacy datagram schema connector is likewise explicit and +lower priority than TCP. Existing hot-state publishers that still package +payloads as schema messages must move to the realtime connector contract; the +new seam does not pretend that migration has already occurred. + +## Security and release gate + +The current TCP connectors are a local-development transport and are plaintext. +Remote release requires authenticated encryption and binding the authenticated +peer to the advertised provider identity. Adding TLS must not move discovery, +authorization, content verification, or application semantics into the TCP +adapter. + +## Local parity evidence + +On 2026-07-19, the parity harness moved a 13 MiB payload on loopback using +1 MiB content chunks. Raw TCP reached 127.4 MiB/s, verified CultMesh TCP content +reached 59.0 MiB/s, raw MsQuic reached 51.5 MiB/s, and legacy CultMesh RUDP +reached 10.7 MiB/s while emitting 18,112,867 wire bytes. CultMesh TCP includes +chunk hashing, durable checkpoints, final hashing, and atomic promotion, so this +is not a claim of raw protocol parity. It is a repeatable workload comparison +and leaves the remaining framework overhead visible. diff --git a/src/GameCult.Networking/CultNetSchemaClient.cs b/src/GameCult.Networking/CultNetSchemaClient.cs index ba8fb699..4071cd1b 100644 --- a/src/GameCult.Networking/CultNetSchemaClient.cs +++ b/src/GameCult.Networking/CultNetSchemaClient.cs @@ -318,6 +318,9 @@ namespace GameCult.Networking return new RudpCultNetSchemaClient(runtimeId, connectionId, connectPayload, maxFragmentBytes, resendDelayMs); } + /// Creates the length-prefixed TCP schema client. + public static ICultNetSchemaClient CreateTcpFramed() => new TcpFramedCultNetSchemaClient(); + /// /// Creates the default schema client adapter for an advertised endpoint URI. /// @@ -341,12 +344,18 @@ namespace GameCult.Networking return CreateRudp(); } + if (string.Equals(uri.Scheme, "cultnet+tcp", StringComparison.OrdinalIgnoreCase)) + { + return CreateTcpFramed(); + } + if (string.Equals(uri.Scheme, "cultnet", StringComparison.OrdinalIgnoreCase)) { return CreateLiteNetLib(security, configureClient); } - throw new FormatException($"CultNet endpoint '{endpoint}' must use cultnet://host:port or rudp://host:port."); + throw new FormatException( + $"CultNet endpoint '{endpoint}' must use cultnet+tcp://host:port, cultnet://host:port, or rudp://host:port."); } } } diff --git a/src/GameCult.Networking/CultNetShardWriteForwarding.cs b/src/GameCult.Networking/CultNetShardWriteForwarding.cs index 44cb4582..b2b9e7a2 100644 --- a/src/GameCult.Networking/CultNetShardWriteForwarding.cs +++ b/src/GameCult.Networking/CultNetShardWriteForwarding.cs @@ -112,7 +112,8 @@ namespace GameCult.Networking !IsSupportedSchemaEndpointScheme(uri.Scheme) || string.IsNullOrWhiteSpace(uri.Host)) { - throw new FormatException($"CultNet endpoint '{endpoint}' must use cultnet://host:port or rudp://host:port."); + throw new FormatException( + $"CultNet endpoint '{endpoint}' must use cultnet+tcp://host:port, cultnet://host:port, or rudp://host:port."); } return (uri.Host, uri.IsDefaultPort ? 3075 : uri.Port); @@ -135,7 +136,8 @@ namespace GameCult.Networking private static bool IsSupportedSchemaEndpointScheme(string scheme) { - return string.Equals(scheme, "cultnet", StringComparison.OrdinalIgnoreCase) || + return string.Equals(scheme, "cultnet+tcp", StringComparison.OrdinalIgnoreCase) || + string.Equals(scheme, "cultnet", StringComparison.OrdinalIgnoreCase) || string.Equals(scheme, "rudp", StringComparison.OrdinalIgnoreCase); } diff --git a/src/GameCult.Networking/CultNetTcpSchemaTransport.cs b/src/GameCult.Networking/CultNetTcpSchemaTransport.cs new file mode 100644 index 00000000..4cdfe06c --- /dev/null +++ b/src/GameCult.Networking/CultNetTcpSchemaTransport.cs @@ -0,0 +1,217 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Net; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; + +namespace GameCult.Networking +{ + /// Schema-v0 client over the length-prefixed TCP lane. + public sealed class TcpFramedCultNetSchemaClient : ICultNetSchemaClient, ICultNetSchemaClientHealth + { + private readonly ConcurrentDictionary> _handlers = new(); + private readonly object _handlerGate = new(); + private readonly object _sendGate = new(); + private readonly TaskCompletionSource _backgroundFailure = new( + TaskCreationOptions.RunContinuationsAsynchronously); + private readonly CancellationTokenSource _shutdown = new(); + private TcpClient? _client; + private TcpFramedTransportConnection? _transport; + private bool _disposed; + + public bool Connected => !_disposed && _client?.Connected == true; + public Task BackgroundFailure => _backgroundFailure.Task; + + public void Connect(string host, int port) + { + if (_disposed) throw new ObjectDisposedException(nameof(TcpFramedCultNetSchemaClient)); + if (_client != null) throw new InvalidOperationException("TCP schema client is already connected."); + var client = new TcpClient(); + try + { + client.Connect(host, port); + _client = client; + _transport = new TcpFramedTransportConnection( + client.GetStream(), + CultNetTransportProfiles.CreateTcpFramed("csharp-tcp-schema-client")); + _ = ReceiveLoopAsync(_shutdown.Token); + } + catch + { + client.Dispose(); + throw; + } + } + + public void SendCultNet(T message) where T : ICultNetSchemaMessage + { + if (_disposed) throw new ObjectDisposedException(nameof(TcpFramedCultNetSchemaClient)); + var transport = _transport ?? throw new InvalidOperationException("TCP schema client is not connected."); + var payload = CultNetSchemaMessageSerialization.Serialize(message); + lock (_sendGate) + transport.SendAsync("schema", payload, _shutdown.Token).GetAwaiter().GetResult(); + } + + public void OnCultNet(Action callback) where T : ICultNetSchemaMessage + { + if (callback == null) throw new ArgumentNullException(nameof(callback)); + lock (_handlerGate) + { + if (!_handlers.TryGetValue(typeof(T), out var handlers)) + _handlers[typeof(T)] = handlers = new List(); + handlers.Add(callback); + } + } + + public void Dispose() + { + if (_disposed) return; + _disposed = true; + _shutdown.Cancel(); + _client?.Dispose(); + _shutdown.Dispose(); + } + + private async Task ReceiveLoopAsync(CancellationToken cancellationToken) + { + try + { + var transport = _transport!; + while (!cancellationToken.IsCancellationRequested) + { + var frame = await transport.ReceiveAsync(cancellationToken).ConfigureAwait(false); + var message = CultNetSchemaMessageSerialization.Deserialize(frame.Payload); + Dispatch(message); + } + } + catch (Exception error) when (!_disposed) + { + _backgroundFailure.TrySetResult(error); + } + } + + private void Dispatch(ICultNetSchemaMessage message) + { + Delegate[] handlers; + lock (_handlerGate) + handlers = _handlers.TryGetValue(message.GetType(), out var registered) + ? registered.ToArray() + : Array.Empty(); + foreach (var handler in handlers) handler.DynamicInvoke(message); + } + } + + /// Schema-v0 server over the length-prefixed TCP lane. + public sealed class TcpFramedCultNetSchemaServer : ICultNetSchemaServer, ICultNetSchemaServerPeerLifecycle, IDisposable + { + private readonly TcpListener _listener; + private readonly ConcurrentDictionary _handlers = new(); + private readonly ConcurrentDictionary _peers = new(); + private readonly CancellationTokenSource _shutdown = new(); + private bool _disposed; + + public TcpFramedCultNetSchemaServer(TcpListener listener) + { + _listener = listener ?? throw new ArgumentNullException(nameof(listener)); + _listener.Start(); + _ = AcceptLoopAsync(_shutdown.Token); + } + + public event Action? PeerDisconnected; + public IPEndPoint LocalEndPoint => (IPEndPoint)_listener.LocalEndpoint; + + public void OnCultNet(Func callback) + where TMessage : ICultNetSchemaMessage + { + if (callback == null) throw new ArgumentNullException(nameof(callback)); + _handlers[typeof(TMessage)] = callback; + } + + public void RemoveCultNetMessageListener(Delegate callback) + where TMessage : ICultNetSchemaMessage + { + if (_handlers.TryGetValue(typeof(TMessage), out var existing) && existing == callback) + _handlers.TryRemove(typeof(TMessage), out _); + } + + public void Dispose() + { + if (_disposed) return; + _disposed = true; + _shutdown.Cancel(); + _listener.Stop(); + foreach (var peer in _peers.Keys) peer.Dispose(); + _peers.Clear(); + _shutdown.Dispose(); + } + + private async Task AcceptLoopAsync(CancellationToken cancellationToken) + { + try + { + while (!cancellationToken.IsCancellationRequested) + { + var client = await _listener.AcceptTcpClientAsync().ConfigureAwait(false); + var peer = new TcpPeer(client); + _peers.TryAdd(peer, 0); + _ = ServePeerAsync(peer, cancellationToken); + } + } + catch (Exception) when (_disposed || cancellationToken.IsCancellationRequested) + { + } + } + + private async Task ServePeerAsync(TcpPeer peer, CancellationToken cancellationToken) + { + try + { + while (!cancellationToken.IsCancellationRequested) + { + var frame = await peer.Transport.ReceiveAsync(cancellationToken).ConfigureAwait(false); + var message = CultNetSchemaMessageSerialization.Deserialize(frame.Payload); + if (_handlers.TryGetValue(message.GetType(), out var handler)) + { + var result = handler.DynamicInvoke(message, peer) as Task; + if (result != null) await result.ConfigureAwait(false); + } + } + } + catch (Exception) when (_disposed || cancellationToken.IsCancellationRequested || !peer.Connected) + { + } + finally + { + if (_peers.TryRemove(peer, out _)) PeerDisconnected?.Invoke(peer); + peer.Dispose(); + } + } + + private sealed class TcpPeer : ICultNetSchemaServerPeer, IDisposable + { + private readonly TcpClient _client; + private readonly object _sendGate = new(); + + public TcpPeer(TcpClient client) + { + _client = client; + Transport = new TcpFramedTransportConnection( + client.GetStream(), + CultNetTransportProfiles.CreateTcpFramed("csharp-tcp-schema-server")); + } + + public bool Connected => _client.Connected; + public TcpFramedTransportConnection Transport { get; } + + public void SendCultNet(TMessage message) where TMessage : ICultNetSchemaMessage + { + var payload = CultNetSchemaMessageSerialization.Serialize(message); + lock (_sendGate) Transport.SendAsync("schema", payload).GetAwaiter().GetResult(); + } + + public void Dispose() => _client.Dispose(); + } + } +} diff --git a/tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs b/tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs index 55e45207..57b8995a 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs @@ -156,6 +156,7 @@ public sealed class CultMeshBodySessionTests private readonly ICultNetSchemaClient _client; public LoopbackConnector(ICultNetSchemaClient client) => _client = client; public string ConnectorId => "loopback"; + public int Priority => 0; public int ConnectCount { get; private set; } public bool CanConnect(CultMeshTransportCandidate candidate) => true; public Task ConnectAsync(CultMeshTransportCandidate candidate, CultMeshProtocolId protocol, diff --git a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs index b7c41dc9..1596b93a 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs @@ -227,6 +227,7 @@ public sealed class CultMeshClientTests { private int _connectCount; public string ConnectorId => "test"; + public int Priority => 0; public int ConnectCount => Volatile.Read(ref _connectCount); public bool CanConnect(CultMeshTransportCandidate candidate) => true; public Task ConnectAsync( @@ -254,6 +255,7 @@ public sealed class CultMeshClientTests public bool SuppressFirstSnapshot { get; set; } public bool SuppressFirstSubscriptionSnapshot { get; set; } public string ConnectorId => "document-test"; + public int Priority => 0; public int SubscribeCount => Clients.Sum(client => client.SubscribeCount); public bool CanConnect(CultMeshTransportCandidate candidate) => true; public Task ConnectAsync( diff --git a/tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs b/tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs index ed300552..3dda96cd 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs @@ -30,19 +30,23 @@ public sealed class CultMeshContentSessionTests } [Test] - public async Task ContentSessionTransfersLargeBodyWithoutSnapshotPayloadsAndReusesCommittedBody() + public async Task TcpContentSessionStreamsLargeBodyAndReusesCommittedBody() { var payload = Enumerable.Range(0, 700_000).Select(value => (byte)(value % 251)).ToArray(); var artifact = CultMesh.PackCdnArtifact("aetheria/world/windows", payload); using var providerCache = new CultCache(CultMesh.CreateCultCacheDocumentRegistry( typeof(CultMeshCdnArtifactManifest), typeof(CultMeshCdnArtifactChunk))); await CultMeshCdn.PublishAsync(providerCache, artifact); - var wireServer = new LoopbackSchemaServer(); - using var contentServer = new CultMeshContentServer(wireServer, providerCache); - var wireClient = new LoopbackSchemaClient(wireServer); - var connector = new LoopbackConnector(wireClient); - using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource() }); - using var sessions = new CultMeshSessionManager(discovery, new[] { connector }); + using var contentServer = new CultMeshTcpContentServer( + new TcpListener(IPAddress.Loopback, 0), + providerCache); + var endpoint = $"{CultMeshTcpContentTransportConnector.Scheme}://127.0.0.1:{contentServer.LocalEndPoint.Port}"; + var connector = new CultMeshTcpContentTransportConnector(); + using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource(endpoint) }); + using var sessions = new CultMeshSessionManager( + discovery, + Array.Empty(), + new ICultMeshContentTransportConnector[] { connector }); var provider = new CultMeshSessionContentProvider( "aetheria.daemon", sessions, @@ -55,16 +59,14 @@ public sealed class CultMeshContentSessionTests new CultMeshContentTransferOptions(_directory)); var first = await transfer.FetchAsync(artifact.Manifest); - var requestsAfterCold = wireClient.ContentRequestCount; + var requestsAfterCold = contentServer.ChunkRequestsServed; var second = await transfer.FetchAsync(artifact.Manifest); File.ReadAllBytes(first).Should().Equal(payload); second.Should().Be(first); requestsAfterCold.Should().Be(artifact.Manifest.Chunks.Length); - wireClient.ContentRequestCount.Should().Be(requestsAfterCold, + contentServer.ChunkRequestsServed.Should().Be(requestsAfterCold, "a committed verified body is the warm-cache authority"); - wireClient.SnapshotRequestCount.Should().Be(0, - "bulk content must never be represented as snapshot records"); connector.ConnectCount.Should().Be(1, "all chunk requests borrow one identity-first content session"); } @@ -86,7 +88,7 @@ public sealed class CultMeshContentSessionTests MaxFragmentBytes = 1024, MaxPendingReliablePackets = 8192 }); - using var contentServer = new CultMeshContentServer(wireServer, providerCache); + using var contentServer = new CultMeshLegacyRudpContentServer(wireServer, providerCache); using var pumpCancellation = new CancellationTokenSource(); var pump = Task.Run(async () => { @@ -101,14 +103,16 @@ public sealed class CultMeshContentSessionTests var endpoint = $"rudp://127.0.0.1:{wireServer.LocalEndPoint.Port}"; using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource(endpoint) }); + var connector = new CultMeshLegacyRudpContentTransportConnector( + new CultMeshLegacyRudpContentTransportOptions { ResponseTimeout = TimeSpan.FromSeconds(20) }); using var sessions = new CultMeshSessionManager( discovery, - new ICultMeshTransportConnector[] { new CultMeshSchemaTransportConnector() }); + Array.Empty(), + new ICultMeshContentTransportConnector[] { connector }); var provider = new CultMeshSessionContentProvider( "aetheria.daemon", sessions, - CultMeshEndpointId.Parse("service:aetheria.daemon"), - new CultMeshSessionContentProviderOptions { ResponseTimeout = TimeSpan.FromSeconds(20) }); + CultMeshEndpointId.Parse("service:aetheria.daemon")); using var transferState = new CultCache(CultMesh.CreateCultCacheDocumentRegistry( typeof(CultMeshContentTransferStateDocument))); var transfer = new CultMeshContentTransferService( @@ -205,6 +209,7 @@ public sealed class CultMeshContentSessionTests private int _connectCount; public LoopbackConnector(ICultNetSchemaClient client) => _client = client; public string ConnectorId => "loopback"; + public int Priority => 0; public int ConnectCount => Volatile.Read(ref _connectCount); public bool CanConnect(CultMeshTransportCandidate candidate) => true; public Task ConnectAsync( diff --git a/tests/GameCult.Mesh.Tests/CultMeshContentTransferTests.cs b/tests/GameCult.Mesh.Tests/CultMeshContentTransferTests.cs index 8bbf43a9..04c6e283 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshContentTransferTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshContentTransferTests.cs @@ -365,7 +365,10 @@ public sealed class CultMeshContentTransferTests public void ReleaseBlockedChunk() => _releaseBlockedChunk.TrySetResult(true); - public async Task GetChunkAsync(CultMeshCdnChunkRef chunk, CancellationToken cancellationToken = default) + public async Task CopyChunkToAsync( + CultMeshCdnChunkRef chunk, + Stream destination, + CancellationToken cancellationToken = default) { Requests.AddOrUpdate(chunk.ChunkHash, 1, (_, count) => count + 1); var request = Interlocked.Increment(ref _requestCount); @@ -384,7 +387,9 @@ public sealed class CultMeshContentTransferTests } if (DelayMilliseconds > 0) await Task.Delay(DelayMilliseconds, cancellationToken); - return _chunks.TryGetValue(chunk.ChunkHash, out var value) ? Clone(value) : null; + if (!_chunks.TryGetValue(chunk.ChunkHash, out var value)) + throw new FileNotFoundException("Synthetic content chunk is missing.", chunk.RecordKey); + await destination.WriteAsync(value.Payload, cancellationToken); } finally { diff --git a/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs b/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs index 4d03f09a..b0aabd03 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs @@ -243,6 +243,7 @@ public sealed class CultMeshSessionManagerTests private int _connectCount; public FakeConnector(Func> connect) { _connect = connect; } public string ConnectorId => "fake"; + public int Priority => 0; public int ConnectCount => Volatile.Read(ref _connectCount); public bool CanConnect(CultMeshTransportCandidate candidate) => true; public Task ConnectAsync(CultMeshTransportCandidate candidate, CultMeshProtocolId protocol, CancellationToken cancellationToken = default) diff --git a/tests/GameCult.Mesh.Tests/CultMeshTransportModularityTests.cs b/tests/GameCult.Mesh.Tests/CultMeshTransportModularityTests.cs new file mode 100644 index 00000000..a280c436 --- /dev/null +++ b/tests/GameCult.Mesh.Tests/CultMeshTransportModularityTests.cs @@ -0,0 +1,272 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Net; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; +using FluentAssertions; +using GameCult.Caching; +using GameCult.Networking; +using NUnit.Framework; + +namespace GameCult.Mesh.Tests; + +[TestFixture] +public sealed class CultMeshTransportModularityTests +{ + [Test] + public async Task SchemaSessionPrefersTcpTierWithoutTouchingLegacyDatagrams() + { + using var server = new TcpFramedCultNetSchemaServer(new TcpListener(IPAddress.Loopback, 0)); + server.OnCultNet((message, peer) => + { + peer.SendCultNet(new CultNetErrorMessage { Error = "tcp:" + message.RuntimeId }); + return Task.CompletedTask; + }); + var tcpEndpoint = $"cultnet+tcp://127.0.0.1:{server.LocalEndPoint.Port}"; + var legacy = new RejectingLegacySchemaConnector(); + using var discovery = new CultMeshDiscoveryService(new[] + { + new RouteSource("rudp://127.0.0.1:9", tcpEndpoint) + }); + using var sessions = new CultMeshSessionManager( + discovery, + new ICultMeshTransportConnector[] + { + legacy, + new CultMeshTcpSchemaTransportConnector() + }); + + var session = await sessions.ConnectAsync( + CultMeshEndpointId.Parse("service:aetheria.daemon"), + CultMeshProtocols.Documents); + var response = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var subscription = session.OnCultNet(message => response.TrySetResult(message)); + session.SendCultNet(new CultNetHelloMessage { RuntimeId = "eve-unity" }); + var received = await response.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + session.State.Path!.Endpoint.Should().Be(tcpEndpoint); + received.Error.Should().Be("tcp:eve-unity"); + legacy.ConnectCount.Should().Be(0, + "a lower transport tier is fallback, not a race against the preferred TCP tier"); + } + + [Test] + public async Task ContentSessionPrefersTcpAndStreamsOutsideSchemaPayloads() + { + var payload = new byte[512 * 1024]; + new Random(1234).NextBytes(payload); + var artifact = CultMesh.PackCdnArtifact("transport/modularity", payload); + using var content = new CultCache(CultMesh.CreateCultCacheDocumentRegistry( + typeof(CultMeshCdnArtifactManifest), typeof(CultMeshCdnArtifactChunk))); + await CultMeshCdn.PublishAsync(content, artifact); + using var server = new CultMeshTcpContentServer(new TcpListener(IPAddress.Loopback, 0), content); + var tcpEndpoint = $"{CultMeshTcpContentTransportConnector.Scheme}://127.0.0.1:{server.LocalEndPoint.Port}"; + var legacy = new RejectingLegacyContentConnector(); + using var discovery = new CultMeshDiscoveryService(new[] + { + new RouteSource("rudp://127.0.0.1:9", tcpEndpoint) + }); + using var sessions = new CultMeshSessionManager( + discovery, + Array.Empty(), + new ICultMeshContentTransportConnector[] + { + legacy, + new CultMeshTcpContentTransportConnector() + }); + + var session = await sessions.ConnectContentAsync(CultMeshEndpointId.Parse("service:aetheria.daemon")); + using var destination = new MemoryStream(); + await session.CopyChunkToAsync(artifact.Manifest.Chunks[0], destination); + + session.TransportId.Should().Be("tcp-content"); + destination.ToArray().Should().Equal(artifact.Chunks[0].Payload); + legacy.ConnectCount.Should().Be(0); + } + + [Test] + public async Task ContentSessionFailsClosedWithoutAnExplicitConnector() + { + using var discovery = new CultMeshDiscoveryService(new[] + { + new RouteSource("rudp://127.0.0.1:3076") + }); + using var sessions = new CultMeshSessionManager( + discovery, + Array.Empty()); + + Func connect = async () => await sessions.ConnectContentAsync( + CultMeshEndpointId.Parse("service:aetheria.daemon")); + + await connect.Should().ThrowAsync() + .WithMessage("*No streaming content connectors are configured*"); + } + + [Test] + public async Task RealtimeSessionUsesDedicatedQuicShapedConnectorContract() + { + var connector = new LoopbackRealtimeConnector(); + using var discovery = new CultMeshDiscoveryService(new[] + { + new RouteSource("cultmesh-state+quic://127.0.0.1:3077") + }); + using var sessions = new CultMeshSessionManager( + discovery, + Array.Empty(), + Array.Empty(), + new ICultMeshRealtimeTransportConnector[] { connector }); + var session = await sessions.ConnectRealtimeAsync( + CultMeshEndpointId.Parse("service:aetheria.daemon")); + var sent = new CultMeshRealtimeFrame + { + ChannelId = "aetheria.entities", + SchemaId = "eve.entity_soa.v1", + BodyId = "body:aetheria:entities", + ProducerEpoch = 7, + Sequence = 42, + Delivery = CultMeshRealtimeDelivery.LatestOnly, + Payload = new byte[] { 1, 2, 3 } + }; + + await session.SendAsync(sent); + var received = await session.ReceiveAsync(); + + session.TransportId.Should().Be("test-quic-state"); + received.Should().BeSameAs(sent); + connector.ConnectCount.Should().Be(1); + } + + [Test] + public async Task RealtimeSessionFailsClosedWithoutAnExplicitConnector() + { + using var discovery = new CultMeshDiscoveryService(new[] + { + new RouteSource("rudp://127.0.0.1:3076") + }); + using var sessions = new CultMeshSessionManager( + discovery, + Array.Empty()); + + Func connect = async () => await sessions.ConnectRealtimeAsync( + CultMeshEndpointId.Parse("service:aetheria.daemon")); + + await connect.Should().ThrowAsync() + .WithMessage("*No realtime state connectors are configured*"); + } + + private sealed class RouteSource : ICultMeshLookupSource + { + private readonly string[] _endpoints; + + public RouteSource(params string[] endpoints) => _endpoints = endpoints; + public string SourceId => "transport-modularity-test"; + + public Task> LookupAsync( + CultMeshDiscoveryQuery query, + CancellationToken cancellationToken = default) + { + var now = DateTimeOffset.UtcNow; + return Task.FromResult>(new[] + { + new CultMeshDiscoveryObservation( + new CultMeshVerseDescriptor( + "aetheria", + "Aetheria", + CultMeshVerseAuthorityModel.OperatorCluster, + new CultMeshVerseCompatibility("cultmesh.v1", "rules"), + _endpoints), + SourceId, + now, + now.AddMinutes(1), + CultMeshDiscoveryTrust.Signed) + }); + } + } + + private sealed class RejectingLegacySchemaConnector : ICultMeshTransportConnector + { + private int _connectCount; + public string ConnectorId => "test-legacy-schema"; + public int Priority => 10_000; + public int ConnectCount => Volatile.Read(ref _connectCount); + public bool CanConnect(CultMeshTransportCandidate candidate) => + candidate.Endpoint.StartsWith("rudp://", StringComparison.OrdinalIgnoreCase); + public Task ConnectAsync( + CultMeshTransportCandidate candidate, + CultMeshProtocolId protocol, + CancellationToken cancellationToken = default) + { + Interlocked.Increment(ref _connectCount); + throw new InvalidOperationException("Legacy schema connector should not be touched."); + } + } + + private sealed class RejectingLegacyContentConnector : ICultMeshContentTransportConnector + { + private int _connectCount; + public string ConnectorId => "test-legacy-content"; + public int Priority => 10_000; + public int ConnectCount => Volatile.Read(ref _connectCount); + public bool CanConnect(CultMeshTransportCandidate candidate) => + candidate.Endpoint.StartsWith("rudp://", StringComparison.OrdinalIgnoreCase); + public Task ConnectAsync( + CultMeshTransportCandidate candidate, + CultMeshEndpointId endpointId, + CancellationToken cancellationToken = default) + { + Interlocked.Increment(ref _connectCount); + throw new InvalidOperationException("Legacy content connector should not be touched."); + } + } + + private sealed class LoopbackRealtimeConnector : ICultMeshRealtimeTransportConnector + { + private int _connectCount; + public string ConnectorId => "test-quic-state"; + public int Priority => 0; + public int ConnectCount => Volatile.Read(ref _connectCount); + public bool CanConnect(CultMeshTransportCandidate candidate) => + candidate.Endpoint.StartsWith("cultmesh-state+quic://", StringComparison.OrdinalIgnoreCase); + public Task ConnectAsync( + CultMeshTransportCandidate candidate, + CultMeshEndpointId endpointId, + CancellationToken cancellationToken = default) + { + Interlocked.Increment(ref _connectCount); + return Task.FromResult(new LoopbackRealtimeTransport(candidate.Endpoint)); + } + } + + private sealed class LoopbackRealtimeTransport : ICultMeshRealtimeTransport + { + private readonly Queue _frames = new(); + private readonly SemaphoreSlim _available = new(0); + private bool _disposed; + + public LoopbackRealtimeTransport(string endpoint) => Endpoint = endpoint; + public string TransportId => "test-quic-state"; + public string Endpoint { get; } + + public Task SendAsync(CultMeshRealtimeFrame frame, CancellationToken cancellationToken = default) + { + if (_disposed) throw new ObjectDisposedException(nameof(LoopbackRealtimeTransport)); + lock (_frames) _frames.Enqueue(frame); + _available.Release(); + return Task.CompletedTask; + } + + public async Task ReceiveAsync(CancellationToken cancellationToken = default) + { + await _available.WaitAsync(cancellationToken); + lock (_frames) return _frames.Dequeue(); + } + + public void Dispose() + { + _disposed = true; + _available.Dispose(); + } + } +} diff --git a/tools/GameCult.TransportParity/Program.cs b/tools/GameCult.TransportParity/Program.cs index d729326a..2661e9fa 100644 --- a/tools/GameCult.TransportParity/Program.cs +++ b/tools/GameCult.TransportParity/Program.cs @@ -20,6 +20,7 @@ Console.WriteLine($"payloadBytes={payloadBytes} quicSupported={QuicListener.IsSu Print(MeasureMappedFile(payload)); var tcp = await MeasureTcpAsync(payload, expectedHash); Print(tcp); +Print(await MeasureCultMeshTcpAsync(payload)); if (QuicListener.IsSupported) Print(await MeasureQuicAsync(payload, expectedHash)); @@ -153,7 +154,8 @@ static async Task MeasureQuicAsync(byte[] payload, string expectedHash) static async Task MeasureCultMeshRudpAsync(byte[] payload) { - var artifact = CultMesh.PackCdnArtifact("parity/body", payload); + var artifact = CultMesh.PackCdnArtifact( + "parity/body", payload, new CultMeshCdnPackOptions { ChunkSizeBytes = 1024 * 1024 }); using var providerCache = new CultCache(CultMesh.CreateCultCacheDocumentRegistry( typeof(CultMeshCdnArtifactManifest), typeof(CultMeshCdnArtifactChunk))); await CultMeshCdn.PublishAsync(providerCache, artifact); @@ -167,7 +169,7 @@ static async Task MeasureCultMeshRudpAsync(byte[] payload) MaxFragmentBytes = 1024, MaxPendingReliablePackets = 8192 }); - using var contentServer = new CultMeshContentServer(wireServer, providerCache); + using var contentServer = new CultMeshLegacyRudpContentServer(wireServer, providerCache); using var pumpCancellation = new CancellationTokenSource(); var pump = Task.Run(async () => { @@ -183,10 +185,12 @@ static async Task MeasureCultMeshRudpAsync(byte[] payload) var endpoint = $"rudp://127.0.0.1:{wireServer.LocalEndPoint.Port}"; using var discovery = new CultMeshDiscoveryService([new RouteSource(endpoint)]); using var sessions = new CultMeshSessionManager( - discovery, [new CultMeshSchemaTransportConnector()]); + discovery, + Array.Empty(), + [new CultMeshLegacyRudpContentTransportConnector( + new CultMeshLegacyRudpContentTransportOptions { ResponseTimeout = TimeSpan.FromMinutes(2) })]); var provider = new CultMeshSessionContentProvider( - "parity.provider", sessions, CultMeshEndpointId.Parse("service:parity.provider"), - new CultMeshSessionContentProviderOptions { ResponseTimeout = TimeSpan.FromMinutes(2) }); + "parity.provider", sessions, CultMeshEndpointId.Parse("service:parity.provider")); var directory = Path.Combine(Path.GetTempPath(), "cultmesh-transport-parity", Guid.NewGuid().ToString("N")); using var transferState = new CultCache(CultMesh.CreateCultCacheDocumentRegistry( typeof(CultMeshContentTransferStateDocument))); @@ -210,6 +214,45 @@ static async Task MeasureCultMeshRudpAsync(byte[] payload) } } +static async Task MeasureCultMeshTcpAsync(byte[] payload) +{ + var artifact = CultMesh.PackCdnArtifact( + "parity/body", payload, new CultMeshCdnPackOptions { ChunkSizeBytes = 1024 * 1024 }); + using var providerCache = new CultCache(CultMesh.CreateCultCacheDocumentRegistry( + typeof(CultMeshCdnArtifactManifest), typeof(CultMeshCdnArtifactChunk))); + await CultMeshCdn.PublishAsync(providerCache, artifact); + + using var contentServer = new CultMeshTcpContentServer( + new TcpListener(IPAddress.Loopback, 0), providerCache); + var endpoint = $"{CultMeshTcpContentTransportConnector.Scheme}://127.0.0.1:{contentServer.LocalEndPoint.Port}"; + using var discovery = new CultMeshDiscoveryService([new RouteSource(endpoint)]); + using var sessions = new CultMeshSessionManager( + discovery, + Array.Empty(), + [new CultMeshTcpContentTransportConnector()]); + var provider = new CultMeshSessionContentProvider( + "parity.provider", sessions, CultMeshEndpointId.Parse("service:parity.provider")); + var directory = Path.Combine(Path.GetTempPath(), "cultmesh-transport-parity", Guid.NewGuid().ToString("N")); + using var transferState = new CultCache(CultMesh.CreateCultCacheDocumentRegistry( + typeof(CultMeshContentTransferStateDocument))); + var transfer = new CultMeshContentTransferService( + transferState, [provider], new CultMeshContentTransferOptions(directory)); + + var elapsed = Stopwatch.StartNew(); + try + { + var path = await transfer.FetchAsync(artifact.Manifest); + elapsed.Stop(); + if (!File.ReadAllBytes(path).AsSpan().SequenceEqual(payload)) + throw new InvalidDataException("CultMesh TCP payload did not match source."); + return Result.Create("cultmesh-tcp-content", payload.LongLength, elapsed.Elapsed); + } + finally + { + if (Directory.Exists(directory)) Directory.Delete(directory, true); + } +} + static X509Certificate2 CreateCertificate() { using var key = RSA.Create(2048);