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

Modularize CultMesh transport planes

Browse source
This commit is contained in:
Meta Vens 2026-07-19 17:57:26 +01:00
commit c16163a82f
19 changed files with 2162 additions and 171 deletions
Download patch file Download diff file Expand all files Collapse all files

37
src/GameCult.Mesh/CultMeshClient.cs
View file

@ -22,9 +22,22 @@ namespace GameCult.Mesh
/// <summary>Gets or sets session path and diagnostic policy.</summary>
public CultMeshSessionManagerOptions Sessions { get; set; } = new();
/// <summary>Gets or sets optional transport connectors. The CultNet schema connector is used by default.</summary>
/// <summary>Gets or sets schema connectors. Length-prefixed TCP is used by default.</summary>
public IReadOnlyList<ICultMeshTransportConnector>? Connectors { get; set; }
/// <summary>
/// Gets or sets streaming content connectors. TCP content delivery is used by default;
/// applications may replace it or explicitly add legacy RUDP.
/// </summary>
public IReadOnlyList<ICultMeshContentTransportConnector>? ContentConnectors { get; set; }
/// <summary>
/// Gets or sets realtime state connectors. No connector is installed implicitly;
/// promoted runtimes should register QUIC and may explicitly add a legacy fallback.
/// </summary>
public IReadOnlyList<ICultMeshRealtimeTransportConnector> RealtimeConnectors { get; set; } =
Array.Empty<ICultMeshRealtimeTransportConnector>();
/// <summary>Gets or sets the deadline before an unanswered subscription intent is replayed.</summary>
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);
}
/// <summary>Connects the reusable realtime state plane by stable endpoint identity.</summary>
public Task<CultMeshRealtimeSession> ConnectRealtimeAsync(
string endpointId,
CancellationToken cancellationToken = default)
{
ThrowIfDisposed();
return _sessions.ConnectRealtimeAsync(CultMeshEndpointId.Parse(endpointId), cancellationToken);
}
/// <summary>Connects to a stable endpoint identity using one application protocol.</summary>
public Task<CultMeshSession> ConnectAsync(
CultMeshEndpointId endpointId,
@ -97,15 +124,13 @@ namespace GameCult.Mesh
/// <summary>Creates a verified-transfer provider over this client's reusable content session owner.</summary>
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));
}
/// <summary>Creates a direct body provider over this client's reusable negotiated session owner.</summary>

209
src/GameCult.Mesh/CultMeshContentSessions.cs
View file

@ -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.
/// </summary>
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;
/// <summary>Attaches content serving to an existing schema host and provider cache.</summary>
public CultMeshContentServer(
public CultMeshLegacyRudpContentServer(
ICultNetSchemaServer server,
CultCache content,
Func<string, bool>? canServeHash = null)
@ -85,8 +85,8 @@ namespace GameCult.Mesh
}
}
/// <summary>Configures bounded response waiting for session-backed content requests.</summary>
public sealed class CultMeshSessionContentProviderOptions
/// <summary>Configures the explicit compatibility path for content over schema-message RUDP.</summary>
public sealed class CultMeshLegacyRudpContentTransportOptions
{
/// <summary>Gets or sets the injected reliability clock.</summary>
public ICultMeshClock Clock { get; set; } = CultMeshSystemClock.Instance;
@ -95,97 +95,148 @@ namespace GameCult.Mesh
}
/// <summary>
/// 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.
/// </summary>
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;
/// <summary>Creates a content provider backed by a reusable identity-first session.</summary>
public CultMeshSessionContentProvider(
string providerId,
CultMeshSessionManager sessions,
CultMeshEndpointId endpointId,
CultMeshSessionContentProviderOptions? options = null)
public CultMeshLegacyRudpContentTransportConnector(
CultMeshLegacyRudpContentTransportOptions? options = null,
Func<string, ICultNetSchemaClient>? 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);
}
/// <inheritdoc />
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);
/// <inheritdoc />
public async Task<CultMeshCdnArtifactChunk?> GetChunkAsync(
CultMeshCdnChunkRef chunk,
public async Task<ICultMeshContentTransport> 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<CultMeshContentChunkResponseMessage>(
TaskCreationOptions.RunContinuationsAsynchronously);
using var responseSubscription = session.OnCultNet<CultMeshContentChunkResponseMessage>(response =>
{
if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal))
completion.TrySetResult(response);
});
using var errorSubscription = session.OnCultNet<CultNetErrorMessage>(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<byte>()
};
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<CultMeshContentChunkResponseMessage> WaitForResponseAsync(
Task<CultMeshContentChunkResponseMessage> 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<string, TaskCompletionSource<CultMeshContentChunkResponseMessage>> _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<CultMeshContentChunkResponseMessage>(OnResponse);
_client.OnCultNet<CultNetErrorMessage>(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<CultMeshContentChunkResponseMessage>(
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<byte>()
};
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<CultMeshContentChunkResponseMessage> WaitForResponseAsync(
Task<CultMeshContentChunkResponseMessage> 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 + "'.");
}
}
}

