GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1

Add C# peer snapshot document handles

Browse source
This commit is contained in:
Meta Vens 2026-06-27 16:37:44 +01:00
commit a57fa1113a
4 changed files with 412 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

281
src/GameCult.Mesh/CultMesh.cs
View file

@ -107,6 +107,39 @@ namespace GameCult.Mesh
}
}
/// <summary>
/// Options for opening one typed document from a remote CultNet snapshot endpoint.
/// </summary>
public sealed class CultMeshPeerSnapshotDocumentOptions
{
/// <summary>Gets or sets the semantic document id. Defaults to the record key.</summary>
public string? DocumentId { get; set; }
/// <summary>Gets or sets the source id advertised by diagnostics. Defaults to the record key.</summary>
public string? SourceId { get; set; }
/// <summary>Gets or sets the route hint for the resulting handle.</summary>
public CultMeshRouteHint? RouteHint { get; set; }
/// <summary>Gets or sets the response timeout for each snapshot request.</summary>
public TimeSpan ResponseTimeout { get; set; } = TimeSpan.FromSeconds(5);
/// <summary>Gets or sets the connection timeout for endpoint-created clients.</summary>
public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(5);
/// <summary>Gets or sets the polling interval for watch fallback.</summary>
public TimeSpan PollInterval { get; set; } = TimeSpan.FromMilliseconds(250);
/// <summary>Gets or sets the message id prefix used for snapshot requests.</summary>
public string? MessageIdPrefix { get; set; }
/// <summary>Gets or sets client security options for endpoint-created clients.</summary>
public ClientSecurityOptions? Security { get; set; }
/// <summary>Gets or sets a callback used to configure endpoint-created LiteNetLib clients.</summary>
public Action<Client>? ConfigureClient { get; set; }
}
/// <summary>
/// A locally hosted CultMesh node over CultCache, CultNet transport, and the mesh database facade.
/// </summary>
@ -1165,6 +1198,144 @@ namespace GameCult.Mesh
return Document<TDocument>(database, key, verse.Context, documentId, sources, routeHint);
}
/// <summary>
/// Creates a typed document handle over one remote CultNet snapshot response provider.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromPeerSnapshot<TDocument>(
Func<CultMeshQueryContext, Task<CultNetSnapshotResponseRawMessage>> 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<TDocument>();
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<CultMeshQueryContext, Task<TDocument>> latest = async queryContext =>
ReadDocumentFromSnapshotResponse<TDocument>(
await snapshot(queryContext).ConfigureAwait(false),
descriptor.SchemaId,
recordKey);
var watch = PollingQueryWatcher<CultMeshDocumentQueryParameters, TDocument>(
async (_parameters, queryContext) => await latest(queryContext).ConfigureAwait(false),
new CultMeshPollingWatchOptions<TDocument>(resolvedOptions.PollInterval));
return Document<TDocument>(
documentId,
context,
latest,
queryContext => watch(CultMeshDocumentQueryParameters.Empty, queryContext),
sources,
route);
}
/// <summary>
/// Creates a typed document handle over one remote CultNet snapshot response provider.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromPeerSnapshot<TDocument>(
Func<CultMeshQueryContext, Task<CultNetSnapshotResponseRawMessage>> snapshot,
string recordKey,
CultMeshVerse verse,
CultMeshPeerSnapshotDocumentOptions? options = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return DocumentFromPeerSnapshot<TDocument>(snapshot, recordKey, verse.Context, options);
}
/// <summary>
/// Creates a typed document handle over one remote CultNet schema endpoint.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromPeerSnapshot<TDocument>(
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<TDocument>(
queryContext => RequestPeerSnapshotAsync<TDocument>(
() => CultNetSchemaClients.CreateForEndpoint(endpoint, resolvedOptions.Security, resolvedOptions.ConfigureClient),
endpoint,
recordKey,
queryContext,
resolvedOptions),
recordKey,
context,
resolvedOptions);
}
/// <summary>
/// Creates a typed document handle over one remote CultNet schema endpoint.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromPeerSnapshot<TDocument>(
string endpoint,
string recordKey,
CultMeshVerse verse,
CultMeshPeerSnapshotDocumentOptions? options = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return DocumentFromPeerSnapshot<TDocument>(endpoint, recordKey, verse.Context, options);
}
/// <summary>
/// Creates a typed document handle over one remote CultNet schema client factory.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromPeerSnapshot<TDocument>(
Func<ICultNetSchemaClient> 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<TDocument>(
queryContext => RequestPeerSnapshotAsync<TDocument>(
createClient,
endpoint,
recordKey,
queryContext,
resolvedOptions),
recordKey,
context,
resolvedOptions);
}
/// <summary>
/// Creates a typed document handle over one remote CultNet schema client factory.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromPeerSnapshot<TDocument>(
Func<ICultNetSchemaClient> createClient,
string endpoint,
string recordKey,
CultMeshVerse verse,
CultMeshPeerSnapshotDocumentOptions? options = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return DocumentFromPeerSnapshot<TDocument>(createClient, endpoint, recordKey, verse.Context, options);
}
/// <summary>
/// Creates a typed document handle directly over one CultMesh node database record.
/// </summary>
@ -1639,6 +1810,116 @@ namespace GameCult.Mesh
});
}
private static async Task<CultNetSnapshotResponseRawMessage> RequestPeerSnapshotAsync<TDocument>(
Func<ICultNetSchemaClient> createClient,
string endpoint,
string recordKey,
CultMeshQueryContext context,
CultMeshPeerSnapshotDocumentOptions options)
where TDocument : class
{
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
var messageId = CreateSnapshotMessageId(options, context);
var completion = new TaskCompletionSource<CultNetSnapshotResponseRawMessage>(
TaskCreationOptions.RunContinuationsAsynchronously);
using var client = createClient();
client.OnCultNet<CultNetSnapshotResponseRawMessage>(response =>
{
if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal))
{
completion.TrySetResult(response);
}
});
client.OnCultNet<CultNetErrorMessage>(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<CultNetSnapshotResponseRawMessage> WaitForPeerSnapshotResponseAsync(
Task<CultNetSnapshotResponseRawMessage> 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<TDocument>(
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);
}
/// <summary>
/// Parses a rudp://host:port endpoint into its host/port parts.
/// </summary>

11
src/GameCult.Mesh/README.md
View file

@ -221,6 +221,17 @@ player.Value!.Health = 90;
await player.CommitAsync();
```
Remote publications use the same document handle shape:
```csharp
var health = CultMesh.DocumentFromPeerSnapshot<DaemonHealthDocument>(
"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

26
src/GameCult.Mesh/docs/public-api.md
View file

@ -273,6 +273,21 @@ using var subscription = cockpit.Watch(next => RenderCockpit(next));
var daemonAlias = cockpit.AsSchemaAlias<DaemonCockpitState>();
```
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<DaemonHealthDocument>(
"cultnet://daemon.local:3075",
"daemon:aetheria.health.v1",
verse);
var health = await remoteHealth.LatestAsync();
var uiHealth = remoteHealth.AsSchemaAlias<DaemonHealthUiDocument>();
```
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:

99
tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
View file

@ -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<MeshNoteDocument>(key));
var registry = new CultNetDocumentRegistry(CultDocumentRegistry.Shared)
.Register(CultNetDocumentBinding.ForDocument<MeshNoteDocument>(CultDocumentRegistry.Shared));
var requests = new List<CultNetSnapshotRequestMessage>();
var handle = CultMesh.DocumentFromPeerSnapshot<MeshNoteDocument>(
() => 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<MeshNoteAliasDocument>();
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<MeshNoteAliasDocument>()).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<CultNetSnapshotRequestMessage, CultNetSnapshotResponseRawMessage> _respond;
private readonly List<Action<CultNetSnapshotResponseRawMessage>> _snapshotHandlers = new();
private readonly List<Action<CultNetErrorMessage>> _errorHandlers = new();
public MeshSnapshotSchemaClient(
Func<CultNetSnapshotRequestMessage, CultNetSnapshotResponseRawMessage> respond)
{
_respond = respond;
}
public bool Connected { get; private set; }
public void Connect(string host, int port)
{
Connected = true;
}
public void SendCultNet<T>(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<T>(Action<T> 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()
{
}
}