From 8b48aa512f2dc0918ccd3d4572439f7dcdf022f2 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 27 Jun 2026 17:45:18 +0100 Subject: [PATCH] Add typed snapshot endpoint surface --- src/GameCult.Mesh/CultMesh.cs | 117 ++++++----- src/GameCult.Mesh/CultMeshSnapshots.cs | 190 +++++++++++++++++- .../CultMeshStreamingTests.cs | 57 +++++- 3 files changed, 300 insertions(+), 64 deletions(-) diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index 6d392918..ecc79c49 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -133,11 +133,35 @@ namespace GameCult.Mesh /// Gets or sets the message id prefix used for snapshot requests. public string? MessageIdPrefix { get; set; } + /// Gets or sets schema ids to request. Empty means the document type schema is requested. + public IReadOnlyList? SchemaIds { get; set; } + + /// Gets or sets the target shard id, when the endpoint is shard-aware. + public string? ShardId { get; set; } + + /// Gets or sets the target shard epoch, when the endpoint is shard-aware. + public long? ShardEpoch { 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; } + + /// Gets or sets the runtime id used by endpoint-created RUDP clients. + public string? RudpRuntimeId { get; set; } + + /// Gets or sets the connection id used by endpoint-created RUDP clients. + public uint RudpConnectionId { get; set; } = 0x43554c54; + + /// Gets or sets the connect payload used by endpoint-created RUDP clients. + public string RudpConnectPayload { get; set; } = "cultnet-schema-rudp"; + + /// Gets or sets the maximum fragment size used by endpoint-created RUDP clients. + public int RudpMaxFragmentBytes { get; set; } = 1024; + + /// Gets or sets the resend delay used by endpoint-created RUDP clients. + public long RudpResendDelayMs { get; set; } = 25; } /// @@ -2086,73 +2110,44 @@ namespace GameCult.Mesh 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, - RecordKeys = new[] { recordKey } - }); - - return await WaitForPeerSnapshotResponseAsync(completion.Task, endpoint, options.ResponseTimeout) + return await FetchSnapshotAsync( + createClient, + endpoint, + ToSnapshotRequestOptions( + options, + context, + options.SchemaIds ?? new[] { descriptor.SchemaId }, + new[] { recordKey })) .ConfigureAwait(false); } - private static string CreateSnapshotMessageId( + private static CultMeshSnapshotRequestOptions ToSnapshotRequestOptions( CultMeshPeerSnapshotDocumentOptions options, - CultMeshQueryContext context) + CultMeshQueryContext context, + IReadOnlyList? schemaIds, + IReadOnlyList? recordKeys) { - 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) + return new CultMeshSnapshotRequestOptions { - 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); + SchemaIds = schemaIds, + RecordKeys = recordKeys, + ShardId = options.ShardId, + ShardEpoch = options.ShardEpoch, + ResponseTimeout = options.ResponseTimeout, + ConnectTimeout = options.ConnectTimeout, + MessageIdPrefix = string.IsNullOrWhiteSpace(options.MessageIdPrefix) + ? $"cultmesh:{context.RuntimeId}:snapshot" + : options.MessageIdPrefix, + Security = options.Security, + ConfigureClient = options.ConfigureClient, + RudpRuntimeId = string.IsNullOrWhiteSpace(options.RudpRuntimeId) + ? context.RuntimeId + : options.RudpRuntimeId, + RudpConnectionId = options.RudpConnectionId, + RudpConnectPayload = options.RudpConnectPayload, + RudpMaxFragmentBytes = options.RudpMaxFragmentBytes, + RudpResendDelayMs = options.RudpResendDelayMs + }; } private static TDocument ReadDocumentFromSnapshotResponse( diff --git a/src/GameCult.Mesh/CultMeshSnapshots.cs b/src/GameCult.Mesh/CultMeshSnapshots.cs index 986b901f..bd595799 100644 --- a/src/GameCult.Mesh/CultMeshSnapshots.cs +++ b/src/GameCult.Mesh/CultMeshSnapshots.cs @@ -59,8 +59,186 @@ namespace GameCult.Mesh public long RudpResendDelayMs { get; set; } = 25; } + /// + /// Options for a typed remote snapshot surface bound to one CultNet endpoint. + /// + public sealed class CultMeshSnapshotEndpointOptions + { + /// Gets or sets the Verse context used by handles from this endpoint. + public CultMeshVerseContext? Context { get; set; } + + /// Gets or sets the document registry used for raw snapshot payload decoding. + public CultNetDocumentRegistry? DocumentRegistry { get; set; } + + /// Gets or sets request options applied to each snapshot fetch. + public CultMeshSnapshotRequestOptions? Request { get; set; } + + /// Gets or sets the source id advertised by diagnostics. Defaults to the endpoint. + public string? SourceId { get; set; } + + /// Gets or sets the route hint for resulting handles. + public CultMeshRouteHint? RouteHint { get; set; } + + /// Gets or sets the polling interval for watch fallback. + public TimeSpan PollInterval { get; set; } = TimeSpan.FromMilliseconds(250); + } + + /// + /// Typed snapshot surface for one remote CultNet endpoint. + /// + public sealed class CultMeshSnapshotEndpoint + { + internal CultMeshSnapshotEndpoint(string endpoint, CultMeshSnapshotEndpointOptions? options) + { + Endpoint = string.IsNullOrWhiteSpace(endpoint) + ? throw new ArgumentException("Value must be non-empty.", nameof(endpoint)) + : endpoint; + + var resolvedOptions = options ?? new CultMeshSnapshotEndpointOptions(); + Context = resolvedOptions.Context ?? CultMesh.Verse("remote", "cultmesh-snapshot-client").Context; + DocumentRegistry = resolvedOptions.DocumentRegistry ?? new CultNetDocumentRegistry(); + Request = CloneSnapshotRequestOptions(resolvedOptions.Request); + RouteHint = resolvedOptions.RouteHint ?? new CultMeshRouteHint(CultMeshLocalityKind.Network, Endpoint); + SourceId = string.IsNullOrWhiteSpace(resolvedOptions.SourceId) ? Endpoint : resolvedOptions.SourceId!; + PollInterval = resolvedOptions.PollInterval; + } + + /// Gets the endpoint this typed surface reads from. + public string Endpoint { get; } + + /// Gets the Verse context used by document handles from this endpoint. + public CultMeshVerseContext Context { get; } + + /// Gets the document registry used for raw snapshot payload decoding. + public CultNetDocumentRegistry DocumentRegistry { get; } + + /// Gets request options applied to each snapshot fetch. + public CultMeshSnapshotRequestOptions Request { get; } + + /// Gets the source id advertised by diagnostics. + public string SourceId { get; } + + /// Gets the route hint for resulting handles. + public CultMeshRouteHint RouteHint { get; } + + /// Gets the polling interval for watch fallback. + public TimeSpan PollInterval { get; } + + /// Fetches one raw snapshot with this endpoint's configured request policy. + public Task FetchSnapshotAsync( + IReadOnlyList? schemaIds = null, + IReadOnlyList? recordKeys = null) + { + return CultMesh.FetchSnapshotAsync(Endpoint, CreateRequest(schemaIds, recordKeys)); + } + + /// Fetches and decodes documents assignable to the requested type or matching its schema. + public Task> FetchDocumentsAsync( + IReadOnlyList? recordKeys = null, + IReadOnlyList? schemaIds = null) + where TDocument : class + { + var descriptor = CultDocumentRegistry.Shared.GetRequired(); + return CultMesh.FetchSnapshotDocumentsAsync( + Endpoint, + CreateRequest(schemaIds ?? new[] { descriptor.SchemaId }, recordKeys), + DocumentRegistry); + } + + /// Fetches one typed document by record key. + public async Task FetchDocumentAsync(string recordKey) + where TDocument : class + { + if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Value must be non-empty.", nameof(recordKey)); + var documents = await FetchDocumentsAsync(new[] { recordKey }).ConfigureAwait(false); + return documents.FirstOrDefault() + ?? throw new InvalidOperationException( + $"CultNet snapshot endpoint '{Endpoint}' did not return {typeof(TDocument).FullName} record '{recordKey}'."); + } + + /// Creates a typed document handle over one remote endpoint record. + public CultMeshDocumentHandle Document( + string recordKey, + string? documentId = null) + where TDocument : class + { + if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Value must be non-empty.", nameof(recordKey)); + var descriptor = CultDocumentRegistry.Shared.GetRequired(); + var sources = new[] + { + CultMesh.ProjectionSource($"{SourceId}:{recordKey}", descriptor.SchemaId, "CultNet snapshot endpoint") + }; + var watch = CultMesh.PollingQueryWatcher( + async (_parameters, _context) => await FetchDocumentAsync(recordKey).ConfigureAwait(false), + new CultMeshPollingWatchOptions(PollInterval)); + + return CultMesh.Document( + string.IsNullOrWhiteSpace(documentId) ? recordKey : documentId!, + Context, + _ => FetchDocumentAsync(recordKey), + queryContext => watch(CultMeshDocumentQueryParameters.Empty, queryContext), + sources, + RouteHint); + } + + /// Creates a schema-aware catalog over endpoint-backed document handles. + public CultMeshDocumentCatalog Documents(params ICultMeshDocumentHandle[] documents) + { + return CultMesh.Documents(documents); + } + + private CultMeshSnapshotRequestOptions CreateRequest( + IReadOnlyList? schemaIds, + IReadOnlyList? recordKeys) + { + var request = CloneSnapshotRequestOptions(Request); + request.SchemaIds = schemaIds ?? request.SchemaIds; + request.RecordKeys = recordKeys ?? request.RecordKeys; + if (string.IsNullOrWhiteSpace(request.RudpRuntimeId)) + request.RudpRuntimeId = Context.RuntimeId; + if (string.IsNullOrWhiteSpace(request.MessageIdPrefix)) + request.MessageIdPrefix = $"cultmesh:{Context.RuntimeId}:snapshot"; + return request; + } + + private static CultMeshSnapshotRequestOptions CloneSnapshotRequestOptions(CultMeshSnapshotRequestOptions? source) + { + if (source == null) + return new CultMeshSnapshotRequestOptions(); + + return new CultMeshSnapshotRequestOptions + { + SchemaIds = source.SchemaIds, + RecordKeys = source.RecordKeys, + ShardId = source.ShardId, + ShardEpoch = source.ShardEpoch, + ResponseTimeout = source.ResponseTimeout, + ConnectTimeout = source.ConnectTimeout, + MessageIdPrefix = source.MessageIdPrefix, + Security = source.Security, + ConfigureClient = source.ConfigureClient, + CreateClient = source.CreateClient, + RudpRuntimeId = source.RudpRuntimeId, + RudpConnectionId = source.RudpConnectionId, + RudpConnectPayload = source.RudpConnectPayload, + RudpMaxFragmentBytes = source.RudpMaxFragmentBytes, + RudpResendDelayMs = source.RudpResendDelayMs + }; + } + } + public static partial class CultMesh { + /// + /// Binds one CultNet endpoint as a typed snapshot surface. + /// + public static CultMeshSnapshotEndpoint SnapshotEndpoint( + string endpoint, + CultMeshSnapshotEndpointOptions? options = null) + { + return new CultMeshSnapshotEndpoint(endpoint, options); + } + /// /// Fetches one scoped raw CultNet snapshot from an endpoint. /// @@ -232,6 +410,7 @@ namespace GameCult.Mesh if (snapshot == null) throw new ArgumentNullException(nameof(snapshot)); if (registry == null) throw new ArgumentNullException(nameof(registry)); + var descriptor = CultDocumentRegistry.Shared.GetRequired(); var documents = new List(snapshot.Documents.Length); foreach (var record in snapshot.Documents) { @@ -239,7 +418,12 @@ namespace GameCult.Mesh continue; var binding = registry.GetBySchemaId(record.SchemaId); - if (binding == null || !typeof(TDocument).IsAssignableFrom(binding.DocumentType)) + var canDeserializeWithBinding = + binding != null && + typeof(TDocument).IsAssignableFrom(binding.DocumentType); + var canDeserializeAsSchemaAlias = + string.Equals(record.SchemaId, descriptor.SchemaId, StringComparison.Ordinal); + if (!canDeserializeWithBinding && !canDeserializeAsSchemaAlias) continue; if (!string.Equals(record.PayloadEncoding, "messagepack", StringComparison.Ordinal)) @@ -248,7 +432,9 @@ namespace GameCult.Mesh $"CultNet raw document payloadEncoding must be \"messagepack\", not \"{record.PayloadEncoding}\"."); } - documents.Add((TDocument)binding.PayloadDeserializer(record.Payload)); + documents.Add(canDeserializeWithBinding + ? (TDocument)binding!.PayloadDeserializer(record.Payload) + : CultDocumentMessagePackSerialization.Deserialize(record.Payload)); } return documents; diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index c1a9240d..c93f5ee7 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -1004,7 +1004,8 @@ public sealed class CultMeshStreamingTests handle.RouteHint.Kind.Should().Be(CultMeshLocalityKind.Network); handle.Sources.Should().ContainSingle().Which.SourceId.Should().Be(key.Value); requests.Should().ContainSingle(); - requests[0].SchemaIds.Should().BeNull(); + requests[0].SchemaIds.Should().ContainSingle().Which + .Should().Be(CultDocumentRegistry.Shared.GetRequired().SchemaId); requests[0].RecordKeys.Should().ContainSingle().Which.Should().Be(key.Value); (await catalog.LatestAsync()).Text.Should().Be("remote-snapshot"); @@ -1100,6 +1101,60 @@ public sealed class CultMeshStreamingTests request.MessageId.StartsWith("mesh-test-scoped-snapshot:", StringComparison.Ordinal)); } + [Test] + public async Task SnapshotEndpoint_ProvidesTypedHandlesAndSchemaAliases() + { + var sourceCache = new CultCache(); + var registry = new CultNetDocumentRegistry(CultDocumentRegistry.Shared); + registry.Register(CultNetDocumentBinding.ForDocument(sourceCache.Registry)); + var key = new CultRecordKey("mesh-note:snapshot-endpoint"); + await sourceCache.UpsertAsync(new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "snapshot-endpoint", + Revision = 19 + }, new CultRecordHandle(key)); + var requests = new List(); + var endpoint = "cultnet://snapshot-endpoint.test:3076"; + var surface = CultMesh.SnapshotEndpoint( + endpoint, + new CultMeshSnapshotEndpointOptions + { + Context = CultMesh.Verse("starbridge", "browser-starfire").Context, + DocumentRegistry = registry, + Request = new CultMeshSnapshotRequestOptions + { + ShardId = "primary", + ShardEpoch = 9, + CreateClient = () => new MeshSnapshotSchemaClient(request => + { + requests.Add(request); + return registry.CreateRawSnapshotResponse(sourceCache, request.MessageId, request); + }) + } + }); + + var fetchedAlias = await surface.FetchDocumentAsync(key.Value); + var aliasHandle = surface.Document(key.Value); + var aliasLatest = await aliasHandle.LatestAsync(); + var catalog = surface.Documents(surface.Document(key.Value)); + var catalogAlias = await catalog.LatestAsync(); + + fetchedAlias.Text.Should().Be("snapshot-endpoint"); + aliasLatest.Revision.Should().Be(19); + catalogAlias.Text.Should().Be("snapshot-endpoint"); + requests.Should().HaveCount(3); + requests.Should().OnlyContain(request => + request.SchemaIds!.SequenceEqual(new[] + { + CultDocumentRegistry.Shared.GetRequired().SchemaId + }) && + request.RecordKeys!.SequenceEqual(new[] { key.Value }) && + request.ShardId == "primary" && + request.ShardEpoch == 9 && + request.MessageId.StartsWith("cultmesh:browser-starfire:snapshot:", StringComparison.Ordinal)); + } + [Test] public async Task CollectionHandle_ReadsAndWatchesDatabaseCollectionsByIndex() {