From c75051017dd6e9123be8d440d24109b672c7e581 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 18 Aug 2026 07:26:55 +0200 Subject: [PATCH] Make CultMesh session identity explicit --- samples/eve-browser-network/Program.cs | 24 ++- scripts/verify-eve-getting-started.ps1 | 12 ++ src/GameCult.Mesh/CultMeshBodySessions.cs | 8 +- src/GameCult.Mesh/CultMeshClient.cs | 117 ++++++-------- .../CultMeshContentTransports.cs | 14 +- src/GameCult.Mesh/CultMeshDiscoveryService.cs | 19 ++- src/GameCult.Mesh/CultMeshPeerExchange.cs | 14 +- .../CultMeshRealtimeTransports.cs | 6 +- src/GameCult.Mesh/CultMeshSessions.cs | 152 +++++++++++------- src/GameCult.Mesh/CultMeshSnapshots.cs | 10 +- src/GameCult.Mesh/docs/content-sessions.md | 5 +- .../getting-started/02-connect-by-identity.md | 23 ++- .../03-publish-an-eve-surface.md | 4 +- .../04-run-the-multiruntime-game.md | 4 +- .../docs/reliability-migration-plan.md | 7 +- src/GameCult.Mesh/docs/transport-planes.md | 6 +- .../CultMeshQuicRealtimeTransportTests.cs | 12 +- .../CultMeshBodySessionTests.cs | 6 +- .../CultMeshClientTests.cs | 62 ++++--- .../CultMeshContentSessionTests.cs | 7 +- .../CultMeshDiscoveryServiceTests.cs | 22 ++- .../CultMeshSessionManagerTests.cs | 47 +++--- .../CultMeshTransportModularityTests.cs | 13 +- tools/GameCult.TransportParity/Program.cs | 6 +- 24 files changed, 364 insertions(+), 236 deletions(-) diff --git a/samples/eve-browser-network/Program.cs b/samples/eve-browser-network/Program.cs index fe6a74b9..e06c235a 100644 --- a/samples/eve-browser-network/Program.cs +++ b/samples/eve-browser-network/Program.cs @@ -183,7 +183,8 @@ static async Task RunHeadlessAsync(Args arguments) _ => CreateClient()) ] }); - using var lease = await mesh.LeaseDocumentAsync(arguments.VerseId, counterKey); + var target = new CultMeshSessionTarget(arguments.VerseId, arguments.AuthorityRuntimeId); + using var lease = await mesh.LeaseDocumentAsync(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 ' on stdin."); var commandId = line["INVOKE ".Length..].Trim(); var result = await mesh.InvokeAsync( - 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 MeasureOperationSessionAsync(CultMeshClient mesh, Args arguments) +static async Task 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 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 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( - arguments.VerseId, + target, "sample.counter", "sample.counter.ping", "sample.ping_request.v1", diff --git a/scripts/verify-eve-getting-started.ps1 b/scripts/verify-eve-getting-started.ps1 index 1805b070..0a690712 100644 --- a/scripts/verify-eve-getting-started.ps1 +++ b/scripts/verify-eve-getting-started.ps1 @@ -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")) diff --git a/src/GameCult.Mesh/CultMeshBodySessions.cs b/src/GameCult.Mesh/CultMeshBodySessions.cs index cc45277e..ee946201 100644 --- a/src/GameCult.Mesh/CultMeshBodySessions.cs +++ b/src/GameCult.Mesh/CultMeshBodySessions.cs @@ -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(TaskCreationOptions.RunContinuationsAsynchronously); using var responseSubscription = session.OnCultNet(response => diff --git a/src/GameCult.Mesh/CultMeshClient.cs b/src/GameCult.Mesh/CultMeshClient.cs index 1d33c3ec..99746a5c 100644 --- a/src/GameCult.Mesh/CultMeshClient.cs +++ b/src/GameCult.Mesh/CultMeshClient.cs @@ -192,76 +192,66 @@ namespace GameCult.Mesh _operationResponseTimeout = options.OperationResponseTimeout; } - /// Connects to a stable endpoint identity using one application protocol. + /// Connects to one stable Verse/provider target using one application protocol. public Task 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); } - /// Connects the reusable realtime state plane by stable endpoint identity. + /// Connects the reusable realtime state plane for one stable Verse/provider target. public Task ConnectRealtimeAsync( - string endpointId, + CultMeshSessionTarget target, CancellationToken cancellationToken = default) { ThrowIfDisposed(); - return _sessions.ConnectRealtimeAsync(CultMeshEndpointId.Parse(endpointId), cancellationToken); - } - - /// Connects to a stable endpoint identity using one application protocol. - public Task ConnectAsync( - CultMeshEndpointId endpointId, - CultMeshProtocolId protocol, - CancellationToken cancellationToken = default) - { - ThrowIfDisposed(); - return _sessions.ConnectAsync(endpointId, protocol, cancellationToken); + return _sessions.ConnectRealtimeAsync(target, cancellationToken); } /// Creates a verified-transfer provider over this client's reusable content session owner. public CultMeshSessionContentProvider ContentProvider( string providerId, - string endpointId) + CultMeshSessionTarget target) { ThrowIfDisposed(); return new CultMeshSessionContentProvider( providerId, _sessions, - CultMeshEndpointId.Parse(endpointId)); + target); } /// Creates a direct body provider over this client's reusable negotiated session owner. public CultMeshSessionBodyProvider BodyProvider( string providerId, - string endpointId, + CultMeshSessionTarget target, CultMeshSessionBodyProviderOptions? options = null) { ThrowIfDisposed(); return new CultMeshSessionBodyProvider( providerId, _sessions, - CultMeshEndpointId.Parse(endpointId), + target, options); } - /// Leases one shared live typed document by stable provider identity and record key. + /// Leases one shared live typed document by stable Verse/provider identity and record key. public async Task> LeaseDocumentAsync( - 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(endpointId.Trim(), recordKey.Trim(), owner).ConfigureAwait(false))); + async owner => await OpenDocumentAsync(target, recordKey.Trim(), owner).ConfigureAwait(false))); if (!resource.TryAcquire()) { RemoveExact(_documents, key, resource); @@ -283,28 +273,28 @@ namespace GameCult.Mesh } } - /// Leases a shared live collection of one typed schema by stable provider identity. + /// Leases a shared live collection by stable Verse/provider identity. public Task> LeaseCollectionAsync( - string endpointId, + CultMeshSessionTarget target, CancellationToken cancellationToken) where TDocument : class => - LeaseCollectionAsync(endpointId, includeInitialSnapshot: true, cancellationToken); + LeaseCollectionAsync(target, includeInitialSnapshot: true, cancellationToken); /// Leases a live collection and optionally skips replaying its initial snapshot. public async Task> LeaseCollectionAsync( - 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(identity, includeInitialSnapshot, owner).ConfigureAwait(false))); + async owner => await OpenCollectionAsync(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. /// public async Task SubmitDocumentAsync( - 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. /// public async Task> InvokeAsync( - 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(TaskCreationOptions.RunContinuationsAsynchronously); using var subscription = session.OnCultNet(response => { @@ -481,69 +470,69 @@ namespace GameCult.Mesh /// Reads one typed record once through the reusable document session. public async Task ReadAsync( - 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(endpointId, new[] { recordKey }, cancellationToken) + var documents = await ReadManyAsync(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}'."); } /// Reads one typed record once with an explicit response deadline. public async Task ReadAsync( - string endpointId, + CultMeshSessionTarget target, string recordKey, TimeSpan responseTimeout, CancellationToken cancellationToken = default) where TDocument : class { var documents = await ReadManyCoreAsync( - 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}'."); } /// Reads typed records once in one request through the reusable document session. public async Task> ReadManyAsync( - string endpointId, + CultMeshSessionTarget target, IReadOnlyList recordKeys, CancellationToken cancellationToken = default) where TDocument : class => await ReadManyCoreAsync( - endpointId, + target, recordKeys, TimeSpan.FromSeconds(10), cancellationToken) .ConfigureAwait(false); private async Task> ReadManyCoreAsync( - string endpointId, + CultMeshSessionTarget target, IReadOnlyList 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 OpenDocumentAsync( - string endpointId, + CultMeshSessionTarget target, string recordKey, SharedResource owner) where TDocument : class { - var session = await ConnectAsync(endpointId, CultMeshProtocols.Documents).ConfigureAwait(false); - var binding = new RemoteDocumentBinding(session, endpointId, recordKey, _subscriptionResponseTimeout); + var session = await ConnectAsync(target, CultMeshProtocols.Documents).ConfigureAwait(false); + var binding = new RemoteDocumentBinding(session, target, recordKey, _subscriptionResponseTimeout); owner.Attach(binding); try { @@ -599,13 +588,13 @@ namespace GameCult.Mesh } private async Task OpenCollectionAsync( - string endpointId, + CultMeshSessionTarget target, bool includeInitialSnapshot, SharedResource owner) where TDocument : class { - var session = await ConnectAsync(endpointId, CultMeshProtocols.Documents).ConfigureAwait(false); - var binding = new RemoteCollectionBinding(session, endpointId, includeInitialSnapshot, _subscriptionResponseTimeout); + var session = await ConnectAsync(target, CultMeshProtocols.Documents).ConfigureAwait(false); + var binding = new RemoteCollectionBinding(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( _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 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( _cache, - routeHint: new CultMeshRouteHint(CultMeshLocalityKind.Network, endpointId)); + routeHint: new CultMeshRouteHint(CultMeshLocalityKind.Network, target.ProviderRuntimeId)); } public CultMeshCollectionHandle Handle { get; } diff --git a/src/GameCult.Mesh/CultMeshContentTransports.cs b/src/GameCult.Mesh/CultMeshContentTransports.cs index f1814b2c..3e2c54b1 100644 --- a/src/GameCult.Mesh/CultMeshContentTransports.cs +++ b/src/GameCult.Mesh/CultMeshContentTransports.cs @@ -52,18 +52,18 @@ namespace GameCult.Mesh private bool _disposed; internal CultMeshContentSession( - CultMeshEndpointId endpointId, + CultMeshSessionTarget target, ICultMeshContentTransport transport, CultMeshSessionState state, Action 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); } } diff --git a/src/GameCult.Mesh/CultMeshDiscoveryService.cs b/src/GameCult.Mesh/CultMeshDiscoveryService.cs index 5e8c43b8..93ea6a84 100644 --- a/src/GameCult.Mesh/CultMeshDiscoveryService.cs +++ b/src/GameCult.Mesh/CultMeshDiscoveryService.cs @@ -33,16 +33,25 @@ namespace GameCult.Mesh public sealed class CultMeshDiscoveryQuery { - public CultMeshDiscoveryQuery(string endpointId, IEnumerable? verseIds = null) + /// Stable lookup identity used for caching and diagnostics, never a physical route. + /// Optional Verse filter. + /// Optional provider runtime that each accepted Verse must advertise. + public CultMeshDiscoveryQuery( + string endpointId, + IEnumerable? 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(); + AuthorityRuntimeId = string.IsNullOrWhiteSpace(authorityRuntimeId) ? null : authorityRuntimeId.Trim(); } public string EndpointId { get; } public IReadOnlyList VerseIds { get; } + /// Gets the optional provider runtime constraint for authority-sensitive resolution. + 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 AwaitForCallerAsync(Task 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)) diff --git a/src/GameCult.Mesh/CultMeshPeerExchange.cs b/src/GameCult.Mesh/CultMeshPeerExchange.cs index 2489ebac..c3045c0d 100644 --- a/src/GameCult.Mesh/CultMeshPeerExchange.cs +++ b/src/GameCult.Mesh/CultMeshPeerExchange.cs @@ -446,15 +446,17 @@ namespace GameCult.Mesh /// Fetches peer cards through the shared session owner. public async Task 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(TaskCreationOptions.RunContinuationsAsynchronously); using var responseSubscription = session.OnCultNet(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); } /// @@ -541,7 +543,7 @@ namespace GameCult.Mesh private async Task WaitForManagedResponseAsync( Task 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); } diff --git a/src/GameCult.Mesh/CultMeshRealtimeTransports.cs b/src/GameCult.Mesh/CultMeshRealtimeTransports.cs index d087b304..7b32ad6a 100644 --- a/src/GameCult.Mesh/CultMeshRealtimeTransports.cs +++ b/src/GameCult.Mesh/CultMeshRealtimeTransports.cs @@ -72,18 +72,18 @@ namespace GameCult.Mesh private bool _disposed; internal CultMeshRealtimeSession( - CultMeshEndpointId endpointId, + CultMeshSessionTarget target, ICultMeshRealtimeTransport transport, CultMeshSessionState state, Action 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; } } diff --git a/src/GameCult.Mesh/CultMeshSessions.cs b/src/GameCult.Mesh/CultMeshSessions.cs index 54a19fe8..c562de83 100644 --- a/src/GameCult.Mesh/CultMeshSessions.cs +++ b/src/GameCult.Mesh/CultMeshSessions.cs @@ -21,6 +21,40 @@ namespace GameCult.Mesh public override string ToString() => Value; } + /// + /// Identifies one provider inside one Verse. Odin resolves this stable pair to + /// disposable physical routes; neither value is a socket address. + /// + public sealed class CultMeshSessionTarget : IEquatable + { + /// Creates one stable provider target inside one Verse. + 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; + } + + /// Gets the stable Verse identity. + public string VerseId { get; } + /// Gets the stable provider runtime identity advertised by that Verse. + 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 { 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 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>( - () => 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 } /// - /// 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. /// public async Task 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>( - () => 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 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 ResolveContentPathAsync(CultMeshEndpointId endpointId) + private async Task 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(); 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); } /// @@ -709,16 +744,16 @@ namespace GameCult.Mesh /// register QUIC explicitly, while RUDP compatibility remains opt-in. /// public async Task 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>( - () => 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 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 ResolveRealtimePathAsync(CultMeshEndpointId endpointId) + private async Task 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(); 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 ConnectOwnedAsync(string key, CultMeshEndpointId endpointId, CultMeshProtocolId protocol) + private async Task 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 ResolveConnectedPathAsync(CultMeshEndpointId endpointId, CultMeshProtocolId protocol) + private async Task 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(); 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)); } diff --git a/src/GameCult.Mesh/CultMeshSnapshots.cs b/src/GameCult.Mesh/CultMeshSnapshots.cs index 03fc4aad..cff94fa4 100644 --- a/src/GameCult.Mesh/CultMeshSnapshots.cs +++ b/src/GameCult.Mesh/CultMeshSnapshots.cs @@ -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 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 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); } diff --git a/src/GameCult.Mesh/docs/content-sessions.md b/src/GameCult.Mesh/docs/content-sessions.md index 1eae6b6c..313a21cf 100644 --- a/src/GameCult.Mesh/docs/content-sessions.md +++ b/src/GameCult.Mesh/docs/content-sessions.md @@ -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, diff --git a/src/GameCult.Mesh/docs/getting-started/02-connect-by-identity.md b/src/GameCult.Mesh/docs/getting-started/02-connect-by-identity.md index 8dffffaa..5cf90a0e 100644 --- a/src/GameCult.Mesh/docs/getting-started/02-connect-by-identity.md +++ b/src/GameCult.Mesh/docs/getting-started/02-connect-by-identity.md @@ -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( - "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); diff --git a/src/GameCult.Mesh/docs/getting-started/03-publish-an-eve-surface.md b/src/GameCult.Mesh/docs/getting-started/03-publish-an-eve-surface.md index 4d5af0dc..265247b2 100644 --- a/src/GameCult.Mesh/docs/getting-started/03-publish-an-eve-surface.md +++ b/src/GameCult.Mesh/docs/getting-started/03-publish-an-eve-surface.md @@ -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 diff --git a/src/GameCult.Mesh/docs/getting-started/04-run-the-multiruntime-game.md b/src/GameCult.Mesh/docs/getting-started/04-run-the-multiruntime-game.md index a748368b..ba9888c3 100644 --- a/src/GameCult.Mesh/docs/getting-started/04-run-the-multiruntime-game.md +++ b/src/GameCult.Mesh/docs/getting-started/04-run-the-multiruntime-game.md @@ -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: diff --git a/src/GameCult.Mesh/docs/reliability-migration-plan.md b/src/GameCult.Mesh/docs/reliability-migration-plan.md index 658d5e0a..47e1b190 100644 --- a/src/GameCult.Mesh/docs/reliability-migration-plan.md +++ b/src/GameCult.Mesh/docs/reliability-migration-plan.md @@ -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); diff --git a/src/GameCult.Mesh/docs/transport-planes.md b/src/GameCult.Mesh/docs/transport-planes.md index 314c55b4..39be9c59 100644 --- a/src/GameCult.Mesh/docs/transport-planes.md +++ b/src/GameCult.Mesh/docs/transport-planes.md @@ -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:** diff --git a/tests/GameCult.Mesh.Quic.Tests/CultMeshQuicRealtimeTransportTests.cs b/tests/GameCult.Mesh.Quic.Tests/CultMeshQuicRealtimeTransportTests.cs index 1b481d28..8d90f147 100644 --- a/tests/GameCult.Mesh.Quic.Tests/CultMeshQuicRealtimeTransportTests.cs +++ b/tests/GameCult.Mesh.Quic.Tests/CultMeshQuicRealtimeTransportTests.cs @@ -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 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), diff --git a/tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs b/tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs index c105ae06..bdf9928f 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs @@ -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) }); } } diff --git a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs index 82934a54..94194abd 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs @@ -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 connect = async () => await mesh.ConnectAsync( + new CultMeshSessionTarget("aetheria", "forged-provider"), + CultMeshProtocols.Documents); + + var failure = await connect.Should().ThrowAsync(); + 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("aetheria", "surface:pilot"); - using var secondLease = await mesh.LeaseDocumentAsync("aetheria", "surface:pilot"); + using var firstLease = await mesh.LeaseDocumentAsync(Target, "surface:pilot"); + using var secondLease = await mesh.LeaseDocumentAsync(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("aetheria", "surface:pilot"); + using var lease = await mesh.LeaseDocumentAsync(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("aetheria", "surface:pilot"); + var opening = mesh.LeaseDocumentAsync(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("aetheria", "surface:pilot"); + using var lease = await mesh.LeaseDocumentAsync(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("aetheria"); + using var lease = await mesh.LeaseCollectionAsync(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("aetheria"); + var opening = mesh.LeaseCollectionAsync(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("aetheria", "surface:pilot"); - var second = await mesh.LeaseDocumentAsync("aetheria", "surface:pilot"); + var first = await mesh.LeaseDocumentAsync(Target, "surface:pilot"); + var second = await mesh.LeaseDocumentAsync(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("aetheria", "surface:pilot"); + using var replacement = await mesh.LeaseDocumentAsync(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("aetheria", recordKey); + using var lease = await mesh.LeaseDocumentAsync(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( - "aetheria", + Target, "tests.counter", "tests.counter.increment", "tests.counter.increment_request.v1", @@ -301,7 +324,7 @@ public sealed class CultMeshClientTests }); Func invoke = async () => await mesh.InvokeAsync( - "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("aetheria", "surface:pilot"); + var opening = mesh.LeaseDocumentAsync(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); diff --git a/tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs b/tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs index c1a4d04b..d63254d1 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs @@ -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), diff --git a/tests/GameCult.Mesh.Tests/CultMeshDiscoveryServiceTests.cs b/tests/GameCult.Mesh.Tests/CultMeshDiscoveryServiceTests.cs index af6334f8..7d2938e4 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshDiscoveryServiceTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshDiscoveryServiceTests.cs @@ -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>( + 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, diff --git a/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs b/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs index 4621c76b..fe9ec370 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs @@ -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(async () => await cancelled); @@ -118,8 +120,8 @@ public sealed class CultMeshSessionManagerTests return Task.FromResult(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(); 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(), new CultMeshSessionManagerOptions { Clock = clock }); var error = Assert.ThrowsAsync(() => - manager.ConnectAsync(CultMeshEndpointId.Parse("aetheria"), CultMeshProtocols.Documents)); + manager.ConnectAsync(Target, CultMeshProtocols.Documents)); error!.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Transport); error.InnerException.Should().BeOfType() @@ -180,9 +182,9 @@ public sealed class CultMeshSessionManagerTests var client = new RespondingSchemaClient(); var connector = new FakeConnector((_, _) => Task.FromResult(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(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(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, diff --git a/tests/GameCult.Mesh.Tests/CultMeshTransportModularityTests.cs b/tests/GameCult.Mesh.Tests/CultMeshTransportModularityTests.cs index 47eab18b..5fe32fed 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshTransportModularityTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshTransportModularityTests.cs @@ -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(TaskCreationOptions.RunContinuationsAsynchronously); using var subscription = session.OnCultNet(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()); Func connect = async () => await sessions.ConnectContentAsync( - CultMeshEndpointId.Parse("aetheria")); + new CultMeshSessionTarget("aetheria", "aetheria.daemon")); await connect.Should().ThrowAsync() .WithMessage("*No streaming content connectors are configured*"); @@ -118,7 +118,7 @@ public sealed class CultMeshTransportModularityTests Array.Empty(), 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()); Func connect = async () => await sessions.ConnectRealtimeAsync( - CultMeshEndpointId.Parse("aetheria")); + new CultMeshSessionTarget("aetheria", "aetheria.daemon")); await connect.Should().ThrowAsync() .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), diff --git a/tools/GameCult.TransportParity/Program.cs b/tools/GameCult.TransportParity/Program.cs index 1d724631..04d7aafc 100644 --- a/tools/GameCult.TransportParity/Program.cs +++ b/tools/GameCult.TransportParity/Program.cs @@ -202,7 +202,8 @@ static async Task 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);