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

Invoke typed operations through retained clients

Browse source
This commit is contained in:
Meta Vens 2026-08-18 03:49:33 +02:00
commit e49f93f56d
8 changed files with 268 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

24
samples/eve-browser-network/Program.cs
View file

@ -208,7 +208,29 @@ static async Task RunHeadlessAsync(Args arguments)
Console.WriteLine("HEADLESS_READY " + JsonSerializer.Serialize(new { count = initial.Count }));
}
if (initial.Count >= arguments.ExpectedCount) completion.TrySetResult();
await completion.Task.WaitAsync(TimeSpan.FromSeconds(45));
var commandTask = Task.Run(async () =>
{
var line = await Console.In.ReadLineAsync();
if (line == null || !line.StartsWith("INVOKE ", StringComparison.Ordinal))
throw new InvalidOperationException("The headless sample expected 'INVOKE <idempotency-key>' on stdin.");
var commandId = line["INVOKE ".Length..].Trim();
var result = await mesh.InvokeAsync<IncrementRequest, IncrementReceipt>(
arguments.VerseId,
"sample.counter",
"sample.counter.increment",
"sample.increment.v1",
"gamecult.eve.command_receipt.v1",
new IncrementRequest { Amount = 1 },
sourceRuntimeId: "sample.csharp",
idempotencyKey: commandId);
Console.WriteLine("HEADLESS_RECEIPT " + JsonSerializer.Serialize(new
{
receiptId = result.Value.ReceiptId,
count = result.Value.Count,
status = result.Status
}));
});
await Task.WhenAll(completion.Task, commandTask).WaitAsync(TimeSpan.FromSeconds(45));
}
static EveSurfaceDocument CreateCounterSurface()

17
samples/eve-browser-network/README.md
View file

@ -7,8 +7,9 @@ This is the real-process continuation of `samples/eve-two-runtime`:
- one real Chromium page leases both documents through `cultmesh-browser` and
lowers the provider-owned Eve surface;
- one retained C# headless client discovers the provider through the same Odin
identity, observes the typed counter and canonical receipt ids, and survives
the provider route replacement without application reconnect code;
identity, observes the typed counter and canonical receipt ids, survives the
provider route replacement, and invokes through the generic CultMesh client
without application transport or reconnect code;
- one separate local Odin fixture answers the canonical CultMesh Verse catalog
over binary CultNet WebSocket messages;
- the browser and C# clients speak the same binary CultNet schema-v0 WebSocket
@ -26,12 +27,12 @@ directly only to diagnose the network layer. Browser discovery is portable:
`CHROME_PATH` wins when configured, then common Chrome/Chromium/Edge locations
and Playwright's installed Chromium are checked.
The verifier clicks the Eve button twice with one idempotency key, requires one
state transition, restarts the provider on a different physical port, updates
the local Odin fixture, and requires the still-open browser to rediscover the
route, resubscribe, execute a second command, and preserve the durable count.
The retained C# lease independently rediscovers the same replacement route and
observes both canonical receipt ids.
The verifier clicks the Eve button twice with one idempotency key and requires
one state transition. It then restarts the provider on a different physical
port, updates the local Odin fixture, and requires both retained clients to
rediscover and resubscribe. The C# client invokes the second command through
`CultMeshClient.InvokeAsync`; Chromium observes the resulting count and both
clients observe both canonical receipt ids.
The fixture speaks the production Verse-catalog contract; a deployment smoke
against the full Odin daemon remains a separate infrastructure gate.

15
scripts/verify-eve-browser-network.mjs
View file