190
src/GameCult.Mesh/CultMeshContentTransfer.cs
View file

@ -32,7 +32,10 @@ namespace GameCult.Mesh
public interface ICultMeshContentProvider
{
string ProviderId { get; }
Task<CultMeshCdnArtifactChunk?> 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<CultMeshCdnArtifactChunk?> 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<CultMeshCdnArtifactChunk>(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<CultMeshContentTransferStateDocument>(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<CultMeshCdnArtifactChunk> FetchVerifiedChunkAsync(
private async Task CopyVerifiedChunkToPartialAsync(
CultMeshCdnChunkRef reference,
string partialPath,
CancellationToken cancellationToken)
{
var failures = new List<Exception>();
@ -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<byte>(), 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,

166
src/GameCult.Mesh/CultMeshContentTransports.cs Normal file
View file

@ -0,0 +1,166 @@
using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
namespace GameCult.Mesh
{
/// <summary>
/// Moves immutable content bytes for one selected physical path. Content identity,
/// verification, checkpoints, and promotion remain owned by
/// <see cref="CultMeshContentTransferService"/>.
/// </summary>
public interface ICultMeshContentTransport : IDisposable
{
string TransportId { get; }
string Endpoint { get; }
/// <summary>
/// Writes exactly one advertised chunk into the caller-owned destination.
/// The transport must honor destination backpressure and must not close it.
/// </summary>
Task CopyChunkToAsync(
CultMeshCdnChunkRef chunk,
Stream destination,
CancellationToken cancellationToken = default);
}
/// <summary>
/// Creates physical content transports. Lower priority values are attempted first;
/// a fallback tier is not raced against a preferred tier.
/// </summary>
public interface ICultMeshContentTransportConnector
{
string ConnectorId { get; }
int Priority { get; }
bool CanConnect(CultMeshTransportCandidate candidate);
Task<ICultMeshContentTransport> ConnectAsync(
CultMeshTransportCandidate candidate,
CultMeshEndpointId endpointId,
CancellationToken cancellationToken = default);
}
/// <summary>
/// Reusable identity-bound content session selected and owned by
/// <see cref="CultMeshSessionManager"/>.
/// </summary>
public sealed class CultMeshContentSession : IDisposable
{
private readonly object _gate = new();
private readonly Action<CultMeshContentSession> _onTransportFailure;
private ICultMeshContentTransport _transport;
private bool _disposed;
internal CultMeshContentSession(
CultMeshEndpointId endpointId,
ICultMeshContentTransport transport,
CultMeshSessionState state,
Action<CultMeshContentSession> 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));
}
}
/// <summary>
/// Adapts a selected streaming transport to the verified content-transfer source port.
/// </summary>
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);
}
}
}

154
src/GameCult.Mesh/CultMeshRealtimeTransports.cs Normal file
View file

