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

Route CultMesh sessions by stable Verse identity

Browse source
This commit is contained in:
Meta Vens 2026-08-17 22:42:08 +02:00
commit 1fc68a4701
6 changed files with 80 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -520,7 +520,9 @@ namespace GameCult.Mesh
CultMeshSessionFailureReason.UnsupportedPath,
"No streaming content connectors are configured. Register the TCP content connector or the explicit legacy RUDP connector.");
var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery(endpointId.Value)).ConfigureAwait(false);
var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery(
endpointId.Value,
new[] { endpointId.Value })).ConfigureAwait(false);
var candidates = discovery.Candidates
.SelectMany(candidate => candidate.Descriptor.DiscoveryEndpoints)
.Distinct(StringComparer.Ordinal)
@ -646,7 +648,9 @@ namespace GameCult.Mesh
CultMeshSessionFailureReason.UnsupportedPath,
"No realtime state connectors are configured. Register a QUIC connector explicitly.");
var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery(endpointId.Value)).ConfigureAwait(false);
var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery(
endpointId.Value,
new[] { endpointId.Value })).ConfigureAwait(false);
var routes = discovery.Candidates
.SelectMany(candidate => candidate.Descriptor.DiscoveryEndpoints)
.Distinct(StringComparer.Ordinal)
@ -733,7 +737,9 @@ namespace GameCult.Mesh
private async Task<ConnectedPath> ResolveConnectedPathAsync(CultMeshEndpointId endpointId, CultMeshProtocolId protocol)
{
var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery(endpointId.Value)).ConfigureAwait(false);
var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery(
endpointId.Value,
new[] { endpointId.Value })).ConfigureAwait(false);
var candidates = discovery.Candidates.SelectMany(candidate => candidate.Descriptor.DiscoveryEndpoints)
.Distinct(StringComparer.Ordinal)
.Select(endpoint => new CultMeshTransportCandidate(endpoint))

4
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("service:aetheria.daemon"));
"aetheria.daemon", sessions, CultMeshEndpointId.Parse("aetheria"));
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("service:aetheria.daemon"),
CultMeshEndpointId.Parse("aetheria"),
new CultMeshSessionBodyProviderOptions { ResponseTimeout = TimeSpan.FromSeconds(10) });
var elapsed = Stopwatch.StartNew();

