From a57fa1113a92e776fe8daa99dd167302c3bfa77c Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 27 Jun 2026 16:37:44 +0100 Subject: [PATCH] Add C# peer snapshot document handles --- src/GameCult.Mesh/CultMesh.cs | 281 ++++++++++++++++++ src/GameCult.Mesh/README.md | 11 + src/GameCult.Mesh/docs/public-api.md | 26 +- .../CultMeshStreamingTests.cs | 99 ++++++ 4 files changed, 412 insertions(+), 5 deletions(-) diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index 08ecfe36..0c015818 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -107,6 +107,39 @@ namespace GameCult.Mesh } } + /// + /// Options for opening one typed document from a remote CultNet snapshot endpoint. + /// + public sealed class CultMeshPeerSnapshotDocumentOptions + { + /// Gets or sets the semantic document id. Defaults to the record key. + public string? DocumentId { get; set; } + + /// Gets or sets the source id advertised by diagnostics. Defaults to the record key. + public string? SourceId { get; set; } + + /// Gets or sets the route hint for the resulting handle. + public CultMeshRouteHint? RouteHint { get; set; } + + /// Gets or sets the response timeout for each snapshot request. + public TimeSpan ResponseTimeout { get; set; } = TimeSpan.FromSeconds(5); + + /// Gets or sets the connection timeout for endpoint-created clients. + public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(5); + + /// Gets or sets the polling interval for watch fallback. + public TimeSpan PollInterval { get; set; } = TimeSpan.FromMilliseconds(250); + + /// Gets or sets the message id prefix used for snapshot requests. + public string? MessageIdPrefix { get; set; } + + /// Gets or sets client security options for endpoint-created clients. + public ClientSecurityOptions? Security { get; set; } + + /// Gets or sets a callback used to configure endpoint-created LiteNetLib clients. + public Action? ConfigureClient { get; set; } + } + /// /// A locally hosted CultMesh node over CultCache, CultNet transport, and the mesh database facade. /// @@ -1165,6 +1198,144 @@ namespace GameCult.Mesh return Document(database, key, verse.Context, documentId, sources, routeHint); } + /// + /// Creates a typed document handle over one remote CultNet snapshot response provider. + /// + public static CultMeshDocumentHandle DocumentFromPeerSnapshot( + Func> snapshot, + string recordKey, + CultMeshVerseContext context, + CultMeshPeerSnapshotDocumentOptions? options = null) + where TDocument : class + { + if (snapshot == null) throw new ArgumentNullException(nameof(snapshot)); + if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Value must be non-empty.", nameof(recordKey)); + if (context == null) throw new ArgumentNullException(nameof(context)); + + var descriptor = CultDocumentRegistry.Shared.GetRequired(); + var resolvedOptions = options ?? new CultMeshPeerSnapshotDocumentOptions(); + var documentId = string.IsNullOrWhiteSpace(resolvedOptions.DocumentId) + ? recordKey + : resolvedOptions.DocumentId!; + var route = resolvedOptions.RouteHint ?? new CultMeshRouteHint(CultMeshLocalityKind.Network, "CultNet snapshot"); + var sources = new[] + { + ProjectionSource( + string.IsNullOrWhiteSpace(resolvedOptions.SourceId) ? recordKey : resolvedOptions.SourceId!, + descriptor.SchemaId, + "CultNet snapshot") + }; + Func> latest = async queryContext => + ReadDocumentFromSnapshotResponse( + await snapshot(queryContext).ConfigureAwait(false), + descriptor.SchemaId, + recordKey); + var watch = PollingQueryWatcher( + async (_parameters, queryContext) => await latest(queryContext).ConfigureAwait(false), + new CultMeshPollingWatchOptions(resolvedOptions.PollInterval)); + + return Document( + documentId, + context, + latest, + queryContext => watch(CultMeshDocumentQueryParameters.Empty, queryContext), + sources, + route); + } + + /// + /// Creates a typed document handle over one remote CultNet snapshot response provider. + /// + public static CultMeshDocumentHandle DocumentFromPeerSnapshot( + Func> snapshot, + string recordKey, + CultMeshVerse verse, + CultMeshPeerSnapshotDocumentOptions? options = null) + where TDocument : class + { + if (verse == null) throw new ArgumentNullException(nameof(verse)); + return DocumentFromPeerSnapshot(snapshot, recordKey, verse.Context, options); + } + + /// + /// Creates a typed document handle over one remote CultNet schema endpoint. + /// + public static CultMeshDocumentHandle DocumentFromPeerSnapshot( + string endpoint, + string recordKey, + CultMeshVerseContext context, + CultMeshPeerSnapshotDocumentOptions? options = null) + where TDocument : class + { + if (string.IsNullOrWhiteSpace(endpoint)) throw new ArgumentException("Value must be non-empty.", nameof(endpoint)); + var resolvedOptions = options ?? new CultMeshPeerSnapshotDocumentOptions(); + return DocumentFromPeerSnapshot( + queryContext => RequestPeerSnapshotAsync( + () => CultNetSchemaClients.CreateForEndpoint(endpoint, resolvedOptions.Security, resolvedOptions.ConfigureClient), + endpoint, + recordKey, + queryContext, + resolvedOptions), + recordKey, + context, + resolvedOptions); + } + + /// + /// Creates a typed document handle over one remote CultNet schema endpoint. + /// + public static CultMeshDocumentHandle DocumentFromPeerSnapshot( + string endpoint, + string recordKey, + CultMeshVerse verse, + CultMeshPeerSnapshotDocumentOptions? options = null) + where TDocument : class + { + if (verse == null) throw new ArgumentNullException(nameof(verse)); + return DocumentFromPeerSnapshot(endpoint, recordKey, verse.Context, options); + } + + /// + /// Creates a typed document handle over one remote CultNet schema client factory. + /// + public static CultMeshDocumentHandle DocumentFromPeerSnapshot( + Func createClient, + string endpoint, + string recordKey, + CultMeshVerseContext context, + CultMeshPeerSnapshotDocumentOptions? options = null) + where TDocument : class + { + if (createClient == null) throw new ArgumentNullException(nameof(createClient)); + if (string.IsNullOrWhiteSpace(endpoint)) throw new ArgumentException("Value must be non-empty.", nameof(endpoint)); + var resolvedOptions = options ?? new CultMeshPeerSnapshotDocumentOptions(); + return DocumentFromPeerSnapshot( + queryContext => RequestPeerSnapshotAsync( + createClient, + endpoint, + recordKey, + queryContext, + resolvedOptions), + recordKey, + context, + resolvedOptions); + } + + /// + /// Creates a typed document handle over one remote CultNet schema client factory. + /// + public static CultMeshDocumentHandle DocumentFromPeerSnapshot( + Func createClient, + string endpoint, + string recordKey, + CultMeshVerse verse, + CultMeshPeerSnapshotDocumentOptions? options = null) + where TDocument : class + { + if (verse == null) throw new ArgumentNullException(nameof(verse)); + return DocumentFromPeerSnapshot(createClient, endpoint, recordKey, verse.Context, options); + } + /// /// Creates a typed document handle directly over one CultMesh node database record. /// @@ -1639,6 +1810,116 @@ namespace GameCult.Mesh }); } + private static async Task RequestPeerSnapshotAsync( + Func createClient, + string endpoint, + string recordKey, + CultMeshQueryContext context, + CultMeshPeerSnapshotDocumentOptions options) + where TDocument : class + { + var descriptor = CultDocumentRegistry.Shared.GetRequired(); + var messageId = CreateSnapshotMessageId(options, context); + var completion = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + using var client = createClient(); + client.OnCultNet(response => + { + if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) + { + completion.TrySetResult(response); + } + }); + client.OnCultNet(error => + completion.TrySetException(new InvalidOperationException(error.Error))); + + var (host, port) = CultNetSchemaWriteForwarder.ParseEndpoint(endpoint); + client.Connect(host, port); + await WaitForSchemaClientConnectionAsync(client, endpoint, options.ConnectTimeout).ConfigureAwait(false); + client.SendCultNet(new CultNetSnapshotRequestMessage + { + MessageId = messageId, + SchemaIds = new[] { descriptor.SchemaId }, + RecordKeys = new[] { recordKey } + }); + + return await WaitForPeerSnapshotResponseAsync(completion.Task, endpoint, options.ResponseTimeout) + .ConfigureAwait(false); + } + + private static string CreateSnapshotMessageId( + CultMeshPeerSnapshotDocumentOptions options, + CultMeshQueryContext context) + { + var prefix = string.IsNullOrWhiteSpace(options.MessageIdPrefix) + ? $"cultmesh:{context.RuntimeId}:snapshot" + : options.MessageIdPrefix!; + return $"{prefix}:{Guid.NewGuid():N}"; + } + + private static async Task WaitForSchemaClientConnectionAsync( + ICultNetSchemaClient client, + string endpoint, + TimeSpan timeout) + { + var deadline = DateTimeOffset.UtcNow + timeout; + while (!client.Connected) + { + if (DateTimeOffset.UtcNow >= deadline) + { + throw new TimeoutException($"Timed out connecting to CultNet snapshot endpoint {endpoint}."); + } + + await Task.Delay(TimeSpan.FromMilliseconds(25)).ConfigureAwait(false); + } + } + + private static async Task WaitForPeerSnapshotResponseAsync( + Task responseTask, + string endpoint, + TimeSpan timeout) + { + var timeoutTask = Task.Delay(timeout); + var completed = await Task.WhenAny(responseTask, timeoutTask).ConfigureAwait(false); + if (completed != responseTask) + { + throw new TimeoutException($"Timed out waiting for CultNet snapshot response from {endpoint}."); + } + + return await responseTask.ConfigureAwait(false); + } + + private static TDocument ReadDocumentFromSnapshotResponse( + CultNetSnapshotResponseRawMessage response, + string schemaId, + string recordKey) + where TDocument : class + { + if (response == null) throw new ArgumentNullException(nameof(response)); + var record = response.Documents.FirstOrDefault(candidate => + string.Equals(candidate.SchemaId, schemaId, StringComparison.Ordinal) && + string.Equals(candidate.RecordKey, recordKey, StringComparison.Ordinal)) + ?? response.Documents.FirstOrDefault(candidate => + string.Equals(candidate.SchemaId, schemaId, StringComparison.Ordinal)) + ?? response.Documents.FirstOrDefault(candidate => + string.Equals(candidate.RecordKey, recordKey, StringComparison.Ordinal)); + if (record == null) + { + throw new InvalidOperationException( + $"CultNet snapshot response '{response.MessageId}' did not contain schema '{schemaId}' record '{recordKey}'."); + } + + if (!string.Equals(record.PayloadEncoding, "messagepack", StringComparison.Ordinal)) + { + throw new InvalidOperationException( + $"CultNet raw document payloadEncoding must be \"messagepack\", not \"{record.PayloadEncoding}\"."); + } + + return (TDocument)CultDocumentMessagePackSerialization.DeserializeUntyped( + typeof(TDocument), + record.Payload); + } + /// /// Parses a rudp://host:port endpoint into its host/port parts. /// diff --git a/src/GameCult.Mesh/README.md b/src/GameCult.Mesh/README.md index a6699bde..77ebb014 100644 --- a/src/GameCult.Mesh/README.md +++ b/src/GameCult.Mesh/README.md @@ -221,6 +221,17 @@ player.Value!.Health = 90; await player.CommitAsync(); ``` +Remote publications use the same document handle shape: + +```csharp +var health = CultMesh.DocumentFromPeerSnapshot( + "cultnet://daemon.local:3075", + "daemon:aetheria.health.v1", + verse); + +var latest = await health.LatestAsync(); +``` + The physical cache table remains SoA for CPU-local scans: ```csharp diff --git a/src/GameCult.Mesh/docs/public-api.md b/src/GameCult.Mesh/docs/public-api.md index d56a5a88..7f5ef40f 100644 --- a/src/GameCult.Mesh/docs/public-api.md +++ b/src/GameCult.Mesh/docs/public-api.md @@ -273,6 +273,21 @@ using var subscription = cockpit.Watch(next => RenderCockpit(next)); var daemonAlias = cockpit.AsSchemaAlias(); ``` +Remote CultNet snapshots use the same document handle surface. The caller names +the endpoint, record key, and typed document; CultMesh owns the snapshot request +id, response filtering, raw MessagePack payload decode, route metadata, and +polling watch fallback: + +```csharp +var remoteHealth = CultMesh.DocumentFromPeerSnapshot( + "cultnet://daemon.local:3075", + "daemon:aetheria.health.v1", + verse); + +var health = await remoteHealth.LatestAsync(); +var uiHealth = remoteHealth.AsSchemaAlias(); +``` + A domain facade can collect handles into one schema-aware catalog: ```csharp @@ -346,11 +361,12 @@ const health = await remoteHealth.latest(); ``` Projected, remote, shared-memory, IPC, WASM, and local-cache documents all use -the same TS handle contract. The caller names the typed document or collection -it wants; the handle owns route defaults, snapshot reads, reactive watches, -schema alias validation, and replacement when the backing source is mutable. -Generated Aetheria and Ymir facades should bind those handles to a Verse once -and keep transport, quorum, and cache mechanics behind CultMesh. +the same C# and TS handle contract. The caller names the typed document or +collection it wants; the handle owns route defaults, snapshot reads, reactive +watches, schema alias validation, prediction submission, and replacement when +the backing source supports those mutations. Generated Aetheria and Ymir +facades should bind those handles to a Verse once and keep transport, quorum, +and cache mechanics behind CultMesh. Native view descriptors can name unmanaged columns without hand-entered byte sizes: diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index e4856572..fed12c54 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -848,6 +848,50 @@ public sealed class CultMeshStreamingTests (await handle.LatestAsync()).Text.Should().Be("predicted-thermal"); } + [Test] + public async Task DocumentHandle_ReadsRemotePeerSnapshotsAsTypedDocuments() + { + var cache = new CultCache(); + var key = new CultRecordKey("mesh-note:remote"); + await cache.UpsertAsync(new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "remote-snapshot", + Revision = 9 + }, new CultRecordHandle(key)); + var registry = new CultNetDocumentRegistry(CultDocumentRegistry.Shared) + .Register(CultNetDocumentBinding.ForDocument(CultDocumentRegistry.Shared)); + var requests = new List(); + + var handle = CultMesh.DocumentFromPeerSnapshot( + () => new MeshSnapshotSchemaClient(request => + { + requests.Add(request); + return registry.CreateRawSnapshotResponse(cache, request.MessageId, request); + }), + "cultnet://snapshot.test:3075", + key.Value, + CultMesh.Verse("starbridge", "unity-pilot"), + new CultMeshPeerSnapshotDocumentOptions + { + PollInterval = TimeSpan.FromMilliseconds(10), + MessageIdPrefix = "mesh-test-snapshot" + }); + var alias = handle.AsSchemaAlias(); + var catalog = CultMesh.Documents(handle); + + var snapshot = await alias.LatestAsync(); + + snapshot.Text.Should().Be("remote-snapshot"); + snapshot.Revision.Should().Be(9); + handle.RouteHint.Kind.Should().Be(CultMeshLocalityKind.Network); + handle.Sources.Should().ContainSingle().Which.SourceId.Should().Be(key.Value); + requests.Should().ContainSingle(); + requests[0].SchemaIds.Should().ContainSingle().Which.Should().Be(handle.SchemaId); + requests[0].RecordKeys.Should().ContainSingle().Which.Should().Be(key.Value); + (await catalog.LatestAsync()).Text.Should().Be("remote-snapshot"); + } + [Test] public async Task CollectionHandle_ReadsAndWatchesDatabaseCollectionsByIndex() { @@ -1538,3 +1582,58 @@ public sealed class MeshObjectsFacade return _visibleObjects.ExecuteAsync(request); } } + +internal sealed class MeshSnapshotSchemaClient : ICultNetSchemaClient +{ + private readonly Func _respond; + private readonly List> _snapshotHandlers = new(); + private readonly List> _errorHandlers = new(); + + public MeshSnapshotSchemaClient( + Func respond) + { + _respond = respond; + } + + public bool Connected { get; private set; } + + public void Connect(string host, int port) + { + Connected = true; + } + + public void SendCultNet(T message) where T : ICultNetSchemaMessage + { + if (message is not CultNetSnapshotRequestMessage request) + { + foreach (var handler in _errorHandlers) + { + handler(new CultNetErrorMessage { Error = $"Unexpected message {typeof(T).Name}." }); + } + + return; + } + + var response = _respond(request); + foreach (var handler in _snapshotHandlers) + { + handler(response); + } + } + + public void OnCultNet(Action callback) where T : ICultNetSchemaMessage + { + if (typeof(T) == typeof(CultNetSnapshotResponseRawMessage)) + { + _snapshotHandlers.Add(message => callback((T)(object)message)); + } + else if (typeof(T) == typeof(CultNetErrorMessage)) + { + _errorHandlers.Add(message => callback((T)(object)message)); + } + } + + public void Dispose() + { + } +}