Add CultMesh peer exchange catalogs
This commit is contained in:
parent
dfa7db607a
commit
c612eb4307
10 changed files with 718 additions and 1 deletions
|
|
@ -196,6 +196,14 @@ namespace GameCult.Mesh
|
|||
return new CultMeshVerseCatalog();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Creates a local reactive catalog for discovered peers.
|
||||
/// </summary>
|
||||
public static CultMeshPeerCatalog CreatePeerCatalog()
|
||||
{
|
||||
return new CultMeshPeerCatalog();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Attaches Verse discovery responses to a CultMesh node.
|
||||
/// </summary>
|
||||
|
|
@ -207,6 +215,26 @@ namespace GameCult.Mesh
|
|||
return new CultMeshVerseDiscoveryServer(node.Server, catalog);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Creates a client for fetching Verse catalogs from discovery endpoints.
|
||||
/// </summary>
|
||||
public static CultMeshVerseDiscoveryClient CreateVerseDiscoveryClient(
|
||||
CultMeshVerseDiscoveryClientOptions? options = null)
|
||||
{
|
||||
return new CultMeshVerseDiscoveryClient(options);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Attaches peer exchange responses to a CultMesh node.
|
||||
/// </summary>
|
||||
public static CultMeshPeerExchangeServer ServePeerExchange(
|
||||
CultMeshNode node,
|
||||
CultMeshPeerCatalog catalog)
|
||||
{
|
||||
if (node == null) throw new ArgumentNullException(nameof(node));
|
||||
return new CultMeshPeerExchangeServer(node.Server, catalog);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Creates a CultNet client configured for CultMesh use.
|
||||
/// </summary>
|
||||
|
|
|
|||
288
src/GameCult.Mesh/CultMeshPeerExchange.cs
Normal file
288
src/GameCult.Mesh/CultMeshPeerExchange.cs
Normal file
|
|
@ -0,0 +1,288 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Threading.Tasks;
|
||||
using GameCult.Networking;
|
||||
using LiteNetLib;
|
||||
using R3;
|
||||
|
||||
namespace GameCult.Mesh
|
||||
{
|
||||
/// <summary>
|
||||
/// Well-known CultMesh peer roles.
|
||||
/// </summary>
|
||||
public static class CultMeshPeerRoles
|
||||
{
|
||||
/// <summary>Peer can answer discovery requests.</summary>
|
||||
public const string Discovery = "discovery";
|
||||
/// <summary>Peer can serve authoritative shard logs.</summary>
|
||||
public const string ShardPrimary = "shard-primary";
|
||||
/// <summary>Peer can serve replicated shard logs.</summary>
|
||||
public const string ShardReplica = "shard-replica";
|
||||
/// <summary>Peer can serve read traffic.</summary>
|
||||
public const string ReadReplica = "read-replica";
|
||||
/// <summary>Peer can observe simulation facts.</summary>
|
||||
public const string SimulationObserver = "simulation-observer";
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Candidate peer contact information for a Verse.
|
||||
/// </summary>
|
||||
public sealed class CultMeshPeerCard
|
||||
{
|
||||
/// <summary>
|
||||
/// Creates a peer card.
|
||||
/// </summary>
|
||||
public CultMeshPeerCard(
|
||||
string peerId,
|
||||
string verseId,
|
||||
IEnumerable<string> endpoints,
|
||||
IEnumerable<string>? roles = null,
|
||||
IEnumerable<string>? shardIds = null,
|
||||
string? region = null,
|
||||
string? authorityLeaseId = null,
|
||||
string? expiresAt = null,
|
||||
string? signature = null)
|
||||
{
|
||||
PeerId = RequireNonEmpty(peerId, nameof(peerId));
|
||||
VerseId = RequireNonEmpty(verseId, nameof(verseId));
|
||||
Endpoints = Clean(endpoints);
|
||||
Roles = Clean(roles);
|
||||
ShardIds = Clean(shardIds);
|
||||
Region = region;
|
||||
AuthorityLeaseId = authorityLeaseId;
|
||||
ExpiresAt = expiresAt;
|
||||
Signature = signature;
|
||||
}
|
||||
|
||||
/// <summary>Gets the stable peer id.</summary>
|
||||
public string PeerId { get; }
|
||||
/// <summary>Gets the Verse id this peer participates in.</summary>
|
||||
public string VerseId { get; }
|
||||
/// <summary>Gets reachable endpoints.</summary>
|
||||
public IReadOnlyList<string> Endpoints { get; }
|
||||
/// <summary>Gets advertised roles.</summary>
|
||||
public IReadOnlyList<string> Roles { get; }
|
||||
/// <summary>Gets shard ids this peer can serve or observe.</summary>
|
||||
public IReadOnlyList<string> ShardIds { get; }
|
||||
/// <summary>Gets an optional region/locality label.</summary>
|
||||
public string? Region { get; }
|
||||
/// <summary>Gets the authority lease id, when present.</summary>
|
||||
public string? AuthorityLeaseId { get; }
|
||||
/// <summary>Gets the expiry timestamp for this card.</summary>
|
||||
public string? ExpiresAt { get; }
|
||||
/// <summary>Gets an optional signature over the card.</summary>
|
||||
public string? Signature { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Returns whether the card advertises a role.
|
||||
/// </summary>
|
||||
public bool HasRole(string role)
|
||||
{
|
||||
return Roles.Contains(role, StringComparer.Ordinal);
|
||||
}
|
||||
|
||||
private static string[] Clean(IEnumerable<string>? values)
|
||||
{
|
||||
return values?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray()
|
||||
?? Array.Empty<string>();
|
||||
}
|
||||
|
||||
private static string RequireNonEmpty(string value, string paramName)
|
||||
{
|
||||
return string.IsNullOrWhiteSpace(value)
|
||||
? throw new ArgumentException("Value must be non-empty.", paramName)
|
||||
: value;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Conversion helpers for CultMesh peer exchange messages.
|
||||
/// </summary>
|
||||
public static class CultMeshPeerMessages
|
||||
{
|
||||
/// <summary>
|
||||
/// Converts a peer card to its schema-v0 wire shape.
|
||||
/// </summary>
|
||||
public static CultMeshPeerCardMessage ToMessage(this CultMeshPeerCard card)
|
||||
{
|
||||
if (card == null) throw new ArgumentNullException(nameof(card));
|
||||
return new CultMeshPeerCardMessage
|
||||
{
|
||||
PeerId = card.PeerId,
|
||||
VerseId = card.VerseId,
|
||||
Endpoints = card.Endpoints.ToArray(),
|
||||
Roles = card.Roles.ToArray(),
|
||||
ShardIds = card.ShardIds.ToArray(),
|
||||
Region = card.Region,
|
||||
AuthorityLeaseId = card.AuthorityLeaseId,
|
||||
ExpiresAt = card.ExpiresAt,
|
||||
Signature = card.Signature
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Converts a wire peer card to the local public API shape.
|
||||
/// </summary>
|
||||
public static CultMeshPeerCard ToPeerCard(this CultMeshPeerCardMessage message)
|
||||
{
|
||||
if (message == null) throw new ArgumentNullException(nameof(message));
|
||||
return new CultMeshPeerCard(
|
||||
message.PeerId,
|
||||
message.VerseId,
|
||||
message.Endpoints,
|
||||
message.Roles,
|
||||
message.ShardIds,
|
||||
message.Region,
|
||||
message.AuthorityLeaseId,
|
||||
message.ExpiresAt,
|
||||
message.Signature);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Reactive local catalog for discovered CultMesh peers.
|
||||
/// </summary>
|
||||
public sealed class CultMeshPeerCatalog : IDisposable
|
||||
{
|
||||
private readonly Dictionary<string, CultMeshPeerCard> _peers = new(StringComparer.Ordinal);
|
||||
private readonly Subject<CultMeshPeerCard> _updates = new();
|
||||
private bool _disposed;
|
||||
|
||||
/// <summary>Gets all known peers.</summary>
|
||||
public IReadOnlyList<CultMeshPeerCard> Peers => _peers.Values.OrderBy(peer => peer.PeerId, StringComparer.Ordinal).ToArray();
|
||||
|
||||
/// <summary>Watches peer card updates.</summary>
|
||||
public Observable<CultMeshPeerCard> Watch()
|
||||
{
|
||||
ThrowIfDisposed();
|
||||
return _updates;
|
||||
}
|
||||
|
||||
/// <summary>Adds or replaces a peer card.</summary>
|
||||
public void Upsert(CultMeshPeerCard card)
|
||||
{
|
||||
ThrowIfDisposed();
|
||||
if (card == null) throw new ArgumentNullException(nameof(card));
|
||||
_peers[card.PeerId] = card;
|
||||
_updates.OnNext(card);
|
||||
}
|
||||
|
||||
/// <summary>Adds or replaces peer cards from a wire response.</summary>
|
||||
public void Upsert(CultMeshPeerExchangeResponseMessage response)
|
||||
{
|
||||
ThrowIfDisposed();
|
||||
if (response == null) throw new ArgumentNullException(nameof(response));
|
||||
foreach (var peer in response.Peers)
|
||||
{
|
||||
Upsert(peer.ToPeerCard());
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Finds peers by Verse and optional role.</summary>
|
||||
public IReadOnlyList<CultMeshPeerCard> Find(string verseId, string? role = null)
|
||||
{
|
||||
ThrowIfDisposed();
|
||||
if (string.IsNullOrWhiteSpace(verseId)) throw new ArgumentException("Value must be non-empty.", nameof(verseId));
|
||||
return _peers.Values
|
||||
.Where(peer => string.Equals(peer.VerseId, verseId, StringComparison.Ordinal))
|
||||
.Where(peer => string.IsNullOrWhiteSpace(role) || peer.HasRole(role!))
|
||||
.OrderBy(peer => peer.PeerId, StringComparer.Ordinal)
|
||||
.ToArray();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Dispose()
|
||||
{
|
||||
if (_disposed)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_disposed = true;
|
||||
_updates.Dispose();
|
||||
_peers.Clear();
|
||||
}
|
||||
|
||||
private void ThrowIfDisposed()
|
||||
{
|
||||
if (_disposed)
|
||||
{
|
||||
throw new ObjectDisposedException(nameof(CultMeshPeerCatalog));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Answers CultMesh peer exchange requests from a local peer catalog.
|
||||
/// </summary>
|
||||
public sealed class CultMeshPeerExchangeServer : IDisposable
|
||||
{
|
||||
private readonly Server _server;
|
||||
private readonly CultMeshPeerCatalog _catalog;
|
||||
private readonly Func<CultMeshPeerExchangeRequestMessage, NetPeer, Task> _requestHandler;
|
||||
private bool _disposed;
|
||||
|
||||
/// <summary>
|
||||
/// Creates and attaches a peer exchange bridge.
|
||||
/// </summary>
|
||||
public CultMeshPeerExchangeServer(Server server, CultMeshPeerCatalog catalog)
|
||||
{
|
||||
_server = server ?? throw new ArgumentNullException(nameof(server));
|
||||
_catalog = catalog ?? throw new ArgumentNullException(nameof(catalog));
|
||||
_requestHandler = HandleRequestAsync;
|
||||
_server.OnCultNet(_requestHandler);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Creates a peer exchange response for a request.
|
||||
/// </summary>
|
||||
public CultMeshPeerExchangeResponseMessage CreateResponse(CultMeshPeerExchangeRequestMessage request)
|
||||
{
|
||||
if (request == null) throw new ArgumentNullException(nameof(request));
|
||||
if (string.IsNullOrWhiteSpace(request.VerseId))
|
||||
{
|
||||
throw new ArgumentException("Peer exchange requires a verseId.", nameof(request));
|
||||
}
|
||||
|
||||
var known = request.KnownPeerIds == null || request.KnownPeerIds.Length == 0
|
||||
? null
|
||||
: new HashSet<string>(request.KnownPeerIds, StringComparer.Ordinal);
|
||||
var roles = request.Roles == null || request.Roles.Length == 0
|
||||
? null
|
||||
: new HashSet<string>(request.Roles, StringComparer.Ordinal);
|
||||
var peers = _catalog.Find(request.VerseId)
|
||||
.Where(peer => known == null || !known.Contains(peer.PeerId))
|
||||
.Where(peer => roles == null || peer.Roles.Any(role => roles.Contains(role)))
|
||||
.Take(request.Limit ?? int.MaxValue)
|
||||
.Select(peer => peer.ToMessage())
|
||||
.ToArray();
|
||||
|
||||
return new CultMeshPeerExchangeResponseMessage
|
||||
{
|
||||
MessageId = string.IsNullOrWhiteSpace(request.MessageId)
|
||||
? Guid.NewGuid().ToString("N")
|
||||
: request.MessageId,
|
||||
Peers = peers
|
||||
};
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Dispose()
|
||||
{
|
||||
if (_disposed)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_disposed = true;
|
||||
_server.RemoveCultNetMessageListener<CultMeshPeerExchangeRequestMessage>(_requestHandler);
|
||||
}
|
||||
|
||||
private Task HandleRequestAsync(CultMeshPeerExchangeRequestMessage request, NetPeer peer)
|
||||
{
|
||||
peer.SendCultNet(CreateResponse(request));
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -131,4 +131,139 @@ namespace GameCult.Mesh
|
|||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Options for schema-v0 CultMesh Verse discovery clients.
|
||||
/// </summary>
|
||||
public sealed class CultMeshVerseDiscoveryClientOptions
|
||||
{
|
||||
/// <summary>
|
||||
/// Gets or sets client security options used to connect to discovery 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 discovery response before failing.
|
||||
/// </summary>
|
||||
public TimeSpan ResponseTimeout { get; set; } = TimeSpan.FromSeconds(5);
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets a callback used to customize each ephemeral discovery client.
|
||||
/// </summary>
|
||||
public Action<Client>? ConfigureClient { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Fetches CultMesh Verse catalogs from discovery endpoints.
|
||||
/// </summary>
|
||||
public sealed class CultMeshVerseDiscoveryClient
|
||||
{
|
||||
private readonly CultMeshVerseDiscoveryClientOptions _options;
|
||||
|
||||
/// <summary>
|
||||
/// Creates a Verse discovery client.
|
||||
/// </summary>
|
||||
public CultMeshVerseDiscoveryClient(CultMeshVerseDiscoveryClientOptions? options = null)
|
||||
{
|
||||
_options = options ?? new CultMeshVerseDiscoveryClientOptions();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Fetches a Verse catalog response from one discovery endpoint.
|
||||
/// </summary>
|
||||
public async Task<CultMeshVerseCatalogResponseMessage> FetchAsync(
|
||||
string endpoint,
|
||||
CultMeshVerseCatalogRequestMessage? request = null)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(endpoint)) throw new ArgumentException("Value must be non-empty.", nameof(endpoint));
|
||||
var (host, port) = CultNetSchemaWriteForwarder.ParseEndpoint(endpoint);
|
||||
var messageId = Guid.NewGuid().ToString("N");
|
||||
var completion = new TaskCompletionSource<CultMeshVerseCatalogResponseMessage>(
|
||||
TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
|
||||
using var client = new Client(_options.Security ?? ClientSecurityOptions.Development())
|
||||
{
|
||||
AllowUnverifiedCultNetMessages = true
|
||||
};
|
||||
_options.ConfigureClient?.Invoke(client);
|
||||
client.OnCultNet<CultMeshVerseCatalogResponseMessage>(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 CultMeshVerseCatalogRequestMessage
|
||||
{
|
||||
MessageId = messageId,
|
||||
TransportVersion = request?.TransportVersion,
|
||||
VerseIds = request?.VerseIds
|
||||
});
|
||||
|
||||
return await WaitForResponseAsync(completion.Task, endpoint).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Fetches Verse catalogs from endpoints and upserts every response into a local catalog.
|
||||
/// </summary>
|
||||
public async Task<int> DiscoverAsync(
|
||||
CultMeshVerseCatalog catalog,
|
||||
IEnumerable<string> endpoints,
|
||||
string? transportVersion = null)
|
||||
{
|
||||
if (catalog == null) throw new ArgumentNullException(nameof(catalog));
|
||||
if (endpoints == null) throw new ArgumentNullException(nameof(endpoints));
|
||||
|
||||
var count = 0;
|
||||
foreach (var endpoint in endpoints.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal))
|
||||
{
|
||||
var response = await FetchAsync(endpoint, new CultMeshVerseCatalogRequestMessage
|
||||
{
|
||||
TransportVersion = transportVersion
|
||||
}).ConfigureAwait(false);
|
||||
catalog.Upsert(response);
|
||||
count += response.Verses.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 Verse discovery endpoint {endpoint}.");
|
||||
}
|
||||
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(25)).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<CultMeshVerseCatalogResponseMessage> WaitForResponseAsync(
|
||||
Task<CultMeshVerseCatalogResponseMessage> 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 Verse discovery response from {endpoint}.");
|
||||
}
|
||||
|
||||
return await responseTask.ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -70,6 +70,28 @@ Peers can request Verse catalogs over schema-v0 with
|
|||
`cultmesh.verse_catalog_request.v0` and receive transport-filtered
|
||||
`cultmesh.verse_catalog_response.v0` descriptors.
|
||||
|
||||
## Peer Exchange
|
||||
|
||||
Verse discovery tells a node which consensus graph exists. Peer exchange tells
|
||||
it who else is already in that graph.
|
||||
|
||||
```csharp
|
||||
using var peers = CultMesh.CreatePeerCatalog();
|
||||
peers.Upsert(new CultMeshPeerCard(
|
||||
"gc-us-east-1",
|
||||
"aetheria-main",
|
||||
["cultnet://us-east.aetheria.gamecult.net:3075"],
|
||||
roles: [CultMeshPeerRoles.Discovery, CultMeshPeerRoles.ShardPrimary],
|
||||
shardIds: ["players-us-east"],
|
||||
authorityLeaseId: "lease:players-us-east:42"));
|
||||
|
||||
using var exchange = CultMesh.ServePeerExchange(node, peers);
|
||||
```
|
||||
|
||||
Peer cards are contact candidates, not authority by themselves. Public Verses
|
||||
should validate authority leases or signatures before trusting a peer for
|
||||
committed state.
|
||||
|
||||
## Client-Side Prediction
|
||||
|
||||
CultMesh lets a runtime declare which input documents it may author
|
||||
|
|
|
|||
|
|
@ -11,6 +11,8 @@ Namespace: `GameCult.Mesh`
|
|||
- `CultMesh.CreateNodeAsync(...)`
|
||||
- `CultMesh.StartNodeAsync(...)`
|
||||
- `CultMesh.CreateVerseCatalog()`
|
||||
- `CultMesh.CreatePeerCatalog()`
|
||||
- `CultMesh.CreateVerseDiscoveryClient(...)`
|
||||
- `CultMesh.CreateClient(...)`
|
||||
- `CultMesh.ConnectClient(...)`
|
||||
|
||||
|
|
@ -48,6 +50,19 @@ responses to a node. The wire contracts are
|
|||
Consumers can apply a response directly with
|
||||
`CultMeshVerseCatalog.Upsert(CultMeshVerseCatalogResponseMessage)`.
|
||||
|
||||
### Peer Exchange
|
||||
|
||||
`CultMeshPeerCatalog` is the local reactive catalog for peer cards.
|
||||
`CultMeshPeerCard` describes a candidate peer endpoint for one Verse with
|
||||
roles, shard hints, optional region, optional authority lease id, expiry, and
|
||||
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`.
|
||||
Peer cards are discovery hints; authority still requires a valid lease or
|
||||
signature for the target Verse.
|
||||
|
||||
### Shard Authority
|
||||
|
||||
Each shard has one primary writer for now. Non-primary writes are rejected or
|
||||
|
|
|
|||
|
|
@ -56,3 +56,11 @@ Authority is a Verse-level policy, not a transport accident.
|
|||
static config, cloud discovery, peer gossip, or mod portals. It does not care
|
||||
where the descriptor came from. That keeps discovery pluggable while the Verse
|
||||
model stays stable.
|
||||
|
||||
Peer exchange is the torrent-shaped layer under Verse discovery. A Verse
|
||||
catalog says which graph exists; a peer catalog says which nodes can currently
|
||||
serve or observe parts of that graph. Peer cards can advertise discovery,
|
||||
primary shard, replica, read replica, and simulation observer roles, but those
|
||||
cards do not grant authority by themselves. Public operator Verses should treat
|
||||
authority lease ids and signatures as mandatory before accepting committed
|
||||
state from a peer.
|
||||
|
|
|
|||
|
|
@ -71,6 +71,8 @@ Implemented:
|
|||
`.cultmesh/shard-logs` path beside the cache file
|
||||
- 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
|
||||
- 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
|
||||
|
|
@ -118,12 +120,14 @@ Not implemented yet:
|
|||
- Consensus candidates are derived from observations. They may inform the
|
||||
authoritative commit path, but they are not themselves committed world state
|
||||
until a shard owner writes them into the log.
|
||||
- Peer cards are contact candidates. They can advertise roles and authority
|
||||
lease ids, but they do not grant authority without Verse-specific validation.
|
||||
|
||||
## Next Coherent Slice
|
||||
|
||||
Add discovery fanout and operational polish:
|
||||
|
||||
- add peer-to-peer Verse discovery fetchers and gossip fanout
|
||||
- add peer exchange clients and periodic gossip fanout
|
||||
- add snapshot throttling and size limits for public mesh edges
|
||||
- after that, add simulation-frame rollback/resimulation helpers around
|
||||
predicted input streams
|
||||
|
|
|
|||
|
|
@ -70,6 +70,8 @@ namespace GameCult.Networking
|
|||
CultNetSchemaVersions.SimulationConsensusCandidate => Deserialize<CultNetSimulationConsensusCandidateMessage>(payload),
|
||||
CultNetSchemaVersions.VerseCatalogRequest => Deserialize<CultMeshVerseCatalogRequestMessage>(payload),
|
||||
CultNetSchemaVersions.VerseCatalogResponse => Deserialize<CultMeshVerseCatalogResponseMessage>(payload),
|
||||
CultNetSchemaVersions.PeerExchangeRequest => Deserialize<CultMeshPeerExchangeRequestMessage>(payload),
|
||||
CultNetSchemaVersions.PeerExchangeResponse => Deserialize<CultMeshPeerExchangeResponseMessage>(payload),
|
||||
CultNetSchemaVersions.SchemaCatalogRequest => Deserialize<CultNetSchemaCatalogRequestMessage>(payload),
|
||||
CultNetSchemaVersions.SchemaCatalogResponse => Deserialize<CultNetSchemaCatalogResponseMessage>(payload),
|
||||
_ => throw new MessagePackSerializationException(
|
||||
|
|
|
|||
|
|
@ -123,6 +123,14 @@ namespace GameCult.Networking
|
|||
/// CultMesh Verse catalog response contract identifier.
|
||||
/// </summary>
|
||||
public const string VerseCatalogResponse = "cultmesh.verse_catalog_response.v0";
|
||||
/// <summary>
|
||||
/// CultMesh peer exchange request contract identifier.
|
||||
/// </summary>
|
||||
public const string PeerExchangeRequest = "cultmesh.peer_exchange_request.v0";
|
||||
/// <summary>
|
||||
/// CultMesh peer exchange response contract identifier.
|
||||
/// </summary>
|
||||
public const string PeerExchangeResponse = "cultmesh.peer_exchange_response.v0";
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
|
@ -1077,6 +1085,102 @@ namespace GameCult.Networking
|
|||
[Key("optionalPluginIds")] public string[] OptionalPluginIds { get; set; } = Array.Empty<string>();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Requests known peers for a CultMesh Verse.
|
||||
/// </summary>
|
||||
[MessagePackObject]
|
||||
public class CultMeshPeerExchangeRequestMessage : ICultNetSchemaMessage
|
||||
{
|
||||
/// <summary>
|
||||
/// Gets or sets the schema version.
|
||||
/// </summary>
|
||||
[Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.PeerExchangeRequest;
|
||||
/// <summary>
|
||||
/// Gets or sets the request id.
|
||||
/// </summary>
|
||||
[Key("messageId")] public string MessageId { get; set; } = string.Empty;
|
||||
/// <summary>
|
||||
/// Gets or sets the Verse id to discover peers for.
|
||||
/// </summary>
|
||||
[Key("verseId")] public string VerseId { get; set; } = string.Empty;
|
||||
/// <summary>
|
||||
/// Gets or sets optional role filters.
|
||||
/// </summary>
|
||||
[Key("roles")] public string[]? Roles { get; set; }
|
||||
/// <summary>
|
||||
/// Gets or sets peer ids the requester already knows.
|
||||
/// </summary>
|
||||
[Key("knownPeerIds")] public string[]? KnownPeerIds { get; set; }
|
||||
/// <summary>
|
||||
/// Gets or sets the maximum number of peer cards to return.
|
||||
/// </summary>
|
||||
[Key("limit")] public int? Limit { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns known peers for a CultMesh Verse.
|
||||
/// </summary>
|
||||
[MessagePackObject]
|
||||
public class CultMeshPeerExchangeResponseMessage : ICultNetSchemaMessage
|
||||
{
|
||||
/// <summary>
|
||||
/// Gets or sets the schema version.
|
||||
/// </summary>
|
||||
[Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.PeerExchangeResponse;
|
||||
/// <summary>
|
||||
/// Gets or sets the request id.
|
||||
/// </summary>
|
||||
[Key("messageId")] public string MessageId { get; set; } = string.Empty;
|
||||
/// <summary>
|
||||
/// Gets or sets peer cards.
|
||||
/// </summary>
|
||||
[Key("peers")] public CultMeshPeerCardMessage[] Peers { get; set; } = Array.Empty<CultMeshPeerCardMessage>();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Wire representation of a CultMesh peer candidate.
|
||||
/// </summary>
|
||||
[MessagePackObject]
|
||||
public class CultMeshPeerCardMessage
|
||||
{
|
||||
/// <summary>
|
||||
/// Gets or sets the stable peer id.
|
||||
/// </summary>
|
||||
[Key("peerId")] public string PeerId { get; set; } = string.Empty;
|
||||
/// <summary>
|
||||
/// Gets or sets the Verse id this peer participates in.
|
||||
/// </summary>
|
||||
[Key("verseId")] public string VerseId { get; set; } = string.Empty;
|
||||
/// <summary>
|
||||
/// Gets or sets reachable endpoints.
|
||||
/// </summary>
|
||||
[Key("endpoints")] public string[] Endpoints { get; set; } = Array.Empty<string>();
|
||||
/// <summary>
|
||||
/// Gets or sets advertised peer roles.
|
||||
/// </summary>
|
||||
[Key("roles")] public string[] Roles { get; set; } = Array.Empty<string>();
|
||||
/// <summary>
|
||||
/// Gets or sets shard ids this peer can serve or observe.
|
||||
/// </summary>
|
||||
[Key("shardIds")] public string[] ShardIds { get; set; } = Array.Empty<string>();
|
||||
/// <summary>
|
||||
/// Gets or sets an optional region/locality label.
|
||||
/// </summary>
|
||||
[Key("region")] public string? Region { get; set; }
|
||||
/// <summary>
|
||||
/// Gets or sets the authority lease id, when the card advertises authority.
|
||||
/// </summary>
|
||||
[Key("authorityLeaseId")] public string? AuthorityLeaseId { get; set; }
|
||||
/// <summary>
|
||||
/// Gets or sets the expiry timestamp for this card.
|
||||
/// </summary>
|
||||
[Key("expiresAt")] public string? ExpiresAt { get; set; }
|
||||
/// <summary>
|
||||
/// Gets or sets an optional signature over the card.
|
||||
/// </summary>
|
||||
[Key("signature")] public string? Signature { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Describes one schema exposed through the CultNet schema catalog.
|
||||
/// </summary>
|
||||
|
|
|
|||
|
|
@ -365,6 +365,39 @@ namespace GameCult.Networking.Tests
|
|||
Assert.That(roundTrip.Verses[0].Compatibility.RequiredPluginIds, Is.EqualTo(["core"]));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void CultNetSchemaMessageSerialization_RoundTrips_PeerExchangeResponse()
|
||||
{
|
||||
var message = new CultMeshPeerExchangeResponseMessage
|
||||
{
|
||||
MessageId = "pex-1",
|
||||
Peers =
|
||||
[
|
||||
new CultMeshPeerCardMessage
|
||||
{
|
||||
PeerId = "peer-a",
|
||||
VerseId = "aetheria-main",
|
||||
Endpoints = ["cultnet://peer-a.example.test:3075"],
|
||||
Roles = [CultMeshPeerRoles.Discovery, CultMeshPeerRoles.ReadReplica],
|
||||
ShardIds = ["players"],
|
||||
Region = "eu-west",
|
||||
AuthorityLeaseId = "lease-1",
|
||||
ExpiresAt = "2026-05-20T12:00:00.0000000Z",
|
||||
Signature = "sig"
|
||||
}
|
||||
]
|
||||
};
|
||||
|
||||
var payload = CultNetSchemaMessageSerialization.Serialize(message);
|
||||
var roundTrip = (CultMeshPeerExchangeResponseMessage)CultNetSchemaMessageSerialization.Deserialize(payload);
|
||||
|
||||
Assert.That(roundTrip.MessageId, Is.EqualTo("pex-1"));
|
||||
Assert.That(roundTrip.Peers, Has.Length.EqualTo(1));
|
||||
Assert.That(roundTrip.Peers[0].PeerId, Is.EqualTo("peer-a"));
|
||||
Assert.That(roundTrip.Peers[0].Roles, Does.Contain(CultMeshPeerRoles.ReadReplica));
|
||||
Assert.That(roundTrip.Peers[0].AuthorityLeaseId, Is.EqualTo("lease-1"));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void CultNetSchemaMessageSerialization_RoundTrips_SimulationObservation()
|
||||
{
|
||||
|
|
@ -786,6 +819,18 @@ namespace GameCult.Networking.Tests
|
|||
Assert.That(client.Connected, Is.False);
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void CultMesh_CreateVerseDiscoveryClient_Returns_DiscoveryClient()
|
||||
{
|
||||
var client = CultMesh.CreateVerseDiscoveryClient(new CultMeshVerseDiscoveryClientOptions
|
||||
{
|
||||
ConnectTimeout = TimeSpan.FromMilliseconds(250),
|
||||
ResponseTimeout = TimeSpan.FromMilliseconds(250)
|
||||
});
|
||||
|
||||
Assert.That(client, Is.Not.Null);
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void CultMeshVerseCatalog_Finds_CompatibleTransferTargets()
|
||||
{
|
||||
|
|
@ -897,6 +942,72 @@ namespace GameCult.Networking.Tests
|
|||
Assert.That(updates, Has.Count.EqualTo(1));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void CultMeshPeerExchangeServer_Creates_FilteredPeerResponse()
|
||||
{
|
||||
using var catalog = CultMesh.CreatePeerCatalog();
|
||||
catalog.Upsert(new CultMeshPeerCard(
|
||||
"peer-primary",
|
||||
"aetheria-main",
|
||||
["cultnet://primary.example.test:3075"],
|
||||
roles: [CultMeshPeerRoles.ShardPrimary],
|
||||
shardIds: ["players"],
|
||||
region: "us-east",
|
||||
authorityLeaseId: "lease-primary"));
|
||||
catalog.Upsert(new CultMeshPeerCard(
|
||||
"peer-read",
|
||||
"aetheria-main",
|
||||
["cultnet://read.example.test:3075"],
|
||||
roles: [CultMeshPeerRoles.ReadReplica],
|
||||
shardIds: ["players"],
|
||||
region: "eu-west"));
|
||||
using var server = new Server(new CultCache(), DevelopmentServerSecurity);
|
||||
using var exchange = new CultMeshPeerExchangeServer(server, catalog);
|
||||
|
||||
var response = exchange.CreateResponse(new CultMeshPeerExchangeRequestMessage
|
||||
{
|
||||
MessageId = "pex-filter",
|
||||
VerseId = "aetheria-main",
|
||||
Roles = [CultMeshPeerRoles.ReadReplica],
|
||||
KnownPeerIds = ["already-known"]
|
||||
});
|
||||
|
||||
Assert.That(response.MessageId, Is.EqualTo("pex-filter"));
|
||||
Assert.That(response.Peers, Has.Length.EqualTo(1));
|
||||
Assert.That(response.Peers[0].PeerId, Is.EqualTo("peer-read"));
|
||||
Assert.That(response.Peers[0].Roles, Does.Contain(CultMeshPeerRoles.ReadReplica));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void CultMeshPeerCatalog_Upserts_WirePeerExchangeResponse()
|
||||
{
|
||||
using var catalog = CultMesh.CreatePeerCatalog();
|
||||
var updates = new List<CultMeshPeerCard>();
|
||||
using var subscription = catalog.Watch().Subscribe(updates.Add);
|
||||
|
||||
catalog.Upsert(new CultMeshPeerExchangeResponseMessage
|
||||
{
|
||||
MessageId = "pex-apply",
|
||||
Peers =
|
||||
[
|
||||
new CultMeshPeerCardMessage
|
||||
{
|
||||
PeerId = "peer-observer",
|
||||
VerseId = "aetheria-main",
|
||||
Endpoints = ["cultnet://observer.example.test:3075"],
|
||||
Roles = [CultMeshPeerRoles.SimulationObserver],
|
||||
ShardIds = ["arena"]
|
||||
}
|
||||
]
|
||||
});
|
||||
|
||||
var peers = catalog.Find("aetheria-main", CultMeshPeerRoles.SimulationObserver);
|
||||
|
||||
Assert.That(peers, Has.Count.EqualTo(1));
|
||||
Assert.That(peers[0].PeerId, Is.EqualTo("peer-observer"));
|
||||
Assert.That(updates, Has.Count.EqualTo(1));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task CultMeshNode_CanEnable_DurableShardLogs_WithDefaultPath()
|
||||
{
|
||||
|
|
|
|||
Loading…
Reference in a new issue