4
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("service:aetheria.daemon"));
CultMeshEndpointId.Parse("aetheria"));
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("service:aetheria.daemon"));
CultMeshEndpointId.Parse("aetheria"));
using var transferState = new CultCache(CultMesh.CreateCultCacheDocumentRegistry(
typeof(CultMeshContentTransferStateDocument)));
var transfer = new CultMeshContentTransferService(

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

@ -13,6 +13,27 @@ namespace GameCult.Mesh.Tests;
public sealed class CultMeshSessionManagerTests
{
[Test]
public async Task Connect_ResolvesOnlyTheRequestedVerseIdentity()
{
var clock = new ManualSessionClock();
using var discovery = new CultMeshDiscoveryService(
new[] { new MultiVerseRouteSource(clock) },
new CultMeshDiscoveryServiceOptions { Clock = clock });
CultMeshTransportCandidate? connected = null;
var connector = new FakeConnector((candidate, _) =>
{
connected = candidate;
return Task.FromResult<ICultNetSchemaClient>(new FakeSchemaClient());
});
using var manager = new CultMeshSessionManager(discovery, new[] { connector }, new CultMeshSessionManagerOptions { Clock = clock });
await manager.ConnectAsync(CultMeshEndpointId.Parse("aetheria"), CultMeshProtocols.Documents);
connected.Should().NotBeNull();
connected!.Endpoint.Should().Be("rudp://aetheria:3076");
}
[Test]
public async Task Connect_ReusesInflightAndEstablishedSession()
{
@ -21,7 +42,7 @@ 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("odin:aetheria");
var endpoint = CultMeshEndpointId.Parse("aetheria");
var first = manager.ConnectAsync(endpoint, CultMeshProtocols.Documents);
var second = manager.ConnectAsync(endpoint, CultMeshProtocols.Documents);
@ -49,7 +70,7 @@ public sealed class CultMeshSessionManagerTests
});
using var manager = new CultMeshSessionManager(discovery, new[] { connector }, new CultMeshSessionManagerOptions { Clock = clock });
var session = await manager.ConnectAsync(CultMeshEndpointId.Parse("odin:aetheria"), CultMeshProtocols.Documents);
var session = await manager.ConnectAsync(CultMeshEndpointId.Parse("aetheria"), CultMeshProtocols.Documents);
session.State.Path!.Endpoint.Should().Be("rudp://fast:3076");
slowRelease.TrySetResult();
await WaitUntilAsync(() => slowClient.DisposeCount == 1);
@ -70,7 +91,7 @@ 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("odin:aetheria");
var endpoint = CultMeshEndpointId.Parse("aetheria");
var cancelled = manager.ConnectAsync(endpoint, CultMeshProtocols.Documents, cancellation.Token);
var survivor = manager.ConnectAsync(endpoint, CultMeshProtocols.Documents);
@ -97,7 +118,7 @@ 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("odin:aetheria");
var endpoint = CultMeshEndpointId.Parse("aetheria");
var first = await manager.ConnectAsync(endpoint, CultMeshProtocols.Documents);
var states = new List<CultMeshSessionStatus>();
using var stateWatch = first.WatchState().Subscribe(state => states.Add(state.Status));
@ -129,7 +150,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("odin:aetheria"), CultMeshProtocols.Documents));
manager.ConnectAsync(CultMeshEndpointId.Parse("aetheria"), CultMeshProtocols.Documents));
error!.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Transport);
error.InnerException.Should().BeOfType<CultMeshSessionException>()
@ -144,7 +165,7 @@ 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("odin:aetheria");
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());
@ -166,7 +187,7 @@ 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("odin:aetheria");
var endpoint = CultMeshEndpointId.Parse("aetheria");
var first = await exchange.FetchAsync(endpoint, new CultMeshPeerExchangeRequestMessage { VerseId = "aetheria" });
var second = await exchange.FetchAsync(endpoint, new CultMeshPeerExchangeRequestMessage { VerseId = "aetheria" });
@ -185,7 +206,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("odin:aetheria"), CultMeshProtocols.Subscriptions);
var session = await manager.ConnectAsync(CultMeshEndpointId.Parse("aetheria"), CultMeshProtocols.Subscriptions);
using var first = session.OpenSchemaClient();
using var second = session.OpenSchemaClient();
var firstCount = 0;
@ -237,6 +258,38 @@ public sealed class CultMeshSessionManagerTests
});
}
private sealed class MultiVerseRouteSource : ICultMeshLookupSource
{
private readonly ManualSessionClock _clock;
public MultiVerseRouteSource(ManualSessionClock clock) => _clock = clock;
public string SourceId => "odin";
public Task<IReadOnlyList<CultMeshDiscoveryObservation>> LookupAsync(
CultMeshDiscoveryQuery query,
CancellationToken cancellationToken = default)
{
var expires = _clock.UtcNow.AddMinutes(1);
return Task.FromResult<IReadOnlyList<CultMeshDiscoveryObservation>>(new[]
{
Observation("norn", "rudp://norn:3076", expires),
Observation("aetheria", "rudp://aetheria:3076", expires)
});
}
private CultMeshDiscoveryObservation Observation(string verseId, string endpoint, DateTimeOffset expires) =>
new(
new CultMeshVerseDescriptor(
verseId,
verseId,
CultMeshVerseAuthorityModel.OperatorCluster,
new CultMeshVerseCompatibility("cultmesh.v0", "rules"),
new[] { endpoint }),
SourceId,
_clock.UtcNow,
expires,
CultMeshDiscoveryTrust.Signed);
}
private sealed class FakeConnector : ICultMeshTransportConnector
{
private readonly Func<CultMeshTransportCandidate, CancellationToken, Task<ICultNetSchemaClient>> _connect;

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

@ -39,7 +39,7 @@ public sealed class CultMeshTransportModularityTests
});
var session = await sessions.ConnectAsync(
CultMeshEndpointId.Parse("service:aetheria.daemon"),
CultMeshEndpointId.Parse("aetheria"),
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("service:aetheria.daemon"));
var session = await sessions.ConnectContentAsync(CultMeshEndpointId.Parse("aetheria"));
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("service:aetheria.daemon"));
CultMeshEndpointId.Parse("aetheria"));
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("service:aetheria.daemon"));
CultMeshEndpointId.Parse("aetheria"));
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("service:aetheria.daemon"));
CultMeshEndpointId.Parse("aetheria"));
await connect.Should().ThrowAsync<CultMeshSessionException>()
.WithMessage("*No realtime state connectors are configured*");

2
unity/org.gamecult.cultlib/package.json
View file

@ -1,7 +1,7 @@
{
"name": "org.gamecult.cultlib",
"displayName": "CultLib",
"version": "1.0.45",
"version": "1.0.46",
"unity": "2021.3",
"description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.",
"license": "MIT",