@ -0,0 +1,154 @@
using System;
using System.Threading;
using System.Threading.Tasks;
namespace GameCult.Mesh
{
/// <summary>Delivery semantics requested from a realtime state transport.</summary>
public enum CultMeshRealtimeDelivery
{
ReliableOrdered,
LatestOnly,
Unreliable
}
/// <summary>
/// Transport-neutral realtime state frame. Schema identity and logical generation
/// remain stable when the physical connector changes.
/// </summary>
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<byte> 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.");
}
}
/// <summary>
/// Bidirectional realtime state plane. Commands, receipts, manifests, and other
/// registered schema documents do not travel through this port.
/// </summary>
public interface ICultMeshRealtimeTransport : IDisposable
{
string TransportId { get; }
string Endpoint { get; }
Task SendAsync(CultMeshRealtimeFrame frame, CancellationToken cancellationToken = default);
Task<CultMeshRealtimeFrame> ReceiveAsync(CancellationToken cancellationToken = default);
}
/// <summary>Creates a realtime state transport, normally QUIC streams/datagrams.</summary>
public interface ICultMeshRealtimeTransportConnector
{
string ConnectorId { get; }
int Priority { get; }
bool CanConnect(CultMeshTransportCandidate candidate);
Task<ICultMeshRealtimeTransport> ConnectAsync(
CultMeshTransportCandidate candidate,
CultMeshEndpointId endpointId,
CancellationToken cancellationToken = default);
}
/// <summary>Reusable identity-bound realtime state session.</summary>
public sealed class CultMeshRealtimeSession : IDisposable
{
private readonly object _gate = new();
private readonly Action<CultMeshRealtimeSession> _onTransportFailure;
private ICultMeshRealtimeTransport _transport;
private bool _disposed;
internal CultMeshRealtimeSession(
CultMeshEndpointId endpointId,
ICultMeshRealtimeTransport transport,
CultMeshSessionState state,
Action<CultMeshRealtimeSession> 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<CultMeshRealtimeFrame> 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;
}
}
}
}

462
src/GameCult.Mesh/CultMeshSessions.cs
View file

@ -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<ICultNetSchemaClient> ConnectAsync(
CultMeshTransportCandidate candidate,
@ -108,6 +110,10 @@ namespace GameCult.Mesh
CancellationToken cancellationToken = default);
}
/// <summary>
/// Explicit compatibility connector for the previous RUDP/LiteNetLib schema lanes.
/// Prefer <see cref="CultMeshTcpSchemaTransportConnector"/> for registered schemas.
/// </summary>
public sealed class CultMeshSchemaTransportConnector : ICultMeshTransportConnector
{
private readonly Func<string, ICultNetSchemaClient> _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
}
}
/// <summary>Preferred connector for registered schema, command, receipt, and manifest traffic.</summary>
public sealed class CultMeshTcpSchemaTransportConnector : ICultMeshTransportConnector
{
private readonly Func<string, ICultNetSchemaClient> _createClient;
private readonly ICultMeshClock _clock;
private readonly TimeSpan _connectTimeout;
public CultMeshTcpSchemaTransportConnector(
Func<string, ICultNetSchemaClient>? 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<ICultNetSchemaClient> 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<CultMeshSessionState> _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<string, Lazy<Task<CultMeshSession>>> _sessions = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, Lazy<Task<CultMeshContentSession>>> _contentSessions = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, Lazy<Task<CultMeshRealtimeSession>>> _realtimeSessions = new(StringComparer.Ordinal);
private long _diagnosticSequence;
private bool _disposed;
@ -337,9 +401,44 @@ namespace GameCult.Mesh
CultMeshDiscoveryService discovery,
IEnumerable<ICultMeshTransportConnector> connectors,
CultMeshSessionManagerOptions? options = null)
: this(
discovery,
connectors,
Array.Empty<ICultMeshContentTransportConnector>(),
Array.Empty<ICultMeshRealtimeTransportConnector>(),
options)
{
}
public CultMeshSessionManager(
CultMeshDiscoveryService discovery,
IEnumerable<ICultMeshTransportConnector> connectors,
IEnumerable<ICultMeshContentTransportConnector> contentConnectors,
CultMeshSessionManagerOptions? options = null)
: this(
discovery,
connectors,
contentConnectors,
Array.Empty<ICultMeshRealtimeTransportConnector>(),
options)
{
}
public CultMeshSessionManager(
CultMeshDiscoveryService discovery,
IEnumerable<ICultMeshTransportConnector> connectors,
IEnumerable<ICultMeshContentTransportConnector> contentConnectors,
IEnumerable<ICultMeshRealtimeTransportConnector> 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
}
}
/// <summary>
/// Connects one reusable streaming content plane by stable endpoint identity.
/// Preferred connector tiers are exhausted before legacy fallback tiers are attempted.
/// </summary>
public async Task<CultMeshContentSession> 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<Task<CultMeshContentSession>>(
() => 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<CultMeshContentSession> 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<ConnectedContentPath> 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<Exception>();
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<ConnectedContentPath> 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);
}
/// <summary>
/// Connects the realtime state plane. No connector is implied: promoted runtimes
/// register QUIC explicitly, while RUDP compatibility remains opt-in.
/// </summary>
public async Task<CultMeshRealtimeSession> 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<Task<CultMeshRealtimeSession>>(
() => 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<CultMeshRealtimeSession> 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<ConnectedRealtimePath> 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<Exception>();
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<ConnectedRealtimePath> 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<CultMeshSession> 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<Exception>();
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<ConnectedPath> ConnectCandidateAsync(CultMeshTransportCandidate candidate, CultMeshProtocolId protocol)
private static async Task<ConnectedPath> 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<T> AwaitForCallerAsync<T>(Task<T> 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; }
}
}
}

