diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index 16178387..68a7e0b8 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -196,6 +196,14 @@ namespace GameCult.Mesh return new CultMeshVerseCatalog(); } + /// + /// Creates a local reactive catalog for discovered peers. + /// + public static CultMeshPeerCatalog CreatePeerCatalog() + { + return new CultMeshPeerCatalog(); + } + /// /// Attaches Verse discovery responses to a CultMesh node. /// @@ -207,6 +215,26 @@ namespace GameCult.Mesh return new CultMeshVerseDiscoveryServer(node.Server, catalog); } + /// + /// Creates a client for fetching Verse catalogs from discovery endpoints. + /// + public static CultMeshVerseDiscoveryClient CreateVerseDiscoveryClient( + CultMeshVerseDiscoveryClientOptions? options = null) + { + return new CultMeshVerseDiscoveryClient(options); + } + + /// + /// Attaches peer exchange responses to a CultMesh node. + /// + public static CultMeshPeerExchangeServer ServePeerExchange( + CultMeshNode node, + CultMeshPeerCatalog catalog) + { + if (node == null) throw new ArgumentNullException(nameof(node)); + return new CultMeshPeerExchangeServer(node.Server, catalog); + } + /// /// Creates a CultNet client configured for CultMesh use. /// diff --git a/src/GameCult.Mesh/CultMeshPeerExchange.cs b/src/GameCult.Mesh/CultMeshPeerExchange.cs new file mode 100644 index 00000000..ac4a36ce --- /dev/null +++ b/src/GameCult.Mesh/CultMeshPeerExchange.cs @@ -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 +{ + /// + /// Well-known CultMesh peer roles. + /// + public static class CultMeshPeerRoles + { + /// Peer can answer discovery requests. + public const string Discovery = "discovery"; + /// Peer can serve authoritative shard logs. + public const string ShardPrimary = "shard-primary"; + /// Peer can serve replicated shard logs. + public const string ShardReplica = "shard-replica"; + /// Peer can serve read traffic. + public const string ReadReplica = "read-replica"; + /// Peer can observe simulation facts. + public const string SimulationObserver = "simulation-observer"; + } + + /// + /// Candidate peer contact information for a Verse. + /// + public sealed class CultMeshPeerCard + { + /// + /// Creates a peer card. + /// + public CultMeshPeerCard( + string peerId, + string verseId, + IEnumerable endpoints, + IEnumerable? roles = null, + IEnumerable? 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; + } + + /// Gets the stable peer id. + public string PeerId { get; } + /// Gets the Verse id this peer participates in. + public string VerseId { get; } + /// Gets reachable endpoints. + public IReadOnlyList Endpoints { get; } + /// Gets advertised roles. + public IReadOnlyList Roles { get; } + /// Gets shard ids this peer can serve or observe. + public IReadOnlyList ShardIds { get; } + /// Gets an optional region/locality label. + public string? Region { get; } + /// Gets the authority lease id, when present. + public string? AuthorityLeaseId { get; } + /// Gets the expiry timestamp for this card. + public string? ExpiresAt { get; } + /// Gets an optional signature over the card. + public string? Signature { get; } + + /// + /// Returns whether the card advertises a role. + /// + public bool HasRole(string role) + { + return Roles.Contains(role, StringComparer.Ordinal); + } + + private static string[] Clean(IEnumerable? values) + { + return values?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray() + ?? Array.Empty(); + } + + private static string RequireNonEmpty(string value, string paramName) + { + return string.IsNullOrWhiteSpace(value) + ? throw new ArgumentException("Value must be non-empty.", paramName) + : value; + } + } + + /// + /// Conversion helpers for CultMesh peer exchange messages. + /// + public static class CultMeshPeerMessages + { + /// + /// Converts a peer card to its schema-v0 wire shape. + /// + 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 + }; + } + + /// + /// Converts a wire peer card to the local public API shape. + /// + 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); + } + } + + /// + /// Reactive local catalog for discovered CultMesh peers. + /// + public sealed class CultMeshPeerCatalog : IDisposable + { + private readonly Dictionary _peers = new(StringComparer.Ordinal); + private readonly Subject _updates = new(); + private bool _disposed; + + /// Gets all known peers. + public IReadOnlyList Peers => _peers.Values.OrderBy(peer => peer.PeerId, StringComparer.Ordinal).ToArray(); + + /// Watches peer card updates. + public Observable Watch() + { + ThrowIfDisposed(); + return _updates; + } + + /// Adds or replaces a peer card. + public void Upsert(CultMeshPeerCard card) + { + ThrowIfDisposed(); + if (card == null) throw new ArgumentNullException(nameof(card)); + _peers[card.PeerId] = card; + _updates.OnNext(card); + } + + /// Adds or replaces peer cards from a wire response. + public void Upsert(CultMeshPeerExchangeResponseMessage response) + { + ThrowIfDisposed(); + if (response == null) throw new ArgumentNullException(nameof(response)); + foreach (var peer in response.Peers) + { + Upsert(peer.ToPeerCard()); + } + } + + /// Finds peers by Verse and optional role. + public IReadOnlyList 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(); + } + + /// + public void Dispose() + { + if (_disposed) + { + return; + } + + _disposed = true; + _updates.Dispose(); + _peers.Clear(); + } + + private void ThrowIfDisposed() + { + if (_disposed) + { + throw new ObjectDisposedException(nameof(CultMeshPeerCatalog)); + } + } + } + + /// + /// Answers CultMesh peer exchange requests from a local peer catalog. + /// + public sealed class CultMeshPeerExchangeServer : IDisposable + { + private readonly Server _server; + private readonly CultMeshPeerCatalog _catalog; + private readonly Func _requestHandler; + private bool _disposed; + + /// + /// Creates and attaches a peer exchange bridge. + /// + 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); + } + + /// + /// Creates a peer exchange response for a request. + /// + 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(request.KnownPeerIds, StringComparer.Ordinal); + var roles = request.Roles == null || request.Roles.Length == 0 + ? null + : new HashSet(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 + }; + } + + /// + public void Dispose() + { + if (_disposed) + { + return; + } + + _disposed = true; + _server.RemoveCultNetMessageListener(_requestHandler); + } + + private Task HandleRequestAsync(CultMeshPeerExchangeRequestMessage request, NetPeer peer) + { + peer.SendCultNet(CreateResponse(request)); + return Task.CompletedTask; + } + } +} diff --git a/src/GameCult.Mesh/CultMeshVerseDiscovery.cs b/src/GameCult.Mesh/CultMeshVerseDiscovery.cs index bbe2086f..b441eeb8 100644 --- a/src/GameCult.Mesh/CultMeshVerseDiscovery.cs +++ b/src/GameCult.Mesh/CultMeshVerseDiscovery.cs @@ -131,4 +131,139 @@ namespace GameCult.Mesh return Task.CompletedTask; } } + + /// + /// Options for schema-v0 CultMesh Verse discovery clients. + /// + public sealed class CultMeshVerseDiscoveryClientOptions + { + /// + /// Gets or sets client security options used to connect to discovery 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 discovery response before failing. + /// + public TimeSpan ResponseTimeout { get; set; } = TimeSpan.FromSeconds(5); + + /// + /// Gets or sets a callback used to customize each ephemeral discovery client. + /// + public Action? ConfigureClient { get; set; } + } + + /// + /// Fetches CultMesh Verse catalogs from discovery endpoints. + /// + public sealed class CultMeshVerseDiscoveryClient + { + private readonly CultMeshVerseDiscoveryClientOptions _options; + + /// + /// Creates a Verse discovery client. + /// + public CultMeshVerseDiscoveryClient(CultMeshVerseDiscoveryClientOptions? options = null) + { + _options = options ?? new CultMeshVerseDiscoveryClientOptions(); + } + + /// + /// Fetches a Verse catalog response from one discovery endpoint. + /// + public async Task 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( + 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 CultMeshVerseCatalogRequestMessage + { + MessageId = messageId, + TransportVersion = request?.TransportVersion, + VerseIds = request?.VerseIds + }); + + return await WaitForResponseAsync(completion.Task, endpoint).ConfigureAwait(false); + } + + /// + /// Fetches Verse catalogs from endpoints and upserts every response into a local catalog. + /// + public async Task DiscoverAsync( + CultMeshVerseCatalog catalog, + IEnumerable 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 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 Verse discovery response from {endpoint}."); + } + + return await responseTask.ConfigureAwait(false); + } + } } diff --git a/src/GameCult.Mesh/README.md b/src/GameCult.Mesh/README.md index d5e1c8bc..0f641e87 100644 --- a/src/GameCult.Mesh/README.md +++ b/src/GameCult.Mesh/README.md @@ -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 diff --git a/src/GameCult.Mesh/docs/public-api.md b/src/GameCult.Mesh/docs/public-api.md index 6adb15b5..ea3ba74f 100644 --- a/src/GameCult.Mesh/docs/public-api.md +++ b/src/GameCult.Mesh/docs/public-api.md @@ -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 diff --git a/src/GameCult.Mesh/docs/verses.md b/src/GameCult.Mesh/docs/verses.md index e5d94ed7..b0588d5f 100644 --- a/src/GameCult.Mesh/docs/verses.md +++ b/src/GameCult.Mesh/docs/verses.md @@ -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. diff --git a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md index 1b24782f..8bcbf0e7 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -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 diff --git a/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs b/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs index 5b15313a..5e9431b4 100644 --- a/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs +++ b/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs @@ -70,6 +70,8 @@ namespace GameCult.Networking CultNetSchemaVersions.SimulationConsensusCandidate => Deserialize(payload), CultNetSchemaVersions.VerseCatalogRequest => Deserialize(payload), CultNetSchemaVersions.VerseCatalogResponse => Deserialize(payload), + CultNetSchemaVersions.PeerExchangeRequest => Deserialize(payload), + CultNetSchemaVersions.PeerExchangeResponse => Deserialize(payload), CultNetSchemaVersions.SchemaCatalogRequest => Deserialize(payload), CultNetSchemaVersions.SchemaCatalogResponse => Deserialize(payload), _ => throw new MessagePackSerializationException( diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs index 4d8726f3..fc21f5ee 100644 --- a/src/GameCult.Networking/CultNetSchemaMessages.cs +++ b/src/GameCult.Networking/CultNetSchemaMessages.cs @@ -123,6 +123,14 @@ namespace GameCult.Networking /// CultMesh Verse catalog response contract identifier. /// public const string VerseCatalogResponse = "cultmesh.verse_catalog_response.v0"; + /// + /// CultMesh peer exchange request contract identifier. + /// + public const string PeerExchangeRequest = "cultmesh.peer_exchange_request.v0"; + /// + /// CultMesh peer exchange response contract identifier. + /// + public const string PeerExchangeResponse = "cultmesh.peer_exchange_response.v0"; } /// @@ -1077,6 +1085,102 @@ namespace GameCult.Networking [Key("optionalPluginIds")] public string[] OptionalPluginIds { get; set; } = Array.Empty(); } + /// + /// Requests known peers for a CultMesh Verse. + /// + [MessagePackObject] + public class CultMeshPeerExchangeRequestMessage : ICultNetSchemaMessage + { + /// + /// Gets or sets the schema version. + /// + [Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.PeerExchangeRequest; + /// + /// Gets or sets the request id. + /// + [Key("messageId")] public string MessageId { get; set; } = string.Empty; + /// + /// Gets or sets the Verse id to discover peers for. + /// + [Key("verseId")] public string VerseId { get; set; } = string.Empty; + /// + /// Gets or sets optional role filters. + /// + [Key("roles")] public string[]? Roles { get; set; } + /// + /// Gets or sets peer ids the requester already knows. + /// + [Key("knownPeerIds")] public string[]? KnownPeerIds { get; set; } + /// + /// Gets or sets the maximum number of peer cards to return. + /// + [Key("limit")] public int? Limit { get; set; } + } + + /// + /// Returns known peers for a CultMesh Verse. + /// + [MessagePackObject] + public class CultMeshPeerExchangeResponseMessage : ICultNetSchemaMessage + { + /// + /// Gets or sets the schema version. + /// + [Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.PeerExchangeResponse; + /// + /// Gets or sets the request id. + /// + [Key("messageId")] public string MessageId { get; set; } = string.Empty; + /// + /// Gets or sets peer cards. + /// + [Key("peers")] public CultMeshPeerCardMessage[] Peers { get; set; } = Array.Empty(); + } + + /// + /// Wire representation of a CultMesh peer candidate. + /// + [MessagePackObject] + public class CultMeshPeerCardMessage + { + /// + /// Gets or sets the stable peer id. + /// + [Key("peerId")] public string PeerId { get; set; } = string.Empty; + /// + /// Gets or sets the Verse id this peer participates in. + /// + [Key("verseId")] public string VerseId { get; set; } = string.Empty; + /// + /// Gets or sets reachable endpoints. + /// + [Key("endpoints")] public string[] Endpoints { get; set; } = Array.Empty(); + /// + /// Gets or sets advertised peer roles. + /// + [Key("roles")] public string[] Roles { get; set; } = Array.Empty(); + /// + /// Gets or sets shard ids this peer can serve or observe. + /// + [Key("shardIds")] public string[] ShardIds { get; set; } = Array.Empty(); + /// + /// Gets or sets an optional region/locality label. + /// + [Key("region")] public string? Region { get; set; } + /// + /// Gets or sets the authority lease id, when the card advertises authority. + /// + [Key("authorityLeaseId")] public string? AuthorityLeaseId { get; set; } + /// + /// Gets or sets the expiry timestamp for this card. + /// + [Key("expiresAt")] public string? ExpiresAt { get; set; } + /// + /// Gets or sets an optional signature over the card. + /// + [Key("signature")] public string? Signature { get; set; } + } + /// /// Describes one schema exposed through the CultNet schema catalog. /// diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 6852fcfe..245d45b9 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -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(); + 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() {