Add CultMesh peer exchange client

This commit is contained in:
Meta Vens 2026-05-20 01:41:12 +01:00
commit 46cb4820ff
6 changed files with 184 additions and 2 deletions

View file

@ -235,6 +235,15 @@ namespace GameCult.Mesh
return new CultMeshPeerExchangeServer(node.Server, catalog);
}
/// <summary>
/// Creates a client for fetching peer cards from peer exchange endpoints.
/// </summary>
public static CultMeshPeerExchangeClient CreatePeerExchangeClient(
CultMeshPeerExchangeClientOptions? options = null)
{
return new CultMeshPeerExchangeClient(options);
}
/// <summary>
/// Creates a CultNet client configured for CultMesh use.
/// </summary>

View file

@ -285,4 +285,150 @@ namespace GameCult.Mesh
return Task.CompletedTask;
}
}
/// <summary>
/// Options for schema-v0 CultMesh peer exchange clients.
/// </summary>
public sealed class CultMeshPeerExchangeClientOptions
{
/// <summary>
/// Gets or sets client security options used to connect to peer exchange endpoints.
/// </summary>
public ClientSecurityOptions? Security { get; set; }
/// <summary>
/// Gets or sets how long to wait for a connection before failing.
/// </summary>
public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(5);
/// <summary>
/// Gets or sets how long to wait for a peer exchange response before failing.
/// </summary>
public TimeSpan ResponseTimeout { get; set; } = TimeSpan.FromSeconds(5);
/// <summary>
/// Gets or sets a callback used to customize each ephemeral exchange client.
/// </summary>
public Action<Client>? ConfigureClient { get; set; }
}
/// <summary>
/// Fetches peer cards from CultMesh peers.
/// </summary>
public sealed class CultMeshPeerExchangeClient
{
private readonly CultMeshPeerExchangeClientOptions _options;
/// <summary>
/// Creates a peer exchange client.
/// </summary>
public CultMeshPeerExchangeClient(CultMeshPeerExchangeClientOptions? options = null)
{
_options = options ?? new CultMeshPeerExchangeClientOptions();
}
/// <summary>
/// Fetches peer cards from one endpoint.
/// </summary>
public async Task<CultMeshPeerExchangeResponseMessage> 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<CultMeshPeerExchangeResponseMessage>(
TaskCreationOptions.RunContinuationsAsynchronously);
using var client = new Client(_options.Security ?? ClientSecurityOptions.Development())
{
AllowUnverifiedCultNetMessages = true
};
_options.ConfigureClient?.Invoke(client);
client.OnCultNet<CultMeshPeerExchangeResponseMessage>(response =>
{
if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal))
{
completion.TrySetResult(response);
}
});
client.OnCultNet<CultNetErrorMessage>(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);
}
/// <summary>
/// Fetches peer cards from endpoints and upserts every response into a local peer catalog.
/// </summary>
public async Task<int> DiscoverAsync(
CultMeshPeerCatalog catalog,
IEnumerable<string> endpoints,
string verseId,
IEnumerable<string>? 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<CultMeshPeerExchangeResponseMessage> WaitForResponseAsync(
Task<CultMeshPeerExchangeResponseMessage> 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);
}
}
}

View file

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

View file

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

View file

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

View file

@ -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()
{