421
src/GameCult.Mesh/CultMeshTcpContentTransport.cs Normal file
View file

@ -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
{
/// <summary>Creates dedicated TCP streams for immutable CultMesh content.</summary>
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<ICultMeshContentTransport> 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<string, PendingCopy> _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<bool>(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<bool> completion,
CancellationToken cancellationToken)
{
Destination = destination;
Completion = completion;
CancellationToken = cancellationToken;
_registration = cancellationToken.Register(() => completion.TrySetCanceled(cancellationToken));
}
public Stream Destination { get; }
public TaskCompletionSource<bool> Completion { get; }
public CancellationToken CancellationToken { get; }
public void Dispose() => _registration.Dispose();
}
}
/// <summary>
/// Serves content-addressed chunks over a dedicated TCP byte stream. Typed headers
/// remain CultNet schema messages; chunk bytes are streamed outside those messages.
/// </summary>
public sealed class CultMeshTcpContentServer : IDisposable
{
private readonly TcpListener _listener;
private readonly CultCache _content;
private readonly Func<string, bool> _canServeHash;
private readonly CancellationTokenSource _shutdown = new();
private readonly ConcurrentDictionary<TcpClient, byte> _clients = new();
private readonly Task _acceptLoop;
private long _chunkRequestsServed;
private bool _disposed;
public CultMeshTcpContentServer(
TcpListener listener,
CultCache content,
Func<string, bool>? 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<byte>()
};
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<CultMeshCdnArtifactChunk>(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<bool>(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<byte[]> 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;
}
}
}
}

13
src/GameCult.Mesh/docs/reliability-migration-plan.md
View file

@ -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,

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

@ -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.

11
src/GameCult.Networking/CultNetSchemaClient.cs
View file

@ -318,6 +318,9 @@ namespace GameCult.Networking
return new RudpCultNetSchemaClient(runtimeId, connectionId, connectPayload, maxFragmentBytes, resendDelayMs);
}
/// <summary>Creates the length-prefixed TCP schema client.</summary>
public static ICultNetSchemaClient CreateTcpFramed() => new TcpFramedCultNetSchemaClient();
/// <summary>
/// Creates the default schema client adapter for an advertised endpoint URI.
/// </summary>
@ -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.");
}
}
}

6
src/GameCult.Networking/CultNetShardWriteForwarding.cs
View file

@ -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);
}

217
src/GameCult.Networking/CultNetTcpSchemaTransport.cs Normal file
View file

@ -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
{
/// <summary>Schema-v0 client over the length-prefixed TCP lane.</summary>
public sealed class TcpFramedCultNetSchemaClient : ICultNetSchemaClient, ICultNetSchemaClientHealth
{
private readonly ConcurrentDictionary<Type, List<Delegate>> _handlers = new();
private readonly object _handlerGate = new();
private readonly object _sendGate = new();
private readonly TaskCompletionSource<Exception> _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<Exception> 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>(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<T>(Action<T> 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<Delegate>();
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<Delegate>();
foreach (var handler in handlers) handler.DynamicInvoke(message);
}
}
/// <summary>Schema-v0 server over the length-prefixed TCP lane.</summary>
public sealed class TcpFramedCultNetSchemaServer : ICultNetSchemaServer, ICultNetSchemaServerPeerLifecycle, IDisposable
{
private readonly TcpListener _listener;
private readonly ConcurrentDictionary<Type, Delegate> _handlers = new();
private readonly ConcurrentDictionary<TcpPeer, byte> _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<ICultNetSchemaServerPeer>? PeerDisconnected;
public IPEndPoint LocalEndPoint => (IPEndPoint)_listener.LocalEndpoint;
public void OnCultNet<TMessage>(Func<TMessage, ICultNetSchemaServerPeer, Task> callback)
where TMessage : ICultNetSchemaMessage
{
if (callback == null) throw new ArgumentNullException(nameof(callback));
_handlers[typeof(TMessage)] = callback;
}
public void RemoveCultNetMessageListener<TMessage>(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>(TMessage message) where TMessage : ICultNetSchemaMessage
{
var payload = CultNetSchemaMessageSerialization.Serialize(message);
lock (_sendGate) Transport.SendAsync("schema", payload).GetAwaiter().GetResult();
}
public void Dispose() => _client.Dispose();
}
}
}

1
tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs
View file

@ -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<ICultNetSchemaClient> ConnectAsync(CultMeshTransportCandidate candidate, CultMeshProtocolId protocol,

2
tests/GameCult.Mesh.Tests/CultMeshClientTests.cs
View file

@ -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<ICultNetSchemaClient> 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<ICultNetSchemaClient> ConnectAsync(

35
tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs
View file

@ -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<ICultMeshTransportConnector>(),
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<ICultMeshTransportConnector>(),
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<ICultNetSchemaClient> ConnectAsync(

9
tests/GameCult.Mesh.Tests/CultMeshContentTransferTests.cs
View file

@ -365,7 +365,10 @@ public sealed class CultMeshContentTransferTests
public void ReleaseBlockedChunk() => _releaseBlockedChunk.TrySetResult(true);
public async Task<CultMeshCdnArtifactChunk?> 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
{

1
tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs
View file

@ -243,6 +243,7 @@ public sealed class CultMeshSessionManagerTests
private int _connectCount;
public FakeConnector(Func<CultMeshTransportCandidate, CancellationToken, Task<ICultNetSchemaClient>> 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<ICultNetSchemaClient> ConnectAsync(CultMeshTransportCandidate candidate, CultMeshProtocolId protocol, CancellationToken cancellationToken = default)

272
tests/GameCult.Mesh.Tests/CultMeshTransportModularityTests.cs Normal file
View file

@ -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<CultNetHelloMessage>((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<CultNetErrorMessage>(TaskCreationOptions.RunContinuationsAsynchronously);
using var subscription = session.OnCultNet<CultNetErrorMessage>(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<ICultMeshTransportConnector>(),
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<ICultMeshTransportConnector>());
Func<Task> connect = async () => await sessions.ConnectContentAsync(
CultMeshEndpointId.Parse("service:aetheria.daemon"));
await connect.Should().ThrowAsync<CultMeshSessionException>()
.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<ICultMeshTransportConnector>(),
Array.Empty<ICultMeshContentTransportConnector>(),
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<ICultMeshTransportConnector>());
Func<Task> connect = async () => await sessions.ConnectRealtimeAsync(
CultMeshEndpointId.Parse("service:aetheria.daemon"));
await connect.Should().ThrowAsync<CultMeshSessionException>()
.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<IReadOnlyList<CultMeshDiscoveryObservation>> LookupAsync(
CultMeshDiscoveryQuery query,
CancellationToken cancellationToken = default)
{
var now = DateTimeOffset.UtcNow;
return Task.FromResult<IReadOnlyList<CultMeshDiscoveryObservation>>(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<ICultNetSchemaClient> 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<ICultMeshContentTransport> 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<ICultMeshRealtimeTransport> ConnectAsync(
CultMeshTransportCandidate candidate,
CultMeshEndpointId endpointId,
CancellationToken cancellationToken = default)
{
Interlocked.Increment(ref _connectCount);
return Task.FromResult<ICultMeshRealtimeTransport>(new LoopbackRealtimeTransport(candidate.Endpoint));
}
}
private sealed class LoopbackRealtimeTransport : ICultMeshRealtimeTransport
{
private readonly Queue<CultMeshRealtimeFrame> _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<CultMeshRealtimeFrame> ReceiveAsync(CancellationToken cancellationToken = default)
{
await _available.WaitAsync(cancellationToken);
lock (_frames) return _frames.Dequeue();
}
public void Dispose()
{
_disposed = true;
_available.Dispose();
}
}
}

53
tools/GameCult.TransportParity/Program.cs
View file

@ -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<Result> MeasureQuicAsync(byte[] payload, string expectedHash)
static async Task<Result> 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<Result> 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<Result> 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<ICultMeshTransportConnector>(),
[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<Result> MeasureCultMeshRudpAsync(byte[] payload)
}
}
static async Task<Result> 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<ICultMeshTransportConnector>(),
[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);