From 46cb4820ff6364b009d04f356326dc23f1d478d1 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 20 May 2026 01:41:12 +0100 Subject: [PATCH] Add CultMesh peer exchange client --- src/GameCult.Mesh/CultMesh.cs | 9 ++ src/GameCult.Mesh/CultMeshPeerExchange.cs | 146 ++++++++++++++++++ src/GameCult.Mesh/README.md | 11 ++ src/GameCult.Mesh/docs/public-api.md | 3 + .../Contracts/cultnet-distributed-database.md | 5 +- .../NetworkingTests.cs | 12 ++ 6 files changed, 184 insertions(+), 2 deletions(-) diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index 68a7e0b8..31a4ab42 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -235,6 +235,15 @@ namespace GameCult.Mesh return new CultMeshPeerExchangeServer(node.Server, catalog); } + /// + /// Creates a client for fetching peer cards from peer exchange endpoints. + /// + public static CultMeshPeerExchangeClient CreatePeerExchangeClient( + CultMeshPeerExchangeClientOptions? options = null) + { + return new CultMeshPeerExchangeClient(options); + } + /// /// Creates a CultNet client configured for CultMesh use. /// diff --git a/src/GameCult.Mesh/CultMeshPeerExchange.cs b/src/GameCult.Mesh/CultMeshPeerExchange.cs index ac4a36ce..85b959ac 100644 --- a/src/GameCult.Mesh/CultMeshPeerExchange.cs +++ b/src/GameCult.Mesh/CultMeshPeerExchange.cs @@ -285,4 +285,150 @@ namespace GameCult.Mesh return Task.CompletedTask; } } + + /// + /// Options for schema-v0 CultMesh peer exchange clients. + /// + public sealed class CultMeshPeerExchangeClientOptions + { + /// + /// Gets or sets client security options used to connect to peer exchange endpoints. + /// + public ClientSecurityOptions? Security { get; set; } + + /// + /// Gets or sets how long to wait for a connection before failing. + /// + public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(5); + + /// + /// Gets or sets how long to wait for a peer exchange response before failing. + /// + public TimeSpan ResponseTimeout { get; set; } = TimeSpan.FromSeconds(5); + + /// + /// Gets or sets a callback used to customize each ephemeral exchange client. + /// + public Action? ConfigureClient { get; set; } + } + + /// + /// Fetches peer cards from CultMesh peers. + /// + public sealed class CultMeshPeerExchangeClient + { + private readonly CultMeshPeerExchangeClientOptions _options; + + /// + /// Creates a peer exchange client. + /// + public CultMeshPeerExchangeClient(CultMeshPeerExchangeClientOptions? options = null) + { + _options = options ?? new CultMeshPeerExchangeClientOptions(); + } + + /// + /// Fetches peer cards from one endpoint. + /// + public async Task FetchAsync( + string endpoint, + CultMeshPeerExchangeRequestMessage request) + { + if (string.IsNullOrWhiteSpace(endpoint)) throw new ArgumentException("Value must be non-empty.", nameof(endpoint)); + if (request == null) throw new ArgumentNullException(nameof(request)); + if (string.IsNullOrWhiteSpace(request.VerseId)) throw new ArgumentException("Peer exchange requires a verseId.", nameof(request)); + + var (host, port) = CultNetSchemaWriteForwarder.ParseEndpoint(endpoint); + var messageId = Guid.NewGuid().ToString("N"); + var completion = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + + using var client = new Client(_options.Security ?? ClientSecurityOptions.Development()) + { + AllowUnverifiedCultNetMessages = true + }; + _options.ConfigureClient?.Invoke(client); + client.OnCultNet(response => + { + if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) + { + completion.TrySetResult(response); + } + }); + client.OnCultNet(error => + completion.TrySetException(new InvalidOperationException(error.Error))); + + client.Connect(host, port); + await WaitForConnectionAsync(client, endpoint).ConfigureAwait(false); + client.SendCultNet(new CultMeshPeerExchangeRequestMessage + { + MessageId = messageId, + VerseId = request.VerseId, + Roles = request.Roles, + KnownPeerIds = request.KnownPeerIds, + Limit = request.Limit + }); + + return await WaitForResponseAsync(completion.Task, endpoint).ConfigureAwait(false); + } + + /// + /// Fetches peer cards from endpoints and upserts every response into a local peer catalog. + /// + public async Task DiscoverAsync( + CultMeshPeerCatalog catalog, + IEnumerable endpoints, + string verseId, + IEnumerable? roles = null, + int? limit = null) + { + if (catalog == null) throw new ArgumentNullException(nameof(catalog)); + if (endpoints == null) throw new ArgumentNullException(nameof(endpoints)); + if (string.IsNullOrWhiteSpace(verseId)) throw new ArgumentException("Value must be non-empty.", nameof(verseId)); + + var count = 0; + foreach (var endpoint in endpoints.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal)) + { + var response = await FetchAsync(endpoint, new CultMeshPeerExchangeRequestMessage + { + VerseId = verseId, + Roles = roles?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(), + KnownPeerIds = catalog.Find(verseId).Select(peer => peer.PeerId).ToArray(), + Limit = limit + }).ConfigureAwait(false); + catalog.Upsert(response); + count += response.Peers.Length; + } + + return count; + } + + private async Task WaitForConnectionAsync(Client client, string endpoint) + { + var deadline = DateTimeOffset.UtcNow + _options.ConnectTimeout; + while (!client.Connected) + { + if (DateTimeOffset.UtcNow >= deadline) + { + throw new TimeoutException($"Timed out connecting to peer exchange endpoint {endpoint}."); + } + + await Task.Delay(TimeSpan.FromMilliseconds(25)).ConfigureAwait(false); + } + } + + private async Task WaitForResponseAsync( + Task responseTask, + string endpoint) + { + var timeoutTask = Task.Delay(_options.ResponseTimeout); + var completed = await Task.WhenAny(responseTask, timeoutTask).ConfigureAwait(false); + if (completed != responseTask) + { + throw new TimeoutException($"Timed out waiting for peer exchange response from {endpoint}."); + } + + return await responseTask.ConfigureAwait(false); + } + } } diff --git a/src/GameCult.Mesh/README.md b/src/GameCult.Mesh/README.md index 0f641e87..cf99b835 100644 --- a/src/GameCult.Mesh/README.md +++ b/src/GameCult.Mesh/README.md @@ -92,6 +92,17 @@ Peer cards are contact candidates, not authority by themselves. Public Verses should validate authority leases or signatures before trusting a peer for committed state. +Clients can fetch more peers from known peers: + +```csharp +var exchangeClient = CultMesh.CreatePeerExchangeClient(); +await exchangeClient.DiscoverAsync( + peers, + aetheria.DiscoveryEndpoints, + "aetheria-main", + roles: [CultMeshPeerRoles.ReadReplica]); +``` + ## Client-Side Prediction CultMesh lets a runtime declare which input documents it may author diff --git a/src/GameCult.Mesh/docs/public-api.md b/src/GameCult.Mesh/docs/public-api.md index ea3ba74f..92664d7e 100644 --- a/src/GameCult.Mesh/docs/public-api.md +++ b/src/GameCult.Mesh/docs/public-api.md @@ -13,6 +13,7 @@ Namespace: `GameCult.Mesh` - `CultMesh.CreateVerseCatalog()` - `CultMesh.CreatePeerCatalog()` - `CultMesh.CreateVerseDiscoveryClient(...)` +- `CultMesh.CreatePeerExchangeClient(...)` - `CultMesh.CreateClient(...)` - `CultMesh.ConnectClient(...)` @@ -60,6 +61,8 @@ signature. `CultMesh.ServePeerExchange(node, catalog)` attaches schema-v0 peer exchange responses to a node. The wire contracts are `cultmesh.peer_exchange_request.v0` and `cultmesh.peer_exchange_response.v0`. +`CultMeshPeerExchangeClient` can fetch peer cards from known endpoints and +upsert them into a local catalog. Peer cards are discovery hints; authority still requires a valid lease or signature for the target Verse. diff --git a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md index 8bcbf0e7..ed93a567 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -72,7 +72,8 @@ Implemented: - schema-v0 Verse catalog request/response messages - CultMesh Verse discovery server bridge over the local Verse catalog - schema-v0 peer exchange request/response messages -- CultMesh peer cards, peer catalog, and peer exchange server bridge +- CultMesh peer cards, peer catalog, peer exchange server bridge, and peer + exchange client - client-authority scopes for locally predicted input documents - predicted and reconciled database change kinds for client-side prediction - simulation witness observations for mesh-side opinions about frame facts @@ -127,7 +128,7 @@ Not implemented yet: Add discovery fanout and operational polish: -- add peer exchange clients and periodic gossip fanout +- add periodic gossip fanout over known peer cards - add snapshot throttling and size limits for public mesh edges - after that, add simulation-frame rollback/resimulation helpers around predicted input streams diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 245d45b9..b23396d0 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -831,6 +831,18 @@ namespace GameCult.Networking.Tests Assert.That(client, Is.Not.Null); } + [Test] + public void CultMesh_CreatePeerExchangeClient_Returns_ExchangeClient() + { + var client = CultMesh.CreatePeerExchangeClient(new CultMeshPeerExchangeClientOptions + { + ConnectTimeout = TimeSpan.FromMilliseconds(250), + ResponseTimeout = TimeSpan.FromMilliseconds(250) + }); + + Assert.That(client, Is.Not.Null); + } + [Test] public void CultMeshVerseCatalog_Finds_CompatibleTransferTargets() {