@ -99,10 +99,12 @@ try {
await page.waitForFunction(() =>
window.__sampleConnectionStates?.includes("reconnecting")
&& window.__sampleConnectionStates.at(-1) === "connected");
await page.evaluate(() => { window.__sampleCommandId = "browser-click-2"; });
await page.locator("button").click();
await page.waitForFunction(() => window.__sampleReceipt?.count === 2 && window.__sampleCount === 2);
const secondReceipt = await page.evaluate(() => window.__sampleReceipt?.receiptId);
headless.sendLine("INVOKE headless-command-2");
const headlessReceipt = JSON.parse(await headless.waitFor("HEADLESS_RECEIPT "));
assert.equal(headlessReceipt.status, "accepted");
assert.equal(headlessReceipt.count, 2);
await page.waitForFunction(() => window.__sampleCount === 2);
const secondReceipt = headlessReceipt.receiptId;
const secondHeadlessUpdate = JSON.parse(await headless.waitFor("HEADLESS_UPDATE_2 "));
assert.equal(secondHeadlessUpdate.count, 2);
assert.ok(secondHeadlessUpdate.receiptIds.includes(firstReceipt));
@ -151,7 +153,7 @@ async function startOdin(port, providerEndpoint) {
function startDotnet(arguments_) {
const process = spawn("dotnet", [
"run", "--no-build", "--project", join(sampleRoot, "EveBrowserNetworkSample.csproj"), "--", ...arguments_,
], { cwd: repoRoot, stdio: ["ignore", "pipe", "pipe"] });
], { cwd: repoRoot, stdio: ["pipe", "pipe", "pipe"] });
let output = "";
let error = "";
const waiters = [];
@ -177,6 +179,9 @@ function startDotnet(arguments_) {
}
return {
process,
sendLine(line) {
process.stdin.write(`${line}\n`);
},
waitFor(prefix, timeoutMs = 30_000) {
const existing = output.split(/\r?\n/u).find(value => value.startsWith(prefix));
if (existing) return Promise.resolve(existing.slice(prefix.length));

136
src/GameCult.Mesh/CultMeshClient.cs
View file

@ -6,6 +6,7 @@ using System.Threading;
using System.Threading.Tasks;
using GameCult.Caching;
using GameCult.Networking;
using MessagePack;
using R3;
namespace GameCult.Mesh
@ -40,6 +41,45 @@ namespace GameCult.Mesh
/// <summary>Gets or sets the deadline before an unanswered subscription intent is replayed.</summary>
public TimeSpan SubscriptionResponseTimeout { get; set; } = TimeSpan.FromSeconds(2);
/// <summary>Gets or sets the deadline for one correlated typed operation response.</summary>
public TimeSpan OperationResponseTimeout { get; set; } = TimeSpan.FromSeconds(10);
}
/// <summary>
/// Carries one provider-authored operation response. Transport delivery does not by itself
/// establish that the provider accepted or committed the requested gameplay change.
/// </summary>
public sealed class CultMeshOperationResult<TResponse>
{
internal CultMeshOperationResult(CultNetOperationResponseMessage envelope, TResponse value)
{
MessageId = envelope.MessageId;
ServiceId = envelope.ServiceId;
Operation = envelope.Operation;
Status = envelope.Status;
PayloadSchema = envelope.PayloadSchema;
SourceRuntimeId = envelope.SourceRuntimeId;
Diagnostics = envelope.Diagnostics ?? Array.Empty<string>();
Value = value;
}
/// <summary>Gets the durable caller-supplied idempotency key.</summary>
public string MessageId { get; }
/// <summary>Gets the provider service that answered.</summary>
public string ServiceId { get; }
/// <summary>Gets the operation that answered.</summary>
public string Operation { get; }
/// <summary>Gets the provider-authored operation status.</summary>
public string Status { get; }
/// <summary>Gets the portable response payload schema.</summary>
public string PayloadSchema { get; }
/// <summary>Gets the responding runtime identity, when supplied.</summary>
public string? SourceRuntimeId { get; }
/// <summary>Gets provider or transport diagnostics carried by the response.</summary>
public IReadOnlyList<string> Diagnostics { get; }
/// <summary>Gets the decoded typed response payload.</summary>
public TResponse Value { get; }
}
/// <summary>Owns one shared live document subscription until disposed.</summary>
@ -85,6 +125,7 @@ namespace GameCult.Mesh
private readonly CultMeshDiscoveryService _discovery;
private readonly CultMeshSessionManager _sessions;
private readonly TimeSpan _subscriptionResponseTimeout;
private readonly TimeSpan _operationResponseTimeout;
private readonly ConcurrentDictionary<string, SharedResource> _documents = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, SharedResource> _collections = new(StringComparer.Ordinal);
private bool _disposed;
@ -94,6 +135,8 @@ namespace GameCult.Mesh
if (options == null) throw new ArgumentNullException(nameof(options));
if (options.SubscriptionResponseTimeout <= TimeSpan.Zero)
throw new ArgumentOutOfRangeException(nameof(options.SubscriptionResponseTimeout));
if (options.OperationResponseTimeout <= TimeSpan.Zero)
throw new ArgumentOutOfRangeException(nameof(options.OperationResponseTimeout));
var endpoints = options.RendezvousEndpoints
.Where(endpoint => !string.IsNullOrWhiteSpace(endpoint))
.Distinct(StringComparer.Ordinal)
@ -123,6 +166,7 @@ namespace GameCult.Mesh
options.RealtimeConnectors,
options.Sessions);
_subscriptionResponseTimeout = options.SubscriptionResponseTimeout;
_operationResponseTimeout = options.OperationResponseTimeout;
}
/// <summary>Connects to a stable endpoint identity using one application protocol.</summary>
@ -296,6 +340,98 @@ namespace GameCult.Mesh
session.OpenSchemaClient().SendCultNet(message);
}
/// <summary>
/// Invokes one typed provider operation through the reusable identity-owned session.
/// The caller supplies a durable idempotency key; the returned status and payload remain
/// provider-authored evidence rather than client-side authority over canonical state.
/// </summary>
public async Task<CultMeshOperationResult<TResponse>> InvokeAsync<TRequest, TResponse>(
string endpointId,
string serviceId,
string operation,
string requestSchema,
string responseSchema,
TRequest request,
string sourceRuntimeId,
string idempotencyKey,
CancellationToken cancellationToken = default)
{
ThrowIfDisposed();
if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId));
if (string.IsNullOrWhiteSpace(serviceId)) throw new ArgumentException("Service identity is required.", nameof(serviceId));
if (string.IsNullOrWhiteSpace(operation)) throw new ArgumentException("Operation identity is required.", nameof(operation));
if (string.IsNullOrWhiteSpace(requestSchema)) throw new ArgumentException("Request schema is required.", nameof(requestSchema));
if (string.IsNullOrWhiteSpace(responseSchema)) throw new ArgumentException("Response schema is required.", nameof(responseSchema));
if (request is null) throw new ArgumentNullException(nameof(request));
if (string.IsNullOrWhiteSpace(sourceRuntimeId)) throw new ArgumentException("Source runtime identity is required.", nameof(sourceRuntimeId));
if (string.IsNullOrWhiteSpace(idempotencyKey)) throw new ArgumentException("A durable idempotency key is required.", nameof(idempotencyKey));
var identity = endpointId.Trim();
var expectedService = serviceId.Trim();
var expectedOperation = operation.Trim();
var expectedResponseSchema = responseSchema.Trim();
var messageId = idempotencyKey.Trim();
var session = await ConnectAsync(identity, CultMeshProtocols.Documents, cancellationToken).ConfigureAwait(false);
var completion = new TaskCompletionSource<CultNetOperationResponseMessage>(TaskCreationOptions.RunContinuationsAsynchronously);
using var subscription = session.OnCultNet<CultNetOperationResponseMessage>(response =>
{
if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal))
completion.TrySetResult(response);
});
using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
timeout.CancelAfter(_operationResponseTimeout);
using var cancellation = timeout.Token.Register(() => completion.TrySetCanceled(timeout.Token));
session.SendCultNet(new CultNetOperationRequestMessage
{
MessageId = messageId,
ServiceId = expectedService,
Operation = expectedOperation,
PayloadSchema = requestSchema.Trim(),
PayloadEncoding = "messagepack-base64",
Payload = Convert.ToBase64String(MessagePackSerializer.Serialize(request, CultNetSchemaMessageSerialization.Options)),
SourceRuntimeId = sourceRuntimeId.Trim()
});
CultNetOperationResponseMessage envelope;
try
{
envelope = await completion.Task.ConfigureAwait(false);
}
catch (TaskCanceledException) when (!cancellationToken.IsCancellationRequested)
{
throw new TimeoutException(
$"CultMesh operation '{expectedService}/{expectedOperation}' did not answer idempotency key '{messageId}' " +
$"within {_operationResponseTimeout}.");
}
cancellationToken.ThrowIfCancellationRequested();
if (!string.Equals(envelope.ServiceId, expectedService, StringComparison.Ordinal) ||
!string.Equals(envelope.Operation, expectedOperation, StringComparison.Ordinal))
throw new InvalidOperationException("CultMesh operation response did not match the requested service and operation.");
if (!string.Equals(envelope.PayloadSchema, expectedResponseSchema, StringComparison.Ordinal))
throw new InvalidOperationException(
$"CultMesh operation returned payload schema '{envelope.PayloadSchema}', expected '{expectedResponseSchema}'.");
if (!string.Equals(envelope.PayloadEncoding, "messagepack-base64", StringComparison.Ordinal))
throw new InvalidOperationException(
$"CultMesh operation returned unsupported payload encoding '{envelope.PayloadEncoding}'.");
if (string.IsNullOrWhiteSpace(envelope.Status))
throw new InvalidOperationException("CultMesh operation response omitted its provider-authored status.");
TResponse value;
try
{
value = MessagePackSerializer.Deserialize<TResponse>(
Convert.FromBase64String(envelope.Payload),
CultNetSchemaMessageSerialization.Options);
}
catch (FormatException error)
{
throw new InvalidOperationException("CultMesh operation response payload was not valid base64.", error);
}
if (value is null) throw new InvalidOperationException("CultMesh operation response decoded to null.");
return new CultMeshOperationResult<TResponse>(envelope, value);
}
/// <summary>Gets the number of currently leased shared document subscriptions.</summary>
public int ActiveDocumentResourceCount => _documents.Count;

17
src/GameCult.Mesh/docs/browser-verse-transport.md
View file

@ -178,10 +178,11 @@ Current executable coverage from `scripts/verify-eve-browser-network.mjs`:
- Eve button to typed operation to provider receipt to shared state;
- duplicate idempotency key produces one effect;
- provider restart on a different physical port rehydrates the `.cc` state;
- the retained browser lease rediscovers through Odin, resubscribes, and
commits a second provider-authored receipt;
- the retained C# lease independently rediscovers, resubscribes, and observes
both canonical receipt ids without application reconnect code;
- the retained browser lease rediscovers through Odin and resubscribes;
- the retained C# client independently rediscovers, resubscribes, invokes the
second command through `CultMeshClient.InvokeAsync`, and receives the
provider-authored receipt without application transport or reconnect code;
- Chromium observes the C# command's canonical state and receipt chronology;
- generated browser bundle contains no `node:` builtin import.
The full public Verse gate still requires the deployed Odin daemon rather than
@ -199,10 +200,12 @@ From an empty temporary consumer using packed/released artifacts only:
6. observe the same state version and receipt id in both clients;
7. disconnect or rotate the physical WebSocket endpoint, then require both
leases to resume without application reconnect code;
8. restart the provider and prove state plus receipt persistence;
9. reject malformed schemas, direct document writes, duplicate command
8. invoke from C# after route replacement and require Chromium to observe the
same provider-authored state and receipt;
9. restart the provider and prove state plus receipt persistence;
10. reject malformed schemas, direct document writes, duplicate command
effects, stale subscription generations, and unsupported authority;
10. inspect the browser bundle and fail if it contains Node builtin imports.
11. inspect the browser bundle and fail if it contains Node builtin imports.
The gate runs on Windows and Linux. A screenshot is useful presentation
evidence, but the receipt/state chronology is the authority proof.

4
src/GameCult.Mesh/docs/getting-started/03-publish-an-eve-surface.md
View file

@ -61,7 +61,9 @@ The second layer runs a durable C# provider, a separate local Odin fixture, a
real Chromium Eve lowerer, and an independent C# headless observer over binary
CultNet WebSocket lanes. The open browser resolves provider identity through
the canonical Verse catalog, survives a provider restart on a different route,
resubscribes, and commits another canonical receipt.
and resubscribes. The retained C# client then invokes a second typed operation
through `CultMeshClient.InvokeAsync`; Chromium and C# observe the same
provider-authored state and receipt chronology.
Current verification for the Unity consumer remains:

5
src/GameCult.Mesh/docs/getting-started/README.md
View file

@ -16,8 +16,9 @@ TypeScript packages into an empty temporary consumer, validates the Eve surface
and receipt, executes one idempotent counter command, proves a DOM lowerer and
headless observer converge, and reopens the durable `.cc` store. Then it boots a
durable C# provider, local Odin fixture, real Chromium lowerer, and independent
C# observer; it proves discovery by identity, commands, canonical receipts,
provider restart on another route, resubscription, and durable convergence.
C# client. Chromium invokes before provider replacement; C# invokes afterward.
Both retained clients prove discovery by identity, canonical receipts, provider
restart on another route, resubscription, and durable convergence.
Set `CHROME_PATH` only when no installed Chrome, Chromium, Edge, or Playwright
Chromium can be found automatically.

74
tests/GameCult.Mesh.Tests/CultMeshClientTests.cs
View file

@ -257,6 +257,37 @@ public sealed class CultMeshClientTests
submitted.Document.SourceRole.Should().Be("pilot");
}
[Test]
public async Task InvokeOperation_CorrelatesProviderReceiptThroughStableIdentitySession()
{
var connector = new DocumentConnector();
using var mesh = new CultMeshClient(new CultMeshClientOptions
{
RendezvousEndpoints = new[] { "rudp://odin:3076" },
Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
Connectors = new[] { connector }
});
var result = await mesh.InvokeAsync<ClientOperationRequest, ClientOperationReceipt>(
"aetheria",
"tests.counter",
"tests.counter.increment",
"tests.counter.increment_request.v1",
"tests.counter.increment_receipt.v1",
new ClientOperationRequest { Amount = 2 },
sourceRuntimeId: "pilot-one",
idempotencyKey: "pilot-command-1");
result.MessageId.Should().Be("pilot-command-1");
result.Status.Should().Be("accepted");
result.SourceRuntimeId.Should().Be("provider-daemon");
result.Value.ReceiptId.Should().Be("receipt:pilot-command-1");
result.Value.Count.Should().Be(2);
connector.Clients.Should().ContainSingle();
connector.Clients[0].Operations.Should().ContainSingle()
.Which.SourceRuntimeId.Should().Be("pilot-one");
}
[Test]
public async Task Dispose_TerminatesDocumentWaitingForInitialSnapshot()
{
@ -360,6 +391,7 @@ public sealed class CultMeshClientTests
private sealed class DocumentClient : ICultNetSchemaClient, ICultNetSchemaClientHealth
{
private readonly List<Action<CultNetSnapshotResponseRawMessage>> _snapshots = new();
private readonly List<Action<CultNetOperationResponseMessage>> _operationResponses = new();
private readonly CultNetDocumentRegistry _documents;
private readonly string _text;
private readonly TaskCompletionSource<Exception> _failure = new(TaskCreationOptions.RunContinuationsAsynchronously);
@ -376,6 +408,7 @@ public sealed class CultMeshClientTests
public int SubscribeCount { get; private set; }
public int HandlerRegistrationCount { get; private set; }
public List<CultNetDocumentPutRawMessage> Submitted { get; } = new();
public List<CultNetOperationRequestMessage> Operations { get; } = new();
public void Connect(string host, int port) { }
public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage
{
@ -384,6 +417,32 @@ public sealed class CultMeshClientTests
Submitted.Add(submitted);
return;
}
if (message is CultNetOperationRequestMessage operation)
{
Operations.Add(operation);
var decodedRequest = MessagePackSerializer.Deserialize<ClientOperationRequest>(
Convert.FromBase64String(operation.Payload),
CultNetSchemaMessageSerialization.Options);
var receipt = new ClientOperationReceipt
{
ReceiptId = "receipt:" + operation.MessageId,
Count = decodedRequest.Amount
};
var operationResponse = new CultNetOperationResponseMessage
{
MessageId = operation.MessageId,
ServiceId = operation.ServiceId,
Operation = operation.Operation,
Status = "accepted",
PayloadSchema = "tests.counter.increment_receipt.v1",
Payload = Convert.ToBase64String(MessagePackSerializer.Serialize(
receipt,
CultNetSchemaMessageSerialization.Options)),
SourceRuntimeId = "provider-daemon"
};
foreach (var handler in _operationResponses.ToArray()) handler(operationResponse);
return;
}
if (message is not CultNetDatabaseSubscribeMessage request) return;
SubscribeCount++;
if (_snapshotsToSuppress > 0)
@ -408,6 +467,8 @@ public sealed class CultMeshClientTests
HandlerRegistrationCount++;
if (typeof(T) == typeof(CultNetSnapshotResponseRawMessage))
_snapshots.Add(response => callback((T)(object)response));
if (typeof(T) == typeof(CultNetOperationResponseMessage))
_operationResponses.Add(response => callback((T)(object)response));
}
public void Dispose() { }
public void Fail(Exception error) => _failure.TrySetResult(error);
@ -440,4 +501,17 @@ public sealed class CultMeshClientTests
[Key(0), CultName] public string Id { get; set; } = string.Empty;
[Key(1)] public string Text { get; set; } = string.Empty;
}
[MessagePackObject]
public sealed class ClientOperationRequest
{
[Key("amount")] public int Amount { get; set; }
}
[MessagePackObject]
public sealed class ClientOperationReceipt
{
[Key("receiptId")] public string ReceiptId { get; set; } = string.Empty;
[Key("count")] public int Count { get; set; }
}
}