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

Make CultMesh session identity explicit

Browse source
This commit is contained in:
Meta Vens 2026-08-18 07:26:55 +02:00
commit c75051017d
24 changed files with 364 additions and 236 deletions
Download patch file Download diff file Expand all files Collapse all files

24
samples/eve-browser-network/Program.cs
View file

@ -183,7 +183,8 @@ static async Task RunHeadlessAsync(Args arguments)
_ => CreateClient())
]
});
using var lease = await mesh.LeaseDocumentAsync<CounterState>(arguments.VerseId, counterKey);
var target = new CultMeshSessionTarget(arguments.VerseId, arguments.AuthorityRuntimeId);
using var lease = await mesh.LeaseDocumentAsync<CounterState>(target, counterKey);
var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var lastCount = -1;
using var watch = lease.Handle.Watch(counter =>
@ -217,7 +218,7 @@ static async Task RunHeadlessAsync(Args arguments)
throw new InvalidOperationException("The headless sample expected 'INVOKE <idempotency-key>' on stdin.");
var commandId = line["INVOKE ".Length..].Trim();
var result = await mesh.InvokeAsync<IncrementRequest, IncrementReceipt>(
arguments.VerseId,
target,
"sample.counter",
"sample.counter.increment",
"sample.increment.v1",
@ -232,18 +233,21 @@ static async Task RunHeadlessAsync(Args arguments)
status = result.Status
}));
Console.WriteLine("HEADLESS_NETWORK_BENCHMARK " + JsonSerializer.Serialize(
await MeasureOperationSessionAsync(mesh, arguments)));
await MeasureOperationSessionAsync(mesh, target, arguments)));
});
await Task.WhenAll(completion.Task, commandTask).WaitAsync(TimeSpan.FromSeconds(45));
}
static async Task<object> MeasureOperationSessionAsync(CultMeshClient mesh, Args arguments)
static async Task<object> MeasureOperationSessionAsync(
CultMeshClient mesh,
CultMeshSessionTarget target,
Args arguments)
{
const int warmupOperations = 100;
if (arguments.BenchmarkOperations <= 0)
throw new ArgumentOutOfRangeException(nameof(arguments.BenchmarkOperations));
for (var index = 0; index < warmupOperations; index++)
await PingAsync(mesh, arguments, "warmup:" + index, index);
await PingAsync(mesh, target, "warmup:" + index, index);
ForceCollection();
using var process = Process.GetCurrentProcess();
@ -255,7 +259,7 @@ static async Task<object> MeasureOperationSessionAsync(CultMeshClient mesh, Args
for (var index = 0; index < arguments.BenchmarkOperations; index++)
{
var started = Stopwatch.GetTimestamp();
await PingAsync(mesh, arguments, "benchmark:" + index, index);
await PingAsync(mesh, target, "benchmark:" + index, index);
latencies[index] = Stopwatch.GetElapsedTime(started).TotalMilliseconds;
}
clock.Stop();
@ -284,10 +288,14 @@ static async Task<object> MeasureOperationSessionAsync(CultMeshClient mesh, Args
};
}
static async Task PingAsync(CultMeshClient mesh, Args arguments, string idempotencyKey, int sequence)
static async Task PingAsync(
CultMeshClient mesh,
CultMeshSessionTarget target,
string idempotencyKey,
int sequence)
{
var result = await mesh.InvokeAsync<PingRequest, PingReceipt>(
arguments.VerseId,
target,
"sample.counter",
"sample.counter.ping",
"sample.ping_request.v1",

12
scripts/verify-eve-getting-started.ps1
View file

@ -17,6 +17,18 @@ foreach ($requiredPrimitive in @("CultNetOperationServer", "EveSurface.Create"))
throw "The getting-started provider is missing the public primitive '$requiredPrimitive'."
}
}
if ($networkSampleSource -notmatch 'new\s+CultMeshSessionTarget\s*\(') {
throw "The getting-started client must address one explicit Verse/provider target."
}
$cultMeshClientSource = Get-Content -LiteralPath (Join-Path $cultLibRoot "src\GameCult.Mesh\CultMeshClient.cs") -Raw
if ($cultMeshClientSource -match 'ConnectAsync\s*\(\s*string\s+(endpointId|verseId)' -or
$cultMeshClientSource -match 'Lease(Document|Collection)Async[^\(]*\(\s*string\s+(endpointId|verseId)') {
throw "CultMeshClient must not expose an ambiguous one-string session identity."
}
$discoverySource = Get-Content -LiteralPath (Join-Path $cultLibRoot "src\GameCult.Mesh\CultMeshDiscoveryService.cs") -Raw
if ($discoverySource -notmatch 'AuthorityRuntimeIds\.Contains\(query\.AuthorityRuntimeId') {
throw "CultMesh discovery must prove that a selected Verse route advertises the requested provider runtime."
}
$dotnetPackageRoot = Join-Path ([IO.Path]::GetTempPath()) `
("eve-getting-started-nuget-" + [guid]::NewGuid().ToString("N"))

8
src/GameCult.Mesh/CultMeshBodySessions.cs
View file

@ -278,15 +278,15 @@ namespace GameCult.Mesh
public sealed class CultMeshSessionBodyProvider
{
private readonly CultMeshSessionManager _sessions;
private readonly CultMeshEndpointId _endpointId;
private readonly CultMeshSessionTarget _target;
private readonly CultMeshSessionBodyProviderOptions _options;
public CultMeshSessionBodyProvider(string providerId, CultMeshSessionManager sessions,
CultMeshEndpointId endpointId, CultMeshSessionBodyProviderOptions? options = null)
CultMeshSessionTarget target, CultMeshSessionBodyProviderOptions? options = 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));
_target = target ?? throw new ArgumentNullException(nameof(target));
_options = options ?? new CultMeshSessionBodyProviderOptions();
if (_options.ResponseTimeout <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(options));
}
@ -300,7 +300,7 @@ namespace GameCult.Mesh
if (descriptor == null) throw new ArgumentNullException(nameof(descriptor));
if (descriptor.TransportKind != CultMeshBodyTransportKind.Network)
throw new NotSupportedException("Direct body provider requires a network descriptor.");
var session = await _sessions.ConnectAsync(_endpointId, CultMeshProtocols.Bodies, cancellationToken).ConfigureAwait(false);
var session = await _sessions.ConnectAsync(_target, CultMeshProtocols.Bodies, cancellationToken).ConfigureAwait(false);
var messageId = Guid.NewGuid().ToString("N");
var completion = new TaskCompletionSource<CultMeshBodyReadResponseMessage>(TaskCreationOptions.RunContinuationsAsynchronously);
using var responseSubscription = session.OnCultNet<CultMeshBodyReadResponseMessage>(response =>

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

@ -192,76 +192,66 @@ namespace GameCult.Mesh
_operationResponseTimeout = options.OperationResponseTimeout;
}
/// <summary>Connects to a stable endpoint identity using one application protocol.</summary>
/// <summary>Connects to one stable Verse/provider target using one application protocol.</summary>
public Task<CultMeshSession> ConnectAsync(
string endpointId,
CultMeshSessionTarget target,
CultMeshProtocolId protocol,
CancellationToken cancellationToken = default)
{
ThrowIfDisposed();
return _sessions.ConnectAsync(CultMeshEndpointId.Parse(endpointId), protocol, cancellationToken);
return _sessions.ConnectAsync(target, protocol, cancellationToken);
}
/// <summary>Connects the reusable realtime state plane by stable endpoint identity.</summary>
/// <summary>Connects the reusable realtime state plane for one stable Verse/provider target.</summary>
public Task<CultMeshRealtimeSession> ConnectRealtimeAsync(
string endpointId,
CultMeshSessionTarget target,
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,
CultMeshProtocolId protocol,
CancellationToken cancellationToken = default)
{
ThrowIfDisposed();
return _sessions.ConnectAsync(endpointId, protocol, cancellationToken);
return _sessions.ConnectRealtimeAsync(target, cancellationToken);
}
/// <summary>Creates a verified-transfer provider over this client's reusable content session owner.</summary>
public CultMeshSessionContentProvider ContentProvider(
string providerId,
string endpointId)
CultMeshSessionTarget target)
{
ThrowIfDisposed();
return new CultMeshSessionContentProvider(
providerId,
_sessions,
CultMeshEndpointId.Parse(endpointId));
target);
}
/// <summary>Creates a direct body provider over this client's reusable negotiated session owner.</summary>
public CultMeshSessionBodyProvider BodyProvider(
string providerId,
string endpointId,
CultMeshSessionTarget target,
CultMeshSessionBodyProviderOptions? options = null)
{
ThrowIfDisposed();
return new CultMeshSessionBodyProvider(
providerId,
_sessions,
CultMeshEndpointId.Parse(endpointId),
target,
options);
}
/// <summary>Leases one shared live typed document by stable provider identity and record key.</summary>
/// <summary>Leases one shared live typed document by stable Verse/provider identity and record key.</summary>
public async Task<CultMeshDocumentLease<TDocument>> LeaseDocumentAsync<TDocument>(
string endpointId,
CultMeshSessionTarget target,
string recordKey,
CancellationToken cancellationToken = default)
where TDocument : class
{
ThrowIfDisposed();
if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId));
if (target == null) throw new ArgumentNullException(nameof(target));
if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Record key is required.", nameof(recordKey));
var key = endpointId.Trim() + "\u001f" + typeof(TDocument).AssemblyQualifiedName + "\u001f" + recordKey.Trim();
var key = target.SessionKey + "\u001f" + typeof(TDocument).AssemblyQualifiedName + "\u001f" + recordKey.Trim();
while (true)
{
var resource = _documents.GetOrAdd(key, _ => new SharedResource(
async owner => await OpenDocumentAsync<TDocument>(endpointId.Trim(), recordKey.Trim(), owner).ConfigureAwait(false)));
async owner => await OpenDocumentAsync<TDocument>(target, recordKey.Trim(), owner).ConfigureAwait(false)));
if (!resource.TryAcquire())
{
RemoveExact(_documents, key, resource);
@ -283,28 +273,28 @@ namespace GameCult.Mesh
}
}
/// <summary>Leases a shared live collection of one typed schema by stable provider identity.</summary>
/// <summary>Leases a shared live collection by stable Verse/provider identity.</summary>
public Task<CultMeshCollectionLease<TDocument>> LeaseCollectionAsync<TDocument>(
string endpointId,
CultMeshSessionTarget target,
CancellationToken cancellationToken)
where TDocument : class =>
LeaseCollectionAsync<TDocument>(endpointId, includeInitialSnapshot: true, cancellationToken);
LeaseCollectionAsync<TDocument>(target, includeInitialSnapshot: true, cancellationToken);
/// <summary>Leases a live collection and optionally skips replaying its initial snapshot.</summary>
public async Task<CultMeshCollectionLease<TDocument>> LeaseCollectionAsync<TDocument>(
string endpointId,
CultMeshSessionTarget target,
bool includeInitialSnapshot = true,
CancellationToken cancellationToken = default)
where TDocument : class
{
ThrowIfDisposed();
if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId));
var identity = endpointId.Trim();
if (target == null) throw new ArgumentNullException(nameof(target));
var identity = target.SessionKey;
var key = identity + "\u001f" + typeof(TDocument).AssemblyQualifiedName + "\u001f" + includeInitialSnapshot;
while (true)
{
var resource = _collections.GetOrAdd(key, _ => new SharedResource(
async owner => await OpenCollectionAsync<TDocument>(identity, includeInitialSnapshot, owner).ConfigureAwait(false)));
async owner => await OpenCollectionAsync<TDocument>(target, includeInitialSnapshot, owner).ConfigureAwait(false)));
if (!resource.TryAcquire())
{
RemoveExact(_collections, key, resource);
@ -331,7 +321,7 @@ namespace GameCult.Mesh
/// only a provider-authored receipt or resulting state can establish acceptance.
/// </summary>
public async Task SubmitDocumentAsync<TDocument>(
string endpointId,
CultMeshSessionTarget target,
string recordKey,
TDocument document,
string sourceRuntimeId,
@ -340,7 +330,7 @@ namespace GameCult.Mesh
where TDocument : class
{
ThrowIfDisposed();
if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId));
if (target == null) throw new ArgumentNullException(nameof(target));
if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Record key is required.", nameof(recordKey));
if (document == null) throw new ArgumentNullException(nameof(document));
if (string.IsNullOrWhiteSpace(sourceRuntimeId)) throw new ArgumentException("Source runtime identity is required.", nameof(sourceRuntimeId));
@ -357,7 +347,7 @@ namespace GameCult.Mesh
SourceRuntimeId = sourceRuntimeId.Trim(),
SourceRole = sourceRole.Trim()
});
var session = await ConnectAsync(endpointId.Trim(), CultMeshProtocols.Documents, cancellationToken)
var session = await ConnectAsync(target, CultMeshProtocols.Documents, cancellationToken)
.ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
session.OpenSchemaClient().SendCultNet(message);
@ -369,7 +359,7 @@ namespace GameCult.Mesh
/// provider-authored evidence rather than client-side authority over canonical state.
/// </summary>
public async Task<CultMeshOperationResult<TResponse>> InvokeAsync<TRequest, TResponse>(
string endpointId,
CultMeshSessionTarget target,
string serviceId,
string operation,
string requestSchema,
@ -380,7 +370,7 @@ namespace GameCult.Mesh
CancellationToken cancellationToken = default)
{
ThrowIfDisposed();
if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId));
if (target == null) throw new ArgumentNullException(nameof(target));
if (string.IsNullOrWhiteSpace(serviceId)) throw new ArgumentException("Service identity is required.", nameof(serviceId));
if (string.IsNullOrWhiteSpace(operation)) throw new ArgumentException("Operation identity is required.", nameof(operation));
if (string.IsNullOrWhiteSpace(requestSchema)) throw new ArgumentException("Request schema is required.", nameof(requestSchema));
@ -389,12 +379,11 @@ namespace GameCult.Mesh
if (string.IsNullOrWhiteSpace(sourceRuntimeId)) throw new ArgumentException("Source runtime identity is required.", nameof(sourceRuntimeId));
if (string.IsNullOrWhiteSpace(idempotencyKey)) throw new ArgumentException("A durable idempotency key is required.", nameof(idempotencyKey));
var identity = endpointId.Trim();
var expectedService = serviceId.Trim();
var expectedOperation = operation.Trim();
var expectedResponseSchema = responseSchema.Trim();
var messageId = idempotencyKey.Trim();
var session = await ConnectAsync(identity, CultMeshProtocols.Documents, cancellationToken).ConfigureAwait(false);
var session = await ConnectAsync(target, CultMeshProtocols.Documents, cancellationToken).ConfigureAwait(false);
var completion = new TaskCompletionSource<CultNetOperationResponseMessage>(TaskCreationOptions.RunContinuationsAsynchronously);
using var subscription = session.OnCultNet<CultNetOperationResponseMessage>(response =>
{
@ -481,69 +470,69 @@ namespace GameCult.Mesh
/// <summary>Reads one typed record once through the reusable document session.</summary>
public async Task<TDocument> ReadAsync<TDocument>(
string endpointId,
CultMeshSessionTarget target,
string recordKey,
CancellationToken cancellationToken = default)
where TDocument : class
{
ThrowIfDisposed();
if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId));
if (target == null) throw new ArgumentNullException(nameof(target));
if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Record key is required.", nameof(recordKey));
var documents = await ReadManyAsync<TDocument>(endpointId, new[] { recordKey }, cancellationToken)
var documents = await ReadManyAsync<TDocument>(target, new[] { recordKey }, cancellationToken)
.ConfigureAwait(false);
return documents.FirstOrDefault()
?? throw new InvalidOperationException(
$"CultMesh endpoint '{endpointId}' did not publish {typeof(TDocument).FullName} record '{recordKey}'.");
$"CultMesh target '{target}' did not publish {typeof(TDocument).FullName} record '{recordKey}'.");
}
/// <summary>Reads one typed record once with an explicit response deadline.</summary>
public async Task<TDocument> ReadAsync<TDocument>(
string endpointId,
CultMeshSessionTarget target,
string recordKey,
TimeSpan responseTimeout,
CancellationToken cancellationToken = default)
where TDocument : class
{
var documents = await ReadManyCoreAsync<TDocument>(
endpointId,
target,
new[] { recordKey },
responseTimeout,
cancellationToken)
.ConfigureAwait(false);
return documents.FirstOrDefault()
?? throw new InvalidOperationException(
$"CultMesh endpoint '{endpointId}' did not publish {typeof(TDocument).FullName} record '{recordKey}'.");
$"CultMesh target '{target}' did not publish {typeof(TDocument).FullName} record '{recordKey}'.");
}
/// <summary>Reads typed records once in one request through the reusable document session.</summary>
public async Task<IReadOnlyList<TDocument>> ReadManyAsync<TDocument>(
string endpointId,
CultMeshSessionTarget target,
IReadOnlyList<string> recordKeys,
CancellationToken cancellationToken = default)
where TDocument : class
=> await ReadManyCoreAsync<TDocument>(
endpointId,
target,
recordKeys,
TimeSpan.FromSeconds(10),
cancellationToken)
.ConfigureAwait(false);
private async Task<IReadOnlyList<TDocument>> ReadManyCoreAsync<TDocument>(
string endpointId,
CultMeshSessionTarget target,
IReadOnlyList<string> recordKeys,
TimeSpan responseTimeout,
CancellationToken cancellationToken)
where TDocument : class
{
ThrowIfDisposed();
if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId));
if (target == null) throw new ArgumentNullException(nameof(target));
if (recordKeys == null || recordKeys.Count == 0 || recordKeys.Any(string.IsNullOrWhiteSpace))
throw new ArgumentException("At least one non-empty record key is required.", nameof(recordKeys));
var cacheRegistry = CultMesh.CreateCultCacheDocumentRegistry(typeof(TDocument));
var networkRegistry = CultMesh.CreateCultNetDocumentRegistry(new[] { typeof(TDocument) }, cacheRegistry);
using var snapshot = await CultMeshSnapshotSession.ConnectAsync(
_sessions,
CultMeshEndpointId.Parse(endpointId),
target,
CultMeshProtocols.Content,
new CultMeshSnapshotRequestOptions
{
@ -577,13 +566,13 @@ namespace GameCult.Mesh
}
private async Task<object> OpenDocumentAsync<TDocument>(
string endpointId,
CultMeshSessionTarget target,
string recordKey,
SharedResource owner)
where TDocument : class
{
var session = await ConnectAsync(endpointId, CultMeshProtocols.Documents).ConfigureAwait(false);
var binding = new RemoteDocumentBinding<TDocument>(session, endpointId, recordKey, _subscriptionResponseTimeout);
var session = await ConnectAsync(target, CultMeshProtocols.Documents).ConfigureAwait(false);
var binding = new RemoteDocumentBinding<TDocument>(session, target, recordKey, _subscriptionResponseTimeout);
owner.Attach(binding);
try
{
@ -599,13 +588,13 @@ namespace GameCult.Mesh
}
private async Task<object> OpenCollectionAsync<TDocument>(
string endpointId,
CultMeshSessionTarget target,
bool includeInitialSnapshot,
SharedResource owner)
where TDocument : class
{
var session = await ConnectAsync(endpointId, CultMeshProtocols.Documents).ConfigureAwait(false);
var binding = new RemoteCollectionBinding<TDocument>(session, endpointId, includeInitialSnapshot, _subscriptionResponseTimeout);
var session = await ConnectAsync(target, CultMeshProtocols.Documents).ConfigureAwait(false);
var binding = new RemoteCollectionBinding<TDocument>(session, target, includeInitialSnapshot, _subscriptionResponseTimeout);
owner.Attach(binding);
try
{
@ -743,14 +732,14 @@ namespace GameCult.Mesh
public RemoteDocumentBinding(
CultMeshSession session,
string endpointId,
CultMeshSessionTarget target,
string recordKey,
TimeSpan responseTimeout)
{
_session = session;
_responseTimeout = responseTimeout;
_recordKey = recordKey;
_subscriptionId = "cultmesh-document:" + endpointId + ":" + recordKey;
_subscriptionId = "cultmesh-document:" + target.SessionKey + ":" + recordKey;
var cacheRegistry = CultMesh.CreateCultCacheDocumentRegistry(typeof(TDocument));
var networkRegistry = CultMesh.CreateCultNetDocumentRegistry(new[] { typeof(TDocument) }, cacheRegistry);
_cache = new CultCache(cacheRegistry);
@ -758,8 +747,8 @@ namespace GameCult.Mesh
Handle = CultMesh.Document<TDocument>(
_cache,
new CultRecordKey(recordKey),
CultMesh.Verse(endpointId, "cultmesh-client").Context,
routeHint: new CultMeshRouteHint(CultMeshLocalityKind.Network, endpointId));
CultMesh.Verse(target.VerseId, "cultmesh-client").Context,
routeHint: new CultMeshRouteHint(CultMeshLocalityKind.Network, target.ProviderRuntimeId));
}
public CultMeshDocumentHandle<TDocument> Handle { get; }
@ -847,13 +836,13 @@ namespace GameCult.Mesh
public RemoteCollectionBinding(
CultMeshSession session,
string endpointId,
CultMeshSessionTarget target,
bool includeInitialSnapshot,
TimeSpan responseTimeout)
{
_session = session;
_responseTimeout = responseTimeout;
_subscriptionId = "cultmesh-collection:" + endpointId + ":" + typeof(TDocument).FullName;
_subscriptionId = "cultmesh-collection:" + target.SessionKey + ":" + typeof(TDocument).FullName;
_includeInitialSnapshot = includeInitialSnapshot;
var cacheRegistry = CultMesh.CreateCultCacheDocumentRegistry(typeof(TDocument));
var networkRegistry = CultMesh.CreateCultNetDocumentRegistry(new[] { typeof(TDocument) }, cacheRegistry);
@ -861,7 +850,7 @@ namespace GameCult.Mesh
_subscription = new CultNetDatabaseSubscriptionClient(session.OpenSchemaClient(), _cache, networkRegistry);
Handle = CultMesh.Collection<TDocument>(
_cache,
routeHint: new CultMeshRouteHint(CultMeshLocalityKind.Network, endpointId));
routeHint: new CultMeshRouteHint(CultMeshLocalityKind.Network, target.ProviderRuntimeId));
}
public CultMeshCollectionHandle<TDocument> Handle { get; }

14
src/GameCult.Mesh/CultMeshContentTransports.cs
View file

@ -52,18 +52,18 @@ namespace GameCult.Mesh
private bool _disposed;
internal CultMeshContentSession(
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
ICultMeshContentTransport transport,
CultMeshSessionState state,
Action<CultMeshContentSession> onTransportFailure)
{
EndpointId = endpointId ?? throw new ArgumentNullException(nameof(endpointId));
Target = target ?? throw new ArgumentNullException(nameof(target));
_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 CultMeshSessionTarget Target { get; }
public CultMeshSessionState State { get; private set; }
public string TransportId { get { lock (_gate) return _transport.TransportId; } }
@ -138,18 +138,18 @@ namespace GameCult.Mesh
public sealed class CultMeshSessionContentProvider : ICultMeshContentProvider
{
private readonly CultMeshSessionManager _sessions;
private readonly CultMeshEndpointId _endpointId;
private readonly CultMeshSessionTarget _target;
public CultMeshSessionContentProvider(
string providerId,
CultMeshSessionManager sessions,
CultMeshEndpointId endpointId)
CultMeshSessionTarget target)
{
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));
_target = target ?? throw new ArgumentNullException(nameof(target));
}
public string ProviderId { get; }
@ -159,7 +159,7 @@ namespace GameCult.Mesh
Stream destination,
CancellationToken cancellationToken = default)
{
var session = await _sessions.ConnectContentAsync(_endpointId, cancellationToken).ConfigureAwait(false);
var session = await _sessions.ConnectContentAsync(_target, cancellationToken).ConfigureAwait(false);
await session.CopyChunkToAsync(chunk, destination, cancellationToken).ConfigureAwait(false);
}
}

19
src/GameCult.Mesh/CultMeshDiscoveryService.cs
View file

@ -33,16 +33,25 @@ namespace GameCult.Mesh
public sealed class CultMeshDiscoveryQuery
{
public CultMeshDiscoveryQuery(string endpointId, IEnumerable<string>? verseIds = null)
/// <param name="endpointId">Stable lookup identity used for caching and diagnostics, never a physical route.</param>
/// <param name="verseIds">Optional Verse filter.</param>
/// <param name="authorityRuntimeId">Optional provider runtime that each accepted Verse must advertise.</param>
public CultMeshDiscoveryQuery(
string endpointId,
IEnumerable<string>? verseIds = null,
string? authorityRuntimeId = null)
{
EndpointId = Require(endpointId, nameof(endpointId));
VerseIds = verseIds?.Where(value => !string.IsNullOrWhiteSpace(value))
.Distinct(StringComparer.Ordinal).OrderBy(value => value, StringComparer.Ordinal).ToArray()
?? Array.Empty<string>();
AuthorityRuntimeId = string.IsNullOrWhiteSpace(authorityRuntimeId) ? null : authorityRuntimeId.Trim();
}
public string EndpointId { get; }
public IReadOnlyList<string> VerseIds { get; }
/// <summary>Gets the optional provider runtime constraint for authority-sensitive resolution.</summary>
public string? AuthorityRuntimeId { get; }
private static string Require(string value, string name) =>
string.IsNullOrWhiteSpace(value) ? throw new ArgumentException("Value must be non-empty.", name) : value;
@ -219,9 +228,11 @@ namespace GameCult.Mesh
}
private static string QueryKey(CultMeshDiscoveryQuery query) =>
query.VerseIds.Count == 0
query.AuthorityRuntimeId == null && query.VerseIds.Count == 0
? query.EndpointId
: query.EndpointId + "\u001f" + string.Join("\u001f", query.VerseIds);
: string.Join(
"\u001f",
new[] { query.EndpointId, query.AuthorityRuntimeId ?? "" }.Concat(query.VerseIds));
private static async Task<T> AwaitForCallerAsync<T>(Task<T> shared, CancellationToken cancellationToken)
{
@ -255,6 +266,8 @@ namespace GameCult.Mesh
var failures = results.Where(result => result.Error != null).Select(result => result.SourceId).ToArray();
var observations = results.SelectMany(result => result.Observations)
.Where(observation => query.VerseIds.Count == 0 || query.VerseIds.Contains(observation.Candidate.VerseId, StringComparer.Ordinal))
.Where(observation => query.AuthorityRuntimeId == null ||
observation.Candidate.AuthorityRuntimeIds.Contains(query.AuthorityRuntimeId, StringComparer.Ordinal))
.ToArray();
foreach (var rejected in observations.Where(observation => observation.Trust == CultMeshDiscoveryTrust.Rejected))

14
src/GameCult.Mesh/CultMeshPeerExchange.cs
View file

@ -446,15 +446,17 @@ namespace GameCult.Mesh
/// <summary>Fetches peer cards through the shared session owner.</summary>
public async Task<CultMeshPeerExchangeResponseMessage> FetchAsync(
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
CultMeshPeerExchangeRequestMessage request,
CancellationToken cancellationToken = default)
{
if (endpointId == null) throw new ArgumentNullException(nameof(endpointId));
if (target == null) throw new ArgumentNullException(nameof(target));
if (request == null) throw new ArgumentNullException(nameof(request));
if (string.IsNullOrWhiteSpace(request.VerseId)) throw new ArgumentException("Peer exchange requires a verseId.", nameof(request));
if (!string.Equals(request.VerseId, target.VerseId, StringComparison.Ordinal))
throw new ArgumentException("Peer exchange request Verse must match its session target.", nameof(request));
var sessions = _options.Sessions ?? throw new InvalidOperationException("Managed peer exchange requires a CultMeshSessionManager.");
var session = await sessions.ConnectAsync(endpointId, CultMeshProtocols.PeerExchange, cancellationToken).ConfigureAwait(false);
var session = await sessions.ConnectAsync(target, CultMeshProtocols.PeerExchange, cancellationToken).ConfigureAwait(false);
var messageId = Guid.NewGuid().ToString("N");
var completion = new TaskCompletionSource<CultMeshPeerExchangeResponseMessage>(TaskCreationOptions.RunContinuationsAsynchronously);
using var responseSubscription = session.OnCultNet<CultMeshPeerExchangeResponseMessage>(response =>
@ -471,7 +473,7 @@ namespace GameCult.Mesh
KnownPeerIds = request.KnownPeerIds,
Limit = request.Limit
});
return await WaitForManagedResponseAsync(completion.Task, endpointId, cancellationToken).ConfigureAwait(false);
return await WaitForManagedResponseAsync(completion.Task, target, cancellationToken).ConfigureAwait(false);
}
/// <summary>
@ -541,7 +543,7 @@ namespace GameCult.Mesh
private async Task<CultMeshPeerExchangeResponseMessage> WaitForManagedResponseAsync(
Task<CultMeshPeerExchangeResponseMessage> responseTask,
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
CancellationToken cancellationToken)
{
var timeout = _options.Clock.DelayAsync(_options.ResponseTimeout, cancellationToken);
@ -549,7 +551,7 @@ namespace GameCult.Mesh
if (completed != responseTask)
{
cancellationToken.ThrowIfCancellationRequested();
throw new TimeoutException($"Timed out waiting for peer exchange response from {endpointId}.");
throw new TimeoutException($"Timed out waiting for peer exchange response from {target}.");
}
return await responseTask.ConfigureAwait(false);
}

6
src/GameCult.Mesh/CultMeshRealtimeTransports.cs
View file

@ -72,18 +72,18 @@ namespace GameCult.Mesh
private bool _disposed;
internal CultMeshRealtimeSession(
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
ICultMeshRealtimeTransport transport,
CultMeshSessionState state,
Action<CultMeshRealtimeSession> onTransportFailure)
{
EndpointId = endpointId ?? throw new ArgumentNullException(nameof(endpointId));
Target = target ?? throw new ArgumentNullException(nameof(target));
_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 CultMeshSessionTarget Target { get; }
public CultMeshSessionState State { get; private set; }
public string TransportId { get { lock (_gate) return _transport.TransportId; } }

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

@ -21,6 +21,40 @@ namespace GameCult.Mesh
public override string ToString() => Value;
}
/// <summary>
/// Identifies one provider inside one Verse. Odin resolves this stable pair to
/// disposable physical routes; neither value is a socket address.
/// </summary>
public sealed class CultMeshSessionTarget : IEquatable<CultMeshSessionTarget>
{
/// <summary>Creates one stable provider target inside one Verse.</summary>
public CultMeshSessionTarget(string verseId, string providerRuntimeId)
{
VerseId = string.IsNullOrWhiteSpace(verseId)
? throw new ArgumentException("Verse identity is required.", nameof(verseId))
: verseId.Trim();
ProviderEndpointId = CultMeshEndpointId.Parse(providerRuntimeId);
ProviderRuntimeId = ProviderEndpointId.Value;
}
/// <summary>Gets the stable Verse identity.</summary>
public string VerseId { get; }
/// <summary>Gets the stable provider runtime identity advertised by that Verse.</summary>
public string ProviderRuntimeId { get; }
internal CultMeshEndpointId ProviderEndpointId { get; }
internal string SessionKey => VerseId + "\u001f" + ProviderRuntimeId;
public bool Equals(CultMeshSessionTarget? other) =>
other != null &&
string.Equals(VerseId, other.VerseId, StringComparison.Ordinal) &&
string.Equals(ProviderRuntimeId, other.ProviderRuntimeId, StringComparison.Ordinal);
public override bool Equals(object? obj) => Equals(obj as CultMeshSessionTarget);
public override int GetHashCode() => HashCode.Combine(
StringComparer.Ordinal.GetHashCode(VerseId),
StringComparer.Ordinal.GetHashCode(ProviderRuntimeId));
public override string ToString() => VerseId + "/" + ProviderRuntimeId;
}
public sealed class CultMeshProtocolId : IEquatable<CultMeshProtocolId>
{
public CultMeshProtocolId(string value) { Value = string.IsNullOrWhiteSpace(value) ? throw new ArgumentException("Protocol identity is required.", nameof(value)) : value; }
@ -285,14 +319,14 @@ namespace GameCult.Mesh
private readonly ManagedSchemaChannel _channel;
private long _physicalGeneration;
private bool _disposed;
internal CultMeshSession(CultMeshEndpointId endpointId, CultMeshProtocolId protocol, ICultNetSchemaClient channel, CultMeshSessionState state)
internal CultMeshSession(CultMeshSessionTarget target, CultMeshProtocolId protocol, ICultNetSchemaClient channel, CultMeshSessionState state)
{
EndpointId = endpointId;
Target = target;
Protocol = protocol;
_channel = new ManagedSchemaChannel(channel);
State = state;
}
public CultMeshEndpointId EndpointId { get; }
public CultMeshSessionTarget Target { get; }
public CultMeshProtocolId Protocol { get; }
internal ICultNetSchemaClient Channel => _channel;
public CultMeshSessionState State { get; private set; }
@ -550,18 +584,18 @@ namespace GameCult.Mesh
}
public async Task<CultMeshSession> ConnectAsync(
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
CultMeshProtocolId protocol,
CancellationToken cancellationToken = default)
{
if (endpointId == null) throw new ArgumentNullException(nameof(endpointId));
if (target == null) throw new ArgumentNullException(nameof(target));
if (protocol == null) throw new ArgumentNullException(nameof(protocol));
ThrowIfDisposed();
var key = endpointId.Value + "\u001f" + protocol.Value;
var key = target.SessionKey + "\u001f" + protocol.Value;
while (true)
{
var lazy = _sessions.GetOrAdd(key, _ => new Lazy<Task<CultMeshSession>>(
() => ConnectOwnedAsync(key, endpointId, protocol), LazyThreadSafetyMode.ExecutionAndPublication));
() => ConnectOwnedAsync(key, target, protocol), LazyThreadSafetyMode.ExecutionAndPublication));
try
{
var session = await AwaitForCallerAsync(lazy.Value, cancellationToken).ConfigureAwait(false);
@ -577,20 +611,20 @@ namespace GameCult.Mesh
}
/// <summary>
/// Connects one reusable streaming content plane by stable endpoint identity.
/// Connects one reusable streaming content plane by stable Verse/provider identity.
/// Preferred connector tiers are exhausted before legacy fallback tiers are attempted.
/// </summary>
public async Task<CultMeshContentSession> ConnectContentAsync(
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
CancellationToken cancellationToken = default)
{
if (endpointId == null) throw new ArgumentNullException(nameof(endpointId));
if (target == null) throw new ArgumentNullException(nameof(target));
ThrowIfDisposed();
var key = endpointId.Value;
var key = target.SessionKey;
while (true)
{
var lazy = _contentSessions.GetOrAdd(key, _ => new Lazy<Task<CultMeshContentSession>>(
() => ConnectContentOwnedAsync(key, endpointId), LazyThreadSafetyMode.ExecutionAndPublication));
() => ConnectContentOwnedAsync(key, target), LazyThreadSafetyMode.ExecutionAndPublication));
try
{
var session = await AwaitForCallerAsync(lazy.Value, cancellationToken).ConfigureAwait(false);
@ -607,19 +641,19 @@ namespace GameCult.Mesh
private async Task<CultMeshContentSession> ConnectContentOwnedAsync(
string key,
CultMeshEndpointId endpointId)
CultMeshSessionTarget target)
{
var result = await ResolveContentPathAsync(endpointId).ConfigureAwait(false);
var result = await ResolveContentPathAsync(target).ConfigureAwait(false);
var session = new CultMeshContentSession(
endpointId,
target,
result.Transport,
new CultMeshSessionState(CultMeshSessionStatus.Online, _options.Clock.UtcNow, result.Candidate),
failed => InvalidateContentSession(key, failed));
Emit(endpointId, CultMeshProtocols.Content, "online", result.Candidate.Endpoint);
Emit(target, CultMeshProtocols.Content, "online", result.Candidate.Endpoint);
return session;
}
private async Task<ConnectedContentPath> ResolveContentPathAsync(CultMeshEndpointId endpointId)
private async Task<ConnectedContentPath> ResolveContentPathAsync(CultMeshSessionTarget target)
{
if (_contentConnectors.Length == 0)
throw Failure(
@ -627,8 +661,9 @@ namespace GameCult.Mesh
"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,
new[] { endpointId.Value })).ConfigureAwait(false);
target.ProviderRuntimeId,
new[] { target.VerseId },
target.ProviderRuntimeId)).ConfigureAwait(false);
var candidates = discovery.Candidates
.SelectMany(candidate => candidate.Descriptor.DiscoveryEndpoints)
.Distinct(StringComparer.Ordinal)
@ -644,13 +679,13 @@ namespace GameCult.Mesh
if (routes.Length == 0)
throw Failure(
CultMeshSessionFailureReason.UnsupportedPath,
$"No streaming content connector supports an advertised route for '{endpointId}'.");
$"No streaming content connector supports an advertised route for '{target}'.");
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))
.Select(route => ConnectContentCandidateAsync(target.ProviderEndpointId, route))
.ToList();
while (attempts.Count > 0)
{
@ -676,7 +711,7 @@ namespace GameCult.Mesh
var last = failures.LastOrDefault();
throw Failure(
last is TimeoutException ? CultMeshSessionFailureReason.Timeout : CultMeshSessionFailureReason.Transport,
$"No streaming content path connected for '{endpointId}'.",
$"No streaming content path connected for '{target}'.",
last);
}
@ -701,7 +736,7 @@ namespace GameCult.Mesh
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);
Emit(session.Target, CultMeshProtocols.Content, "offline", session.State.Path?.Endpoint ?? string.Empty);
}
/// <summary>
@ -709,16 +744,16 @@ namespace GameCult.Mesh
/// register QUIC explicitly, while RUDP compatibility remains opt-in.
/// </summary>
public async Task<CultMeshRealtimeSession> ConnectRealtimeAsync(
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
CancellationToken cancellationToken = default)
{
if (endpointId == null) throw new ArgumentNullException(nameof(endpointId));
if (target == null) throw new ArgumentNullException(nameof(target));
ThrowIfDisposed();
var key = endpointId.Value;
var key = target.SessionKey;
while (true)
{
var lazy = _realtimeSessions.GetOrAdd(key, _ => new Lazy<Task<CultMeshRealtimeSession>>(
() => ConnectRealtimeOwnedAsync(key, endpointId), LazyThreadSafetyMode.ExecutionAndPublication));
() => ConnectRealtimeOwnedAsync(key, target), LazyThreadSafetyMode.ExecutionAndPublication));
try
{
var session = await AwaitForCallerAsync(lazy.Value, cancellationToken).ConfigureAwait(false);
@ -735,19 +770,19 @@ namespace GameCult.Mesh
private async Task<CultMeshRealtimeSession> ConnectRealtimeOwnedAsync(
string key,
CultMeshEndpointId endpointId)
CultMeshSessionTarget target)
{
var result = await ResolveRealtimePathAsync(endpointId).ConfigureAwait(false);
var result = await ResolveRealtimePathAsync(target).ConfigureAwait(false);
var session = new CultMeshRealtimeSession(
endpointId,
target,
result.Transport,
new CultMeshSessionState(CultMeshSessionStatus.Online, _options.Clock.UtcNow, result.Candidate),
failed => InvalidateRealtimeSession(key, failed));
Emit(endpointId, CultMeshProtocols.RealtimeState, "online", result.Candidate.Endpoint);
Emit(target, CultMeshProtocols.RealtimeState, "online", result.Candidate.Endpoint);
return session;
}
private async Task<ConnectedRealtimePath> ResolveRealtimePathAsync(CultMeshEndpointId endpointId)
private async Task<ConnectedRealtimePath> ResolveRealtimePathAsync(CultMeshSessionTarget target)
{
if (_realtimeConnectors.Length == 0)
throw Failure(
@ -755,8 +790,9 @@ namespace GameCult.Mesh
"No realtime state connectors are configured. Register a QUIC connector explicitly.");
var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery(
endpointId.Value,
new[] { endpointId.Value })).ConfigureAwait(false);
target.ProviderRuntimeId,
new[] { target.VerseId },
target.ProviderRuntimeId)).ConfigureAwait(false);
var routes = discovery.Candidates
.SelectMany(candidate => candidate.Descriptor.DiscoveryEndpoints)
.Distinct(StringComparer.Ordinal)
@ -770,13 +806,13 @@ namespace GameCult.Mesh
if (routes.Length == 0)
throw Failure(
CultMeshSessionFailureReason.UnsupportedPath,
$"No realtime connector supports an advertised route for '{endpointId}'.");
$"No realtime connector supports an advertised route for '{target}'.");
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))
.Select(route => ConnectRealtimeCandidateAsync(target.ProviderEndpointId, route))
.ToList();
while (attempts.Count > 0)
{
@ -802,7 +838,7 @@ namespace GameCult.Mesh
var last = failures.LastOrDefault();
throw Failure(
last is TimeoutException ? CultMeshSessionFailureReason.Timeout : CultMeshSessionFailureReason.Transport,
$"No realtime state path connected for '{endpointId}'.",
$"No realtime state path connected for '{target}'.",
last);
}
@ -828,24 +864,25 @@ namespace GameCult.Mesh
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);
Emit(session.Target, CultMeshProtocols.RealtimeState, "offline", endpoint);
}
private async Task<CultMeshSession> ConnectOwnedAsync(string key, CultMeshEndpointId endpointId, CultMeshProtocolId protocol)
private async Task<CultMeshSession> ConnectOwnedAsync(string key, CultMeshSessionTarget target, CultMeshProtocolId protocol)
{
var result = await ResolveConnectedPathAsync(endpointId, protocol).ConfigureAwait(false);
var session = new CultMeshSession(endpointId, protocol, result.Client,
var result = await ResolveConnectedPathAsync(target, protocol).ConfigureAwait(false);
var session = new CultMeshSession(target, protocol, result.Client,
new CultMeshSessionState(CultMeshSessionStatus.Online, _options.Clock.UtcNow, result.Candidate));
ObservePhysicalFailure(key, session, result);
Emit(endpointId, protocol, "online", result.Candidate.Endpoint);
Emit(target, protocol, "online", result.Candidate.Endpoint);
return session;
}
private async Task<ConnectedPath> ResolveConnectedPathAsync(CultMeshEndpointId endpointId, CultMeshProtocolId protocol)
private async Task<ConnectedPath> ResolveConnectedPathAsync(CultMeshSessionTarget target, CultMeshProtocolId protocol)
{
var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery(
endpointId.Value,
new[] { endpointId.Value })).ConfigureAwait(false);
target.ProviderRuntimeId,
new[] { target.VerseId },
target.ProviderRuntimeId)).ConfigureAwait(false);
var candidates = discovery.Candidates.SelectMany(candidate => candidate.Descriptor.DiscoveryEndpoints)
.Distinct(StringComparer.Ordinal)
.Select(endpoint => new CultMeshTransportCandidate(endpoint))
@ -858,14 +895,14 @@ namespace GameCult.Mesh
.ThenBy(route => route.Candidate.Priority)
.ToArray();
if (candidates.Length == 0)
throw Failure(CultMeshSessionFailureReason.Resolution, $"No route candidates for '{endpointId}'.");
throw Failure(CultMeshSessionFailureReason.Resolution, $"No route candidates for '{target}'.");
if (routes.Length == 0)
throw Failure(
CultMeshSessionFailureReason.Transport,
$"No transport path connected for '{endpointId}' and protocol '{protocol}'.",
$"No transport path connected for '{target}' and protocol '{protocol}'.",
Failure(
CultMeshSessionFailureReason.UnsupportedPath,
$"No registered connector supports an advertised route for '{endpointId}'."));
$"No registered connector supports an advertised route for '{target}'."));
var failures = new List<Exception>();
foreach (var tier in routes.GroupBy(route => route.Connector.Priority).OrderBy(group => group.Key))
@ -889,7 +926,7 @@ namespace GameCult.Mesh
}
var last = failures.LastOrDefault();
throw Failure(last is TimeoutException ? CultMeshSessionFailureReason.Timeout : CultMeshSessionFailureReason.Transport,
$"No transport path connected for '{endpointId}' and protocol '{protocol}'.", last);
$"No transport path connected for '{target}' and protocol '{protocol}'.", last);
}
private void ObservePhysicalFailure(string key, CultMeshSession session, ConnectedPath path)
@ -919,20 +956,21 @@ namespace GameCult.Mesh
_options.Clock.UtcNow,
failedPath.Candidate,
failure));
Emit(session.EndpointId, session.Protocol, "reconnecting", failedPath.Candidate.Endpoint);
Emit(session.Target, session.Protocol, "reconnecting", failedPath.Candidate.Endpoint);
try
{
_discovery.Invalidate(new CultMeshDiscoveryQuery(
session.EndpointId.Value,
new[] { session.EndpointId.Value }));
var replacement = await ResolveConnectedPathAsync(session.EndpointId, session.Protocol).ConfigureAwait(false);
session.Target.ProviderRuntimeId,
new[] { session.Target.VerseId },
session.Target.ProviderRuntimeId));
var replacement = await ResolveConnectedPathAsync(session.Target, session.Protocol).ConfigureAwait(false);
session.ReplacePhysicalChannel(replacement.Client);
session.Transition(new CultMeshSessionState(
CultMeshSessionStatus.Online,
_options.Clock.UtcNow,
replacement.Candidate));
ObservePhysicalFailure(key, session, replacement);
Emit(session.EndpointId, session.Protocol, "online", replacement.Candidate.Endpoint);
Emit(session.Target, session.Protocol, "online", replacement.Candidate.Endpoint);
}
catch (Exception reconnectError)
{
@ -945,7 +983,7 @@ namespace GameCult.Mesh
reconnectError.Message,
failedPath.Candidate.Endpoint)));
_sessions.TryRemove(key, out _);
Emit(session.EndpointId, session.Protocol, "offline", failedPath.Candidate.Endpoint);
Emit(session.Target, session.Protocol, "offline", failedPath.Candidate.Endpoint);
}
}
@ -970,12 +1008,12 @@ namespace GameCult.Mesh
private CultMeshSessionException Failure(CultMeshSessionFailureReason reason, string message, Exception? inner = null) =>
new CultMeshSessionException(new CultMeshSessionFailure(reason, message), inner);
private void Emit(CultMeshEndpointId endpoint, CultMeshProtocolId protocol, string state, string path)
private void Emit(CultMeshSessionTarget target, CultMeshProtocolId protocol, string state, string path)
{
_options.Diagnostics.Emit(new CultMeshDiagnosticEvent(
Interlocked.Increment(ref _diagnosticSequence), _options.Clock.UtcNow,
CultMeshReliabilityOrgan.Session, CultMeshDiagnosticKind.PathChanged,
endpoint.Value + ":" + protocol.Value, endpoint.Value, state,
target.SessionKey + ":" + protocol.Value, target.ProviderRuntimeId, state,
sourceId: "session-manager", endpoint: path));
}

10
src/GameCult.Mesh/CultMeshSnapshots.cs
View file

@ -102,7 +102,7 @@ namespace GameCult.Mesh
CultNetDocumentRegistry? registry)
{
_sharedSession = session ?? throw new ArgumentNullException(nameof(session));
_endpoint = session.State.Path?.Endpoint ?? session.EndpointId.Value;
_endpoint = session.State.Path?.Endpoint ?? session.Target.ProviderRuntimeId;
_defaults = CultMesh.CloneSnapshotRequestOptions(options);
_registry = registry ?? new CultNetDocumentRegistry();
_client = session.Channel;
@ -112,26 +112,26 @@ namespace GameCult.Mesh
public static async Task<CultMeshSnapshotSession> ConnectAsync(
CultMeshSessionManager sessions,
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
CultMeshSnapshotRequestOptions options,
CultNetDocumentRegistry? registry = null,
CancellationToken cancellationToken = default)
{
if (sessions == null) throw new ArgumentNullException(nameof(sessions));
var session = await sessions.ConnectAsync(endpointId, CultMeshProtocols.Documents, cancellationToken).ConfigureAwait(false);
var session = await sessions.ConnectAsync(target, CultMeshProtocols.Documents, cancellationToken).ConfigureAwait(false);
return new CultMeshSnapshotSession(session, options ?? new CultMeshSnapshotRequestOptions(), registry);
}
public static async Task<CultMeshSnapshotSession> ConnectAsync(
CultMeshSessionManager sessions,
CultMeshEndpointId endpointId,
CultMeshSessionTarget target,
CultMeshProtocolId protocol,
CultMeshSnapshotRequestOptions options,
CultNetDocumentRegistry? registry = null,
CancellationToken cancellationToken = default)
{
if (sessions == null) throw new ArgumentNullException(nameof(sessions));
var session = await sessions.ConnectAsync(endpointId, protocol, cancellationToken).ConfigureAwait(false);
var session = await sessions.ConnectAsync(target, protocol, cancellationToken).ConfigureAwait(false);
return new CultMeshSnapshotSession(session, options ?? new CultMeshSnapshotRequestOptions(), registry);
}

5
src/GameCult.Mesh/docs/content-sessions.md
View file

@ -15,9 +15,12 @@ The consumer creates a provider over its existing identity-first session
manager, then gives that provider to the transfer owner:
```csharp
var target = new CultMeshSessionTarget(
"aetheria.public",
"service:aetheria.daemon");
var provider = mesh.ContentProvider(
"aetheria.daemon",
"service:aetheria.daemon");
target);
var transfer = new CultMeshContentTransferService(
transferStateCache,

23
src/GameCult.Mesh/docs/getting-started/02-connect-by-identity.md
View file

@ -1,21 +1,28 @@
# 2. Connect A Client By Stable Identity
Clients address a Verse session by stable identity. In this tutorial,
`sample.counter-provider` is the sole authority runtime Odin advertises for the
`sample.counter` Verse; neither identity is a socket address. Discovery owns
current physical candidates and the session manager owns connection reuse,
path rotation, and transport failure state.
Clients address a live session by stable Verse/provider identity. In this tutorial,
`sample.counter` is the Verse id and `sample.counter-provider` is the authority
runtime Odin currently advertises for it. Application clients pass that typed
pair, never a socket address and never one ambiguous string that might mean
either identity. A product may expose only a Verse choice to its user; its
selection owner then resolves the provider runtime from Odin's descriptor before
opening the session. Discovery owns the current physical routes, and the session
manager owns connection reuse, path rotation, and transport failure state.
```csharp
using GameCult.Mesh;
var odinEndpoint = Environment.GetEnvironmentVariable("ODIN_ENDPOINT")
?? throw new InvalidOperationException("Set ODIN_ENDPOINT to the configured rendezvous route.");
const string verseId = "sample.counter";
var target = new CultMeshSessionTarget(verseId, "sample.counter-provider");
using var mesh = new CultMeshClient(new CultMeshClientOptions
{
RendezvousEndpoints = new[] { "rudp://odin.gamecult.net:3076" }
RendezvousEndpoints = new[] { odinEndpoint }
});
using var counterLease = await mesh.LeaseDocumentAsync<CounterState>(
"sample.counter",
target,
"counter:main",
cancellationToken);
var counter = counterLease.Handle;
@ -40,7 +47,7 @@ protocol-level state or diagnostics:
```csharp
var session = await mesh.ConnectAsync(
"aetheria.daemon",
target,
CultMeshProtocols.Documents,
cancellationToken);

4
src/GameCult.Mesh/docs/getting-started/03-publish-an-eve-surface.md
View file

@ -118,8 +118,8 @@ survive reopening the `.cc` store.
The second layer runs a durable C# provider, a separate local Odin fixture, a
real Chromium Eve lowerer, and an independent C# headless observer over binary
CultNet WebSocket lanes. The open browser resolves provider identity through
the canonical Verse catalog, survives a provider restart on a different route,
CultNet WebSocket lanes. The open browser resolves the `sample.counter` Verse
through the canonical catalog, survives its provider restarting on a different route,
and resubscribes. The retained C# client then invokes a second typed operation
through `CultMeshClient.InvokeAsync`; Chromium and C# observe the same
provider-authored state and receipt chronology. The same retained C# session

4
src/GameCult.Mesh/docs/getting-started/04-run-the-multiruntime-game.md
View file

@ -8,8 +8,8 @@ The counter is now one small multiplayer game:
invokes Increment;
- an independent C# client leases the same counter and invokes through the same
typed operation boundary;
- both clients remain attached when Odin moves the provider identity to a new
physical route.
- both clients remain attached when Odin moves the Verse to a replacement
physical provider route.
Run the whole chronology from the CultLib checkout:

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

@ -101,8 +101,9 @@ state for inspection and compatibility.
**Owner:** `CultMeshSessionManager`.
**Inputs:** stable endpoint identity, application protocol identity, discovery
candidate stream, transport connectors, path policy, clock, and retry policy.
**Inputs:** stable Verse/provider identity, application protocol identity,
discovery candidate stream, transport connectors, path policy, clock, and retry
policy.
**Outputs:** a reusable `CultMeshSession` with typed connection state, chosen
path, protocol channel, diagnostics, and watches for state/path changes.
@ -249,7 +250,7 @@ The target API is identity-first and watchable:
```csharp
await using var session = await mesh.ConnectAsync(
CultMeshEndpointId.Parse("odin:yggdrasil"),
new CultMeshSessionTarget("yggdrasil.public", "odin:yggdrasil"),
CultMeshProtocols.Documents,
cancellationToken);

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

@ -9,9 +9,9 @@ be equally good at files, realtime state, and same-machine memory access.
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.
**Inputs:** stable Verse/provider identity, advertised physical candidates,
connector support and priority, caller cancellation, and locality evidence for
body negotiation.
**Outputs:**

12
tests/GameCult.Mesh.Quic.Tests/CultMeshQuicRealtimeTransportTests.cs
View file

@ -79,11 +79,12 @@ public sealed class CultMeshQuicRealtimeTransportTests
ServerCertificate = certificate
});
var endpointId = CultMeshEndpointId.Parse("service:aetheria.daemon");
var target = new CultMeshSessionTarget("aetheria", endpointId.Value);
using var sessions = CreateSessions(server, out var validatedIdentity);
CultMeshRealtimeSession session;
try
{
session = await sessions.ConnectRealtimeAsync(endpointId);
session = await sessions.ConnectRealtimeAsync(target);
}
catch when (server.BackgroundFailure.IsCompleted)
{
@ -118,8 +119,9 @@ public sealed class CultMeshQuicRealtimeTransportTests
ServerCertificate = certificate
});
var endpointId = CultMeshEndpointId.Parse("service:aetheria.daemon");
var target = new CultMeshSessionTarget("aetheria", endpointId.Value);
using var sessions = CreateSessions(server, out _);
var session = await sessions.ConnectRealtimeAsync(endpointId);
var session = await sessions.ConnectRealtimeAsync(target);
await WaitUntilAsync(() => server.ConnectionCount == 1);
await server.BroadcastAsync(Frame(100, CultMeshRealtimeDelivery.LatestOnly, producerEpoch: 7));
@ -147,8 +149,9 @@ public sealed class CultMeshQuicRealtimeTransportTests
ServerCertificate = certificate
});
var endpointId = CultMeshEndpointId.Parse("service:aetheria.daemon");
var target = new CultMeshSessionTarget("aetheria", endpointId.Value);
using var sessions = CreateSessions(server, out _);
var session = await sessions.ConnectRealtimeAsync(endpointId);
var session = await sessions.ConnectRealtimeAsync(target);
Func<Task> send = () => session.SendAsync(Frame(1, CultMeshRealtimeDelivery.Unreliable));
@ -244,7 +247,8 @@ public sealed class CultMeshQuicRealtimeTransportTests
"Aetheria",
CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v1", "test"),
new[] { endpoint }),
new[] { endpoint },
new[] { "service:aetheria.daemon" }),
SourceId,
now,
now.AddMinutes(1),

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

@ -28,7 +28,7 @@ public sealed class CultMeshBodySessionTests
using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource() });
using var sessions = new CultMeshSessionManager(discovery, new[] { connector });
var provider = new CultMeshSessionBodyProvider(
"aetheria.daemon", sessions, CultMeshEndpointId.Parse("aetheria"));
"aetheria.daemon", sessions, new CultMeshSessionTarget("aetheria", "aetheria.daemon"));
var received = await provider.FetchAsync(descriptor);
@ -69,7 +69,7 @@ public sealed class CultMeshBodySessionTests
using var sessions = new CultMeshSessionManager(discovery,
new ICultMeshTransportConnector[] { new CultMeshSchemaTransportConnector() });
var provider = new CultMeshSessionBodyProvider("aetheria.daemon", sessions,
CultMeshEndpointId.Parse("aetheria"),
new CultMeshSessionTarget("aetheria", "aetheria.daemon"),
new CultMeshSessionBodyProviderOptions { ResponseTimeout = TimeSpan.FromSeconds(10) });
var elapsed = Stopwatch.StartNew();
@ -146,7 +146,7 @@ public sealed class CultMeshBodySessionTests
{
new CultMeshDiscoveryObservation(new CultMeshVerseDescriptor("aetheria", "Aetheria",
CultMeshVerseAuthorityModel.OperatorCluster, new CultMeshVerseCompatibility("cultmesh.v1", "rules"),
new[] { _endpoint }), SourceId, now, now.AddMinutes(1), CultMeshDiscoveryTrust.Signed)
new[] { _endpoint }, new[] { "aetheria.daemon" }), SourceId, now, now.AddMinutes(1), CultMeshDiscoveryTrust.Signed)
});
}
}

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

@ -13,6 +13,8 @@ namespace GameCult.Mesh.Tests;
public sealed class CultMeshClientTests
{
private static CultMeshSessionTarget Target => new("aetheria", "aetheria-daemon");
[Test]
public void Construction_RequiresOneRendezvousEndpoint()
{
@ -41,16 +43,37 @@ public sealed class CultMeshClientTests
Connectors = new[] { connector }
});
var first = await mesh.ConnectAsync("aetheria", CultMeshProtocols.Documents);
var second = await mesh.ConnectAsync("aetheria", CultMeshProtocols.Documents);
var first = await mesh.ConnectAsync(Target, CultMeshProtocols.Documents);
var second = await mesh.ConnectAsync(Target, CultMeshProtocols.Documents);
first.Should().BeSameAs(second);
first.EndpointId.Value.Should().Be("aetheria");
first.Target.Should().Be(Target);
first.State.Path!.Endpoint.Should().Be("rudp://aetheria:3081");
discoveryClients.Should().Be(1);
connector.ConnectCount.Should().Be(1);
}
[Test]
public async Task Connect_RejectsProviderNotAdvertisedBySelectedVerseBeforeTransport()
{
var connector = new RecordingConnector();
using var mesh = new CultMeshClient(new CultMeshClientOptions
{
RendezvousEndpoints = new[] { "rudp://odin:3076" },
Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
Connectors = new[] { connector }
});
Func<Task> connect = async () => await mesh.ConnectAsync(
new CultMeshSessionTarget("aetheria", "forged-provider"),
CultMeshProtocols.Documents);
var failure = await connect.Should().ThrowAsync<CultMeshSessionException>();
failure.Which.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Resolution);
connector.ConnectCount.Should().Be(0,
"a valid Verse route must not authorize an unadvertised provider identity");
}
[Test]
public async Task Document_ReplicatesTypedRecordBehindOneIntentDeclaration()
{
@ -62,8 +85,8 @@ public sealed class CultMeshClientTests
Connectors = new[] { connector }
});
using var firstLease = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
using var secondLease = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
using var firstLease = await mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "surface:pilot");
using var secondLease = await mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "surface:pilot");
var first = firstLease.Handle;
var second = secondLease.Handle;
@ -82,7 +105,7 @@ public sealed class CultMeshClientTests
Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
Connectors = new[] { connector }
});
using var lease = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
using var lease = await mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "surface:pilot");
var document = lease.Handle;
connector.Clients[0].Fail(new InvalidOperationException("path lost"));
@ -105,7 +128,7 @@ public sealed class CultMeshClientTests
Connectors = new[] { connector }
});
var opening = mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
var opening = mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "surface:pilot");
await WaitUntilAsync(() => connector.Clients.Count == 1 && connector.Clients[0].SubscribeCount == 1);
connector.Clients[0].Fail(new InvalidOperationException("path lost before initial snapshot"));
@ -129,7 +152,7 @@ public sealed class CultMeshClientTests
SubscriptionResponseTimeout = TimeSpan.FromMilliseconds(20)
});
using var lease = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
using var lease = await mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "surface:pilot");
var document = lease.Handle;
(await document.LatestAsync()).Text.Should().Be("pilot surface 1");
@ -148,7 +171,7 @@ public sealed class CultMeshClientTests
Connectors = new[] { connector }
});
using var lease = await mesh.LeaseCollectionAsync<ClientTestDocument>("aetheria");
using var lease = await mesh.LeaseCollectionAsync<ClientTestDocument>(Target);
var collection = lease.Handle;
(await collection.LatestAsync()).Should().ContainSingle()
@ -167,7 +190,7 @@ public sealed class CultMeshClientTests
Connectors = new[] { connector }
});
var opening = mesh.LeaseCollectionAsync<ClientTestDocument>("aetheria");
var opening = mesh.LeaseCollectionAsync<ClientTestDocument>(Target);
await WaitUntilAsync(() => connector.Clients.Count == 1 && connector.Clients[0].SubscribeCount == 1);
connector.Clients[0].Fail(new InvalidOperationException("path lost before initial collection snapshot"));
@ -191,8 +214,8 @@ public sealed class CultMeshClientTests
Connectors = new[] { connector }
});
var first = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
var second = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
var first = await mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "surface:pilot");
var second = await mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "surface:pilot");
mesh.ActiveDocumentResourceCount.Should().Be(1);
first.Dispose();
@ -200,7 +223,7 @@ public sealed class CultMeshClientTests
second.Dispose();
mesh.ActiveDocumentResourceCount.Should().Be(0);
using var replacement = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
using var replacement = await mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "surface:pilot");
replacement.Handle.Should().NotBeSameAs(first.Handle);
connector.SubscribeCount.Should().Be(2);
}
@ -220,7 +243,7 @@ public sealed class CultMeshClientTests
for (var index = 0; index < leaseCount; index++)
{
var recordKey = "surface:churn:" + index;
using var lease = await mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", recordKey);
using var lease = await mesh.LeaseDocumentAsync<ClientTestDocument>(Target, recordKey);
(await lease.Handle.LatestAsync()).Id.Should().Be(recordKey);
}
@ -244,7 +267,7 @@ public sealed class CultMeshClientTests
var request = new ClientTestDocument { Id = "command:1", Text = "move" };
await mesh.SubmitDocumentAsync(
"aetheria",
Target,
request.Id,
request,
sourceRuntimeId: "pilot-one",
@ -269,7 +292,7 @@ public sealed class CultMeshClientTests
});
var result = await mesh.InvokeAsync<ClientOperationRequest, ClientOperationReceipt>(
"aetheria",
Target,
"tests.counter",
"tests.counter.increment",
"tests.counter.increment_request.v1",
@ -301,7 +324,7 @@ public sealed class CultMeshClientTests
});
Func<Task> invoke = async () => await mesh.InvokeAsync<ClientOperationRequest, ClientOperationReceipt>(
"aetheria",
Target,
"tests.counter",
"tests.counter.increment",
"wrong.request.v1",
@ -328,7 +351,7 @@ public sealed class CultMeshClientTests
Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
Connectors = new[] { connector }
});
var opening = mesh.LeaseDocumentAsync<ClientTestDocument>("aetheria", "surface:pilot");
var opening = mesh.LeaseDocumentAsync<ClientTestDocument>(Target, "surface:pilot");
await WaitUntilAsync(() => connector.Clients.Count == 1 && connector.Clients[0].SubscribeCount == 1);
mesh.Dispose();
@ -355,7 +378,8 @@ public sealed class CultMeshClientTests
"Aetheria",
CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v0", "rules"),
new[] { "rudp://aetheria:3081" }).ToMessage()
new[] { "rudp://aetheria:3081" },
new[] { "aetheria-daemon" }).ToMessage()
}
};
foreach (var handler in _handlers.ToArray()) handler(response);

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

@ -52,7 +52,7 @@ public sealed class CultMeshContentSessionTests
var provider = new CultMeshSessionContentProvider(
"aetheria.daemon",
sessions,
CultMeshEndpointId.Parse("aetheria"));
new CultMeshSessionTarget("aetheria", "aetheria.daemon"));
using var transferState = new CultCache(CultMesh.CreateCultCacheDocumentRegistry(
typeof(CultMeshContentTransferStateDocument)));
var transfer = new CultMeshContentTransferService(
@ -114,7 +114,7 @@ public sealed class CultMeshContentSessionTests
var provider = new CultMeshSessionContentProvider(
"aetheria.daemon",
sessions,
CultMeshEndpointId.Parse("aetheria"));
new CultMeshSessionTarget("aetheria", "aetheria.daemon"));
using var transferState = new CultCache(CultMesh.CreateCultCacheDocumentRegistry(
typeof(CultMeshContentTransferStateDocument)));
var transfer = new CultMeshContentTransferService(
@ -196,7 +196,8 @@ public sealed class CultMeshContentSessionTests
"Aetheria",
CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v1", "rules"),
new[] { _endpoint }),
new[] { _endpoint },
new[] { "aetheria.daemon" }),
SourceId,
now,
now.AddMinutes(1),

22
tests/GameCult.Mesh.Tests/CultMeshDiscoveryServiceTests.cs
View file

@ -263,6 +263,25 @@ public sealed class CultMeshDiscoveryServiceTests
source.LookupCount.Should().Be(2);
}
[Test]
public async Task Resolve_RejectsVerseRouteThatDoesNotAdvertiseRequestedProvider()
{
var clock = new ManualClock(new DateTimeOffset(2026, 8, 18, 0, 0, 0, TimeSpan.Zero));
var source = new DelegateSource("odin", _ => Task.FromResult<IReadOnlyList<CultMeshDiscoveryObservation>>(
new[] { Observation("aetheria", "odin", clock, TimeSpan.FromMinutes(1), CultMeshDiscoveryTrust.Signed) }));
using var service = new CultMeshDiscoveryService(
new[] { source },
new CultMeshDiscoveryServiceOptions { Clock = clock });
var state = await service.ResolveAsync(new CultMeshDiscoveryQuery(
"forged-provider",
new[] { "aetheria" },
"forged-provider"));
state.Freshness.Should().Be(CultMeshDiscoveryFreshness.Unavailable);
state.Candidates.Should().BeEmpty();
}
private static CultMeshDiscoveryObservation Observation(
string verseId,
string sourceId,
@ -275,7 +294,8 @@ public sealed class CultMeshDiscoveryServiceTests
"Aetheria",
CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v0", "rules"),
new[] { endpoint }),
new[] { endpoint },
new[] { "odin:aetheria", "odin:games", "aetheria" }),
sourceId,
clock.UtcNow,
clock.UtcNow + ttl,

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

@ -13,6 +13,8 @@ namespace GameCult.Mesh.Tests;
public sealed class CultMeshSessionManagerTests
{
private static CultMeshSessionTarget Target => new("aetheria", "aetheria-daemon");
[Test]
public async Task Connect_ResolvesOnlyTheRequestedVerseIdentity()
{
@ -28,7 +30,7 @@ public sealed class CultMeshSessionManagerTests
});
using var manager = new CultMeshSessionManager(discovery, new[] { connector }, new CultMeshSessionManagerOptions { Clock = clock });
await manager.ConnectAsync(CultMeshEndpointId.Parse("aetheria"), CultMeshProtocols.Documents);
await manager.ConnectAsync(Target, CultMeshProtocols.Documents);
connected.Should().NotBeNull();
connected!.Endpoint.Should().Be("rudp://aetheria:3076");
@ -42,14 +44,14 @@ public sealed class CultMeshSessionManagerTests
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var connector = new FakeConnector(async (_, _) => { await release.Task; return new FakeSchemaClient(); });
using var manager = new CultMeshSessionManager(discovery, new[] { connector }, new CultMeshSessionManagerOptions { Clock = clock });
var endpoint = CultMeshEndpointId.Parse("aetheria");
var target = Target;
var first = manager.ConnectAsync(endpoint, CultMeshProtocols.Documents);
var second = manager.ConnectAsync(endpoint, CultMeshProtocols.Documents);
var first = manager.ConnectAsync(target, CultMeshProtocols.Documents);
var second = manager.ConnectAsync(target, CultMeshProtocols.Documents);
await WaitUntilAsync(() => connector.ConnectCount == 1);
release.TrySetResult();
var sessions = await Task.WhenAll(first, second);
var third = await manager.ConnectAsync(endpoint, CultMeshProtocols.Documents);
var third = await manager.ConnectAsync(target, CultMeshProtocols.Documents);
sessions[0].Should().BeSameAs(sessions[1]).And.BeSameAs(third);
connector.ConnectCount.Should().Be(1);
@ -70,7 +72,7 @@ public sealed class CultMeshSessionManagerTests
});
using var manager = new CultMeshSessionManager(discovery, new[] { connector }, new CultMeshSessionManagerOptions { Clock = clock });
var session = await manager.ConnectAsync(CultMeshEndpointId.Parse("aetheria"), CultMeshProtocols.Documents);
var session = await manager.ConnectAsync(Target, CultMeshProtocols.Documents);
session.State.Path!.Endpoint.Should().Be("rudp://fast:3076");
slowRelease.TrySetResult();
await WaitUntilAsync(() => slowClient.DisposeCount == 1);
@ -91,10 +93,10 @@ public sealed class CultMeshSessionManagerTests
});
using var manager = new CultMeshSessionManager(discovery, new[] { connector }, new CultMeshSessionManagerOptions { Clock = clock });
using var cancellation = new CancellationTokenSource();
var endpoint = CultMeshEndpointId.Parse("aetheria");
var target = Target;
var cancelled = manager.ConnectAsync(endpoint, CultMeshProtocols.Documents, cancellation.Token);
var survivor = manager.ConnectAsync(endpoint, CultMeshProtocols.Documents);
var cancelled = manager.ConnectAsync(target, CultMeshProtocols.Documents, cancellation.Token);
var survivor = manager.ConnectAsync(target, CultMeshProtocols.Documents);
await WaitUntilAsync(() => connector.ConnectCount == 1);
cancellation.Cancel();
Assert.ThrowsAsync<OperationCanceledException>(async () => await cancelled);
@ -118,8 +120,8 @@ public sealed class CultMeshSessionManagerTests
return Task.FromResult<ICultNetSchemaClient>(client);
});
using var manager = new CultMeshSessionManager(discovery, new[] { connector }, new CultMeshSessionManagerOptions { Clock = clock });
var endpoint = CultMeshEndpointId.Parse("aetheria");
var first = await manager.ConnectAsync(endpoint, CultMeshProtocols.Documents);
var target = Target;
var first = await manager.ConnectAsync(target, CultMeshProtocols.Documents);
var states = new List<CultMeshSessionStatus>();
using var stateWatch = first.WatchState().Subscribe(state => states.Add(state.Status));
using var schemaLease = first.OpenSchemaClient();
@ -130,7 +132,7 @@ public sealed class CultMeshSessionManagerTests
route = "rudp://second:3076";
clients[0].Fail(new IOException("partition"));
await WaitUntilAsync(() => first.State.Status == CultMeshSessionStatus.Online && connector.ConnectCount == 2);
var second = await manager.ConnectAsync(endpoint, CultMeshProtocols.Documents);
var second = await manager.ConnectAsync(target, CultMeshProtocols.Documents);
clients[1].Emit(new CultNetErrorMessage { Error = "after rotation" });
second.Should().BeSameAs(first);
@ -165,7 +167,7 @@ public sealed class CultMeshSessionManagerTests
using var manager = new CultMeshSessionManager(discovery, Array.Empty<ICultMeshTransportConnector>(), new CultMeshSessionManagerOptions { Clock = clock });
var error = Assert.ThrowsAsync<CultMeshSessionException>(() =>
manager.ConnectAsync(CultMeshEndpointId.Parse("aetheria"), CultMeshProtocols.Documents));
manager.ConnectAsync(Target, CultMeshProtocols.Documents));
error!.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Transport);
error.InnerException.Should().BeOfType<CultMeshSessionException>()
@ -180,9 +182,9 @@ public sealed class CultMeshSessionManagerTests
var client = new RespondingSchemaClient();
var connector = new FakeConnector((_, _) => Task.FromResult<ICultNetSchemaClient>(client));
using var manager = new CultMeshSessionManager(discovery, new[] { connector }, new CultMeshSessionManagerOptions { Clock = clock });
var endpoint = CultMeshEndpointId.Parse("aetheria");
using var first = await CultMeshSnapshotSession.ConnectAsync(manager, endpoint, new CultMeshSnapshotRequestOptions());
using var second = await CultMeshSnapshotSession.ConnectAsync(manager, endpoint, new CultMeshSnapshotRequestOptions());
var target = Target;
using var first = await CultMeshSnapshotSession.ConnectAsync(manager, target, new CultMeshSnapshotRequestOptions());
using var second = await CultMeshSnapshotSession.ConnectAsync(manager, target, new CultMeshSnapshotRequestOptions());
(await first.FetchSnapshotAsync()).MessageId.Should().NotBeEmpty();
first.Dispose();
@ -202,10 +204,10 @@ public sealed class CultMeshSessionManagerTests
var connector = new FakeConnector((_, _) => Task.FromResult<ICultNetSchemaClient>(client));
using var manager = new CultMeshSessionManager(discovery, new[] { connector }, new CultMeshSessionManagerOptions { Clock = clock });
var exchange = new CultMeshPeerExchangeClient(new CultMeshPeerExchangeClientOptions { Sessions = manager, Clock = clock });
var endpoint = CultMeshEndpointId.Parse("aetheria");
var target = Target;
var first = await exchange.FetchAsync(endpoint, new CultMeshPeerExchangeRequestMessage { VerseId = "aetheria" });
var second = await exchange.FetchAsync(endpoint, new CultMeshPeerExchangeRequestMessage { VerseId = "aetheria" });
var first = await exchange.FetchAsync(target, new CultMeshPeerExchangeRequestMessage { VerseId = "aetheria" });
var second = await exchange.FetchAsync(target, new CultMeshPeerExchangeRequestMessage { VerseId = "aetheria" });
first.Peers.Single().PeerId.Should().Be("peer-a");
second.Peers.Single().PeerId.Should().Be("peer-a");
@ -221,7 +223,7 @@ public sealed class CultMeshSessionManagerTests
var client = new LeaseSchemaClient();
var connector = new FakeConnector((_, _) => Task.FromResult<ICultNetSchemaClient>(client));
using var manager = new CultMeshSessionManager(discovery, new[] { connector }, new CultMeshSessionManagerOptions { Clock = clock });
var session = await manager.ConnectAsync(CultMeshEndpointId.Parse("aetheria"), CultMeshProtocols.Subscriptions);
var session = await manager.ConnectAsync(Target, CultMeshProtocols.Subscriptions);
using var first = session.OpenSchemaClient();
using var second = session.OpenSchemaClient();
var firstCount = 0;
@ -268,7 +270,7 @@ public sealed class CultMeshSessionManagerTests
{
new CultMeshDiscoveryObservation(
new CultMeshVerseDescriptor("aetheria", "Aetheria", CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v0", "rules"), _endpoints()),
new CultMeshVerseCompatibility("cultmesh.v0", "rules"), _endpoints(), new[] { "aetheria-daemon" }),
SourceId, _clock.UtcNow, _clock.UtcNow + _ttl, CultMeshDiscoveryTrust.Signed)
});
}
@ -298,7 +300,8 @@ public sealed class CultMeshSessionManagerTests
verseId,
CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v0", "rules"),
new[] { endpoint }),
new[] { endpoint },
new[] { "aetheria-daemon" }),
SourceId,
_clock.UtcNow,
expires,

13
tests/GameCult.Mesh.Tests/CultMeshTransportModularityTests.cs
View file

@ -39,7 +39,7 @@ public sealed class CultMeshTransportModularityTests
});
var session = await sessions.ConnectAsync(
CultMeshEndpointId.Parse("aetheria"),
new CultMeshSessionTarget("aetheria", "aetheria.daemon"),
CultMeshProtocols.Documents);
var response = new TaskCompletionSource<CultNetErrorMessage>(TaskCreationOptions.RunContinuationsAsynchronously);
using var subscription = session.OnCultNet<CultNetErrorMessage>(message => response.TrySetResult(message));
@ -77,7 +77,7 @@ public sealed class CultMeshTransportModularityTests
new CultMeshTcpContentTransportConnector()
});
var session = await sessions.ConnectContentAsync(CultMeshEndpointId.Parse("aetheria"));
var session = await sessions.ConnectContentAsync(new CultMeshSessionTarget("aetheria", "aetheria.daemon"));
using var destination = new MemoryStream();
await session.CopyChunkToAsync(artifact.Manifest.Chunks[0], destination);
@ -98,7 +98,7 @@ public sealed class CultMeshTransportModularityTests
Array.Empty<ICultMeshTransportConnector>());
Func<Task> connect = async () => await sessions.ConnectContentAsync(
CultMeshEndpointId.Parse("aetheria"));
new CultMeshSessionTarget("aetheria", "aetheria.daemon"));
await connect.Should().ThrowAsync<CultMeshSessionException>()
.WithMessage("*No streaming content connectors are configured*");
@ -118,7 +118,7 @@ public sealed class CultMeshTransportModularityTests
Array.Empty<ICultMeshContentTransportConnector>(),
new ICultMeshRealtimeTransportConnector[] { connector });
var session = await sessions.ConnectRealtimeAsync(
CultMeshEndpointId.Parse("aetheria"));
new CultMeshSessionTarget("aetheria", "aetheria.daemon"));
var sent = new CultMeshRealtimeFrame
{
ChannelId = "aetheria.entities",
@ -150,7 +150,7 @@ public sealed class CultMeshTransportModularityTests
Array.Empty<ICultMeshTransportConnector>());
Func<Task> connect = async () => await sessions.ConnectRealtimeAsync(
CultMeshEndpointId.Parse("aetheria"));
new CultMeshSessionTarget("aetheria", "aetheria.daemon"));
await connect.Should().ThrowAsync<CultMeshSessionException>()
.WithMessage("*No realtime state connectors are configured*");
@ -176,7 +176,8 @@ public sealed class CultMeshTransportModularityTests
"Aetheria",
CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v1", "rules"),
_endpoints),
_endpoints,
new[] { "aetheria.daemon" }),
SourceId,
now,
now.AddMinutes(1),

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

@ -202,7 +202,8 @@ static async Task<StateResult> MeasureCultMeshQuicStateAsync(int payloadBytes, i
{
ValidateProviderCertificate = (_, _, _, _) => true
})]);
var session = await sessions.ConnectRealtimeAsync(CultMeshEndpointId.Parse("service:parity.provider"));
var session = await sessions.ConnectRealtimeAsync(
new CultMeshSessionTarget("parity", "service:parity.provider"));
var deadline = DateTime.UtcNow.AddSeconds(5);
while (server.ConnectionCount == 0)
{
@ -478,7 +479,8 @@ internal sealed class RouteSource(string endpoint) : ICultMeshLookupSource
new(
new CultMeshVerseDescriptor(
"parity", "Transport parity", CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v1", "parity"), [endpoint]),
new CultMeshVerseCompatibility("cultmesh.v1", "parity"), [endpoint],
["service:parity.provider"]),
SourceId, now, now.AddMinutes(1), CultMeshDiscoveryTrust.Signed)
];
return Task.FromResult(observations);