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

Add typed CultNet operation dispatch

Browse source
This commit is contained in:
Meta Vens 2026-08-18 05:59:11 +02:00
commit b0a876a220
13 changed files with 913 additions and 128 deletions
Download patch file Download diff file Expand all files Collapse all files

6
packages/cultmesh-browser/README.md
View file

@ -30,6 +30,12 @@ const counter = await mesh.leaseRawDocument({
counter.watch(record => renderCounter(decodeCultNetPayload(record)));
```
`invoke(...)` returns the correlated provider response. Routing, schema, and
envelope failures reject immediately with `CultMeshBrowserOperationError`;
callers can branch on its stable `status` and `code` without parsing prose or
waiting for the request timeout. Domain rejections remain typed provider
responses because the domain still owns their receipt schema.
`CultMeshBrowserOdinRendezvous` speaks the canonical CultNet Verse-catalog
messages to one or more configured Odin WebSocket endpoints. Application code
supplies stable identity; it does not own endpoint selection, reconnect, or

36
packages/cultmesh-browser/src/index.ts
View file

@ -176,6 +176,29 @@ export interface CultMeshOperationOptions {
targetRuntimeId?: string;
}
/** Portable schema returned when routing or envelope validation fails before a domain reply exists. */
export const cultNetOperationFailureSchema = "gamecult.cultnet.operation_failure.v1";
export interface CultNetOperationFailure {
code: string;
message: string;
}
/** Correlated framework-level failure returned by a CultMesh operation provider. */
export class CultMeshBrowserOperationError extends Error {
readonly status: string;
readonly code: string;
readonly diagnostics: readonly string[];
constructor(response: CultNetOperationResponseMessage, failure: CultNetOperationFailure) {
super(failure.message || "CultMesh operation failed.");
this.name = "CultMeshBrowserOperationError";
this.status = response.status || "error";
this.code = failure.code || "operation-error";
this.diagnostics = response.diagnostics ?? [];
}
}
export type CultMeshConnectionState = "connecting" | "connected" | "reconnecting" | "disposed";
type DocumentWatcher = (record: CultNetRawDocumentRecord | undefined) => void;
@ -554,6 +577,19 @@ export class CultMeshBrowserClient implements AsyncDisposable {
if (!pending) return;
this.#pendingOperations.delete(message.messageId);
clearTimeout(pending.timer);
if (message.payloadSchema === cultNetOperationFailureSchema) {
try {
pending.reject(new CultMeshBrowserOperationError(
message,
decodeCultNetOperationPayload<CultNetOperationFailure>(message),
));
} catch (error) {
pending.reject(error instanceof Error
? error
: new Error("CultMesh operation failure payload was malformed."));
}
return;
}
pending.resolve(message);
return;
}

52
packages/cultmesh-browser/test/cultmesh-browser.test.ts
View file

@ -13,6 +13,7 @@ import {
import {
CultMeshBrowserClient,
CultMeshBrowserOperationError,
CultMeshBrowserOdinRendezvous,
decodeCultNetOperationPayload,
decodeCultNetPayload,
@ -142,6 +143,35 @@ test("browser client bounds unanswered provider operations", async () => {
await client.dispose();
});
test("browser client rejects a correlated framework failure without waiting for timeout", async () => {
const provider = new FakeProvider("ws://127.0.0.1:4351/mesh", 1, true);
const client = await CultMeshBrowserClient.connect({
...provider.route,
runtimeId: "browser-failure-test",
rendezvous: { resolve: async () => provider.route },
requestTimeoutMs: 60_000,
socketFactory: () => provider.open(),
});
await assert.rejects(
client.invoke({
serviceId: "sample.counter",
operation: "increment",
payloadSchema: "wrong.request.v1",
payload: { amount: 1 },
idempotencyKey: "failure-1",
}),
(error: unknown) => {
assert.ok(error instanceof CultMeshBrowserOperationError);
assert.equal(error.status, "invalid");
assert.equal(error.code, "request-schema-mismatch");
assert.match(error.message, /expected payload schema/i);
return true;
},
);
await client.dispose();
});
test("browser client rejects oversized outbound schema messages", async () => {
const route = {
verseId: "sample.counter",
@ -189,8 +219,9 @@ class FakeProvider {
readonly route: CultMeshBrowserRoute;
#count: number;
#subscriptions = new Map<FakeSocket, Map<string, CultNetDatabaseSubscribeMessage>>();
#rejectOperations: boolean;
constructor(endpoint: string, count: number) {
constructor(endpoint: string, count: number, rejectOperations = false) {
this.route = {
verseId: "sample.counter",
providerId: "sample.counter-provider",
@ -198,6 +229,7 @@ class FakeProvider {
generation: endpoint,
};
this.#count = count;
this.#rejectOperations = rejectOperations;
}
get activeSubscriptionCount(): number {
@ -242,6 +274,24 @@ class FakeProvider {
}
private invoke(socket: FakeSocket, message: CultNetOperationRequestMessage): void {
if (this.#rejectOperations) {
socket.deliver({
schemaVersion: "cultnet.operation_response.v0",
messageId: message.messageId,
serviceId: message.serviceId,
operation: message.operation,
status: "invalid",
payloadSchema: "gamecult.cultnet.operation_failure.v1",
payloadEncoding: "messagepack-base64",
payload: bytesToBase64(encode({
code: "request-schema-mismatch",
message: "Expected payload schema 'sample.increment.v1'.",
})),
diagnostics: ["Request schema did not match."],
sourceRuntimeId: "sample.counter-provider",
});
return;
}
const payload = decode(base64ToBytes(message.payload)) as { amount: number };
this.#count += payload.amount;
for (const [peer, subscriptions] of this.#subscriptions) {

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

@ -95,61 +95,50 @@ static async Task RunProviderAsync(Args arguments)
await using var schemaServer = new CultNetWebSocketSchemaServer();
using var subscriptions = new CultNetDatabaseSubscriptionServer(schemaServer, database);
var operationGate = new SemaphoreSlim(1, 1);
schemaServer.OnCultNet<CultNetOperationRequestMessage>(async (request, peer) =>
{
if (request.ServiceId != "sample.counter")
{
peer.SendCultNet(new CultNetErrorMessage { Error = "Unsupported sample operation." });
return;
}
if (request.Operation == "sample.counter.ping")
{
var ping = MessagePackSerializer.Deserialize<PingRequest>(
Convert.FromBase64String(request.Payload),
CultNetSchemaMessageSerialization.Options);
SendOperationResponse(
peer,
request,
"sample.ping_receipt.v1",
new PingReceipt { Sequence = ping.Sequence });
return;
}
if (request.Operation != "sample.counter.increment")
{
peer.SendCultNet(new CultNetErrorMessage { Error = "Unsupported sample operation." });
return;
}
await operationGate.WaitAsync();
try
{
var current = cache.Get<CounterState>(new CultRecordKey(counterKey))
?? throw new InvalidOperationException("Counter state is unavailable.");
if (!current.Receipts.TryGetValue(request.MessageId, out var receipt))
using var operationGate = new SemaphoreSlim(1, 1);
using var operations = new CultNetOperationServer(schemaServer, "sample.counter-provider")
.Register<PingRequest, PingReceipt>(
"sample.counter",
"sample.counter.ping",
"sample.ping_request.v1",
"sample.ping_receipt.v1",
context => Task.FromResult(new PingReceipt { Sequence = context.Value.Sequence }))
.Register<IncrementRequest, IncrementReceipt>(
"sample.counter",
"sample.counter.increment",
"sample.increment.v1",
"gamecult.eve.command_receipt.v1",
async context =>
{
var input = MessagePackSerializer.Deserialize<IncrementRequest>(
Convert.FromBase64String(request.Payload),
CultNetSchemaMessageSerialization.Options);
receipt = IncrementReceipt.Accepted(request.MessageId, current.Count + input.Amount);
var receipts = new Dictionary<string, IncrementReceipt>(current.Receipts, StringComparer.Ordinal)
await operationGate.WaitAsync();
try
{
[request.MessageId] = receipt
};
await database.PutAsync(new CultRecordKey(counterKey), new CounterState
var current = cache.Get<CounterState>(new CultRecordKey(counterKey))
?? throw new InvalidOperationException("Counter state is unavailable.");
if (current.Receipts.TryGetValue(context.IdempotencyKey, out var existing))
return existing;
var receipt = IncrementReceipt.Accepted(
context.IdempotencyKey,
current.Count + context.Value.Amount);
var receipts = new Dictionary<string, IncrementReceipt>(current.Receipts, StringComparer.Ordinal)
{
[context.IdempotencyKey] = receipt
};
await database.PutAsync(new CultRecordKey(counterKey), new CounterState
{
CounterId = counterKey,
Count = receipt.Count,
Receipts = receipts
});
await cache.FlushAsync();
return receipt;
}
finally
{
CounterId = counterKey,
Count = receipt.Count,
Receipts = receipts
});
await cache.FlushAsync();
}
SendOperationResponse(peer, request, "gamecult.eve.command_receipt.v1", receipt);
}
finally
{
operationGate.Release();
}
});
operationGate.Release();
}
});
var builder = WebApplication.CreateSlimBuilder();
builder.WebHost.ConfigureKestrel(options => options.Listen(IPAddress.Loopback, arguments.Port));
@ -245,26 +234,6 @@ static async Task RunHeadlessAsync(Args arguments)
await Task.WhenAll(completion.Task, commandTask).WaitAsync(TimeSpan.FromSeconds(45));
}
static void SendOperationResponse<T>(
ICultNetSchemaServerPeer peer,
CultNetOperationRequestMessage request,
string payloadSchema,
T payload)
{
peer.SendCultNet(new CultNetOperationResponseMessage
{
MessageId = request.MessageId,
ServiceId = request.ServiceId,
Operation = request.Operation,
Status = "accepted",
PayloadSchema = payloadSchema,
Payload = Convert.ToBase64String(MessagePackSerializer.Serialize(
payload,
CultNetSchemaMessageSerialization.Options)),
SourceRuntimeId = "sample.counter-provider"
});
}
static async Task<object> MeasureOperationSessionAsync(CultMeshClient mesh, Args arguments)
{
const int warmupOperations = 100;
@ -337,54 +306,27 @@ static void ForceCollection()
static EveSurfaceDocument CreateCounterSurface()
{
var networkRoute = new CultMeshRouteHint(CultMeshLocalityKind.Network, "cultmesh");
return new EveSurfaceDocument(
providerId: "sample.counter-provider",
providerKind: "sample.daemon",
title: "CultMesh browser counter",
version: 1,
updatedAtUtc: "2026-08-17T00:00:00Z",
surface: new EveSurfaceTree(
surfaceKey,
new EveSurfaceComponent(
"counter.root",
"column",
new Dictionary<string, string>(StringComparer.Ordinal),
[
new EveSurfaceComponent(
"counter.value",
"metric",
new Dictionary<string, string>(StringComparer.Ordinal)
{
["label"] = "Canonical count",
["value"] = "0"
},
[],
[
new CultMeshStateBindingDescriptor(
"value",
"sample.counter.count",
"sample.counter_state:counter:main",
"sample.counter_state.v1",
networkRoute)
]),
new EveSurfaceComponent(
"counter.increment",
"control.button",
new Dictionary<string, string>(StringComparer.Ordinal)
{
["label"] = "Increment",
["command"] = "sample.counter.increment"
},
[])
]),
[]),
commands:
[
new EveCommandTemplate(new CultMeshOperationBindingDescriptor(
"sample.counter.increment",
"Increment",
routeHint: networkRoute))
]);
var count = new CultMeshStateBindingDescriptor(
"value",
"sample.counter.count",
"sample.counter_state:counter:main",
"sample.counter_state.v1",
networkRoute);
var increment = CultMesh.OperationBinding(
"sample.counter.increment",
"Increment",
"sample.increment.v1",
networkRoute);
return EveSurface.Create(surfaceKey)
.Provider("sample.counter-provider", "sample.daemon")
.Title("CultMesh browser counter")
.Version(1)
.UpdatedAtUtc("2026-08-17T00:00:00Z")
.RootColumn("counter.root", root => root
.Title("counter.title", "CultMesh browser counter")
.Metric("counter.value", "Canonical count", "0", count)
.Button("counter.increment", "Increment", increment))
.Build();
}
sealed class Args

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

@ -15,6 +15,12 @@ This is the real-process continuation of `samples/eve-two-runtime`:
- the browser and C# clients speak the same binary CultNet schema-v0 WebSocket
messages through the authenticated host adapter.
The provider registers typed commands through `CultNetOperationServer`.
Application code receives `CultNetOperationContext<TRequest>` and returns a
typed receipt; it does not parse base64, switch on wire envelopes, or construct
correlation responses. The counter's durable transaction still owns
idempotency and state mutation.
Run the proof from the CultLib repository:
```powershell

11
scripts/verify-eve-getting-started.ps1
View file

@ -6,6 +6,17 @@ param(
$ErrorActionPreference = "Stop"
$cultLibRoot = Split-Path -Parent $PSScriptRoot
$EveRoot = [IO.Path]::GetFullPath($EveRoot)
$networkSample = Join-Path $cultLibRoot "samples\eve-browser-network\Program.cs"
$networkSampleSource = Get-Content -LiteralPath $networkSample -Raw
if ($networkSampleSource -match 'OnCultNet\s*<\s*CultNetOperationRequestMessage' -or
$networkSampleSource -match 'Convert\.FromBase64String\s*\(\s*request\.Payload') {
throw "The getting-started provider has regressed to hand-written CultNet operation envelope dispatch."
}
foreach ($requiredPrimitive in @("CultNetOperationServer", "EveSurface.Create")) {
if ($networkSampleSource -notmatch [regex]::Escape($requiredPrimitive)) {
throw "The getting-started provider is missing the public primitive '$requiredPrimitive'."
}
}
& (Join-Path $PSScriptRoot "verify-eve-two-runtime-sample.ps1") `
-EveRoot $EveRoot `

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

@ -82,6 +82,29 @@ namespace GameCult.Mesh
public TResponse Value { get; }
}
/// <summary>Correlated framework-level failure returned before a typed domain reply exists.</summary>
public sealed class CultMeshRemoteOperationException : Exception
{
internal CultMeshRemoteOperationException(
CultNetOperationResponseMessage envelope,
CultNetOperationFailure failure)
: base(failure?.Message ?? "CultMesh operation failed.")
{
Status = envelope?.Status ?? "error";
Code = failure?.Code ?? "operation-error";
Diagnostics = envelope?.Diagnostics ?? Array.Empty<string>();
}
/// <summary>Gets the provider-authored failure status.</summary>
public string Status { get; }
/// <summary>Gets the stable machine-readable failure code.</summary>
public string Code { get; }
/// <summary>Gets provider diagnostics carried by the response.</summary>
public IReadOnlyList<string> Diagnostics { get; }
}
/// <summary>Owns one shared live document subscription until disposed.</summary>
public sealed class CultMeshDocumentLease<TDocument> : IDisposable where TDocument : class
{
@ -408,6 +431,24 @@ namespace GameCult.Mesh
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, CultNetOperationServer.FailureSchemaId, StringComparison.Ordinal))
{
if (!string.Equals(envelope.PayloadEncoding, "messagepack-base64", StringComparison.Ordinal))
throw new InvalidOperationException(
$"CultMesh operation failure returned unsupported payload encoding '{envelope.PayloadEncoding}'.");
CultNetOperationFailure failure;
try
{
failure = MessagePackSerializer.Deserialize<CultNetOperationFailure>(
Convert.FromBase64String(envelope.Payload),
CultNetSchemaMessageSerialization.Options);
}
catch (Exception error) when (error is FormatException || error is MessagePackSerializationException)
{
throw new InvalidOperationException("CultMesh operation failure payload was malformed.", error);
}
throw new CultMeshRemoteOperationException(envelope, failure);
}
if (!string.Equals(envelope.PayloadSchema, expectedResponseSchema, StringComparison.Ordinal))
throw new InvalidOperationException(
$"CultMesh operation returned payload schema '{envelope.PayloadSchema}', expected '{expectedResponseSchema}'.");

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

@ -34,13 +34,17 @@ wire, see [docs/typed-document-path.md](docs/typed-document-path.md).
## Eve / CultUI Surfaces
CultMesh includes the portable Eve/CultUI surface document contract and a fluent
builder for daemon-published UI. The builder is deliberately shaped like the
legacy CultUI/Unity properties panel API: compose menus and forms in readable
imperative code, while the emitted document carries typed CultMesh operation
bindings, state bindings, style tokens, and embedded surface slots.
CultMesh supplies the typed state and operation binding descriptors used by Eve.
The renderer-neutral `GameCult.Eve.Surface` package in the Eve repository owns
the portable surface document and fluent builder. Add that package when a daemon
publishes UI; do not add EveUnity to the provider. The emitted document carries
typed CultMesh operation bindings, state bindings, style tokens, and embedded
surface slots while remaining independent of any lowering runtime.
```csharp
using GameCult.Eve.Surface;
using GameCult.Mesh;
var menu = EveSurface.Create("aetheria.main_menu.root")
.Provider("aetheria", "game.menu")
.TitleSubtitle("AETHERIA", "TERMINUS")

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

@ -29,6 +29,58 @@ CultMesh document session. Other runtime lowerers must preserve the same
boundary: they may reject or report unsupported capabilities, but they may not
import provider internals.
## The Application-Owned Part
Install the renderer-neutral `GameCult.Eve.Surface` package in the provider.
Do not install EveUnity or the browser lowerer there. A provider defines typed
state, one typed operation transaction, and one Eve surface. CultNet envelope
encoding is framework work:
```csharp
using GameCult.Eve.Surface;
using GameCult.Mesh;
using GameCult.Networking;
using var operations = new CultNetOperationServer(schemaServer, "sample.counter-provider")
.Register<IncrementRequest, IncrementReceipt>(
"sample.counter",
"sample.counter.increment",
"sample.increment.v1",
"gamecult.eve.command_receipt.v1",
async command => await IncrementExactlyOnceAsync(
command.IdempotencyKey,
command.Value));
var route = new CultMeshRouteHint(CultMeshLocalityKind.Network, "cultmesh");
var count = new CultMeshStateBindingDescriptor(
"value",
"sample.counter.count",
"sample.counter_state:counter:main",
"sample.counter_state.v1",
route);
var increment = CultMesh.OperationBinding(
"sample.counter.increment",
"Increment",
"sample.increment.v1",
route);
var surface = EveSurface.Create("sample.counter")
.Provider("sample.counter-provider", "sample.daemon")
.Title("CultMesh browser counter")
.RootColumn("counter.root", root => root
.Metric("counter.value", "Canonical count", "0", count)
.Button("counter.increment", "Increment", increment))
.Build();
```
`schemaServer` is a transport-neutral CultNet server port. WebSocket, RUDP, and
test transports attach the same dispatcher. The handler owns validation,
idempotency, state mutation, and its durable receipt. `CultNetOperationServer`
owns only route/schema checks and typed envelope serialization. The complete
provider transaction is in
[`samples/eve-browser-network/Program.cs`](../../../../samples/eve-browser-network/Program.cs);
it contains no hand-written operation base64 or MessagePack dispatch switch.
## Multi-Runtime Checkpoint
Run the same provider and connect at least two lowerers. Both must report the

348
src/GameCult.Networking/CultNetOperationServer.cs Normal file
View file

@ -0,0 +1,348 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using MessagePack;
namespace GameCult.Networking
{
/// <summary>Portable failure payload for operation routing and envelope validation.</summary>
[MessagePackObject]
public sealed class CultNetOperationFailure
{
/// <summary>Gets or sets the stable machine-readable failure code.</summary>
[Key("code")] public string Code { get; set; } = "operation-error";
/// <summary>Gets or sets the human-readable failure description.</summary>
[Key("message")] public string Message { get; set; } = "CultNet operation failed.";
}
/// <summary>
/// Typed application input delivered through one CultNet operation envelope.
/// The message id is the caller's idempotency key; the application transaction
/// remains responsible for applying it exactly once.
/// </summary>
public sealed class CultNetOperationContext<TRequest>
{
internal CultNetOperationContext(CultNetOperationRequestMessage envelope, TRequest value)
{
Envelope = envelope ?? throw new ArgumentNullException(nameof(envelope));
Value = value;
}
/// <summary>Gets the decoded application request.</summary>
public TRequest Value { get; }
/// <summary>Gets the durable caller-supplied idempotency key.</summary>
public string IdempotencyKey => Envelope.MessageId;
/// <summary>Gets the calling runtime identity, when supplied.</summary>
public string? SourceRuntimeId => Envelope.SourceRuntimeId;
/// <summary>Gets the optional target runtime identity.</summary>
public string? TargetRuntimeId => Envelope.TargetRuntimeId;
/// <summary>Gets the original transport-neutral envelope for diagnostics.</summary>
public CultNetOperationRequestMessage Envelope { get; }
}
/// <summary>Provider-authored typed result for one CultNet operation.</summary>
public sealed class CultNetOperationReply<TResponse>
{
/// <summary>Creates one provider-authored typed reply.</summary>
public CultNetOperationReply(
string status,
TResponse value,
IEnumerable<string>? diagnostics = null)
{
Status = RequireNonEmpty(status, nameof(status));
Value = value;
Diagnostics = diagnostics == null ? Array.Empty<string>() : new List<string>(diagnostics).ToArray();
}
/// <summary>Gets the provider-authored operation status.</summary>
public string Status { get; }
/// <summary>Gets the typed response payload.</summary>
public TResponse Value { get; }
/// <summary>Gets provider-authored diagnostics.</summary>
public IReadOnlyList<string> Diagnostics { get; }
/// <summary>Creates an accepted typed reply.</summary>
public static CultNetOperationReply<TResponse> Accepted(TResponse value) => new("accepted", value);
/// <summary>Creates a rejected typed reply whose payload remains schema-valid.</summary>
public static CultNetOperationReply<TResponse> Rejected(
TResponse value,
params string[] diagnostics) => new("rejected", value, diagnostics);
private static string RequireNonEmpty(string value, string paramName) =>
string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value must be non-empty.", paramName)
: value.Trim();
}
/// <summary>
/// Transport-neutral typed dispatcher for CultNet application operations.
/// It owns envelope validation and serialization only; registered handlers own
/// domain validation, idempotency, state mutation, and durable receipts.
/// </summary>
public sealed class CultNetOperationServer : IDisposable
{
/// <summary>Schema used by correlated framework-level operation failures.</summary>
public const string FailureSchemaId = "gamecult.cultnet.operation_failure.v1";
private const string PayloadEncoding = "messagepack-base64";
private readonly ICultNetSchemaServer _server;
private readonly string? _sourceRuntimeId;
private readonly Dictionary<Route, IBinding> _bindings = new();
private readonly Func<CultNetOperationRequestMessage, ICultNetSchemaServerPeer, Task> _handler;
private bool _disposed;
/// <summary>Attaches one typed operation dispatcher to a schema server.</summary>
public CultNetOperationServer(ICultNetSchemaServer server, string? sourceRuntimeId = null)
{
_server = server ?? throw new ArgumentNullException(nameof(server));
_sourceRuntimeId = string.IsNullOrWhiteSpace(sourceRuntimeId) ? null : sourceRuntimeId!.Trim();
_handler = HandleAsync;
_server.OnCultNet(_handler);
}
/// <summary>
/// Registers one typed operation. Duplicate service/operation routes are rejected.
/// </summary>
public CultNetOperationServer Register<TRequest, TResponse>(
string serviceId,
string operation,
string requestSchema,
string responseSchema,
Func<CultNetOperationContext<TRequest>, Task<CultNetOperationReply<TResponse>>> handler)
{
ThrowIfDisposed();
var route = new Route(serviceId, operation);
if (_bindings.ContainsKey(route))
throw new InvalidOperationException($"CultNet operation '{route}' is already registered.");
_bindings.Add(route, new Binding<TRequest, TResponse>(
route,
requestSchema,
responseSchema,
handler ?? throw new ArgumentNullException(nameof(handler)),
_sourceRuntimeId));
return this;
}
/// <summary>
/// Registers one typed operation whose successful handler result is always accepted.
/// </summary>
public CultNetOperationServer Register<TRequest, TResponse>(
string serviceId,
string operation,
string requestSchema,
string responseSchema,
Func<CultNetOperationContext<TRequest>, Task<TResponse>> handler)
{
if (handler == null) throw new ArgumentNullException(nameof(handler));
return Register<TRequest, TResponse>(
serviceId,
operation,
requestSchema,
responseSchema,
async context => CultNetOperationReply<TResponse>.Accepted(await handler(context).ConfigureAwait(false)));
}
/// <inheritdoc />
public void Dispose()
{
if (_disposed) return;
_disposed = true;
_server.RemoveCultNetMessageListener<CultNetOperationRequestMessage>(_handler);
_bindings.Clear();
}
private async Task HandleAsync(CultNetOperationRequestMessage request, ICultNetSchemaServerPeer peer)
{
if (_disposed) return;
if (request == null) throw new ArgumentNullException(nameof(request));
if (peer == null) throw new ArgumentNullException(nameof(peer));
Route route;
try
{
route = new Route(request.ServiceId, request.Operation);
}
catch (ArgumentException error)
{
peer.SendCultNet(Failure(request, "invalid", "invalid-route", error.Message));
return;
}
if (!_bindings.TryGetValue(route, out var binding))
{
peer.SendCultNet(Failure(
request,
"unsupported",
"operation-not-registered",
$"CultNet operation '{route}' is not registered."));
return;
}
try
{
peer.SendCultNet(await binding.InvokeAsync(request).ConfigureAwait(false));
}
catch (RequestException error)
{
peer.SendCultNet(Failure(request, "invalid", error.Code, error.Message));
}
catch (Exception error)
{
peer.SendCultNet(Failure(
request,
"error",
"handler-failed",
$"CultNet operation '{route}' failed: {error.Message}"));
}
}
private CultNetOperationResponseMessage Failure(
CultNetOperationRequestMessage request,
string status,
string code,
string message)
{
return new CultNetOperationResponseMessage
{
MessageId = request.MessageId ?? "",
ServiceId = request.ServiceId ?? "",
Operation = request.Operation ?? "",
Status = status,
PayloadSchema = FailureSchemaId,
PayloadEncoding = PayloadEncoding,
Payload = Convert.ToBase64String(MessagePackSerializer.Serialize(
new CultNetOperationFailure { Code = code, Message = message },
CultNetSchemaMessageSerialization.Options)),
Diagnostics = new[] { message },
SourceRuntimeId = _sourceRuntimeId
};
}
private void ThrowIfDisposed()
{
if (_disposed) throw new ObjectDisposedException(nameof(CultNetOperationServer));
}
private interface IBinding
{
Task<CultNetOperationResponseMessage> InvokeAsync(CultNetOperationRequestMessage request);
}
private sealed class Binding<TRequest, TResponse> : IBinding
{
private readonly Route _route;
private readonly string _requestSchema;
private readonly string _responseSchema;
private readonly Func<CultNetOperationContext<TRequest>, Task<CultNetOperationReply<TResponse>>> _handler;
private readonly string? _sourceRuntimeId;
public Binding(
Route route,
string requestSchema,
string responseSchema,
Func<CultNetOperationContext<TRequest>, Task<CultNetOperationReply<TResponse>>> handler,
string? sourceRuntimeId)
{
_route = route;
_requestSchema = RequireNonEmpty(requestSchema, nameof(requestSchema));
_responseSchema = RequireNonEmpty(responseSchema, nameof(responseSchema));
_handler = handler;
_sourceRuntimeId = sourceRuntimeId;
}
public async Task<CultNetOperationResponseMessage> InvokeAsync(CultNetOperationRequestMessage request)
{
if (string.IsNullOrWhiteSpace(request.MessageId))
throw new RequestException("missing-idempotency-key", "CultNet operation omitted its idempotency key.");
if (!string.Equals(request.PayloadSchema, _requestSchema, StringComparison.Ordinal))
throw new RequestException(
"request-schema-mismatch",
$"CultNet operation '{_route}' expected payload schema '{_requestSchema}', got '{request.PayloadSchema}'.");
if (!string.Equals(request.PayloadEncoding, PayloadEncoding, StringComparison.Ordinal))
throw new RequestException(
"request-encoding-mismatch",
$"CultNet operation '{_route}' expected payload encoding '{PayloadEncoding}', got '{request.PayloadEncoding}'.");
TRequest value;
try
{
value = MessagePackSerializer.Deserialize<TRequest>(
Convert.FromBase64String(request.Payload),
CultNetSchemaMessageSerialization.Options);
}
catch (FormatException error)
{
throw new RequestException("request-payload-invalid", "CultNet operation payload was not valid base64.", error);
}
catch (MessagePackSerializationException error)
{
throw new RequestException("request-payload-invalid", "CultNet operation payload was not valid MessagePack.", error);
}
if (value == null) throw new RequestException("request-payload-null", "CultNet operation payload decoded to null.");
var reply = await _handler(new CultNetOperationContext<TRequest>(request, value)).ConfigureAwait(false)
?? throw new InvalidOperationException("CultNet operation handler returned no reply.");
if (reply.Value == null) throw new InvalidOperationException("CultNet operation handler returned a null payload.");
return new CultNetOperationResponseMessage
{
MessageId = request.MessageId,
ServiceId = _route.ServiceId,
Operation = _route.Operation,
Status = reply.Status,
PayloadSchema = _responseSchema,
PayloadEncoding = PayloadEncoding,
Payload = Convert.ToBase64String(MessagePackSerializer.Serialize(
reply.Value,
CultNetSchemaMessageSerialization.Options)),
Diagnostics = new List<string>(reply.Diagnostics).ToArray(),
SourceRuntimeId = _sourceRuntimeId
};
}
}
private sealed class RequestException : Exception
{
public RequestException(string code, string message, Exception? inner = null)
: base(message, inner)
{
Code = code;
}
public string Code { get; }
}
private readonly struct Route : IEquatable<Route>
{
public Route(string serviceId, string operation)
{
ServiceId = RequireNonEmpty(serviceId, nameof(serviceId));
Operation = RequireNonEmpty(operation, nameof(operation));
}
public string ServiceId { get; }
public string Operation { get; }
public bool Equals(Route other) =>
string.Equals(ServiceId, other.ServiceId, StringComparison.Ordinal) &&
string.Equals(Operation, other.Operation, StringComparison.Ordinal);
public override bool Equals(object? value) => value is Route route && Equals(route);
public override int GetHashCode() =>
(StringComparer.Ordinal.GetHashCode(ServiceId) * 397) ^ StringComparer.Ordinal.GetHashCode(Operation);
public override string ToString() => ServiceId + "/" + Operation;
}
private static string RequireNonEmpty(string value, string paramName) =>
string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value must be non-empty.", paramName)
: value.Trim();
}
}

27
src/GameCult.Networking/README.md
View file

@ -390,6 +390,33 @@ peer lookup and handshake before returning the same RUDP transport. Packet
semantics, schema-channel send/receive helpers, and reconnect state still live
in `GameCult.Networking`.
## Typed Operations
`CultNetOperationServer` attaches typed application handlers to any
`ICultNetSchemaServer`. It owns route/schema validation, MessagePack envelope
encoding, correlation, and framework failure replies. The application handler
owns domain validation, state mutation, idempotency, and durable receipts:
```csharp
using var operations = new CultNetOperationServer(schemaServer, "counter.provider")
.Register<IncrementRequest, IncrementReceipt>(
"counter",
"counter.increment",
"counter.increment_request.v1",
"counter.increment_receipt.v1",
command => IncrementExactlyOnceAsync(
command.IdempotencyKey,
command.Value));
```
Returning `CultNetOperationReply<T>.Rejected(...)` preserves a schema-valid
domain receipt and status. Unknown routes, malformed payloads, and schema
mismatches return the correlated `gamecult.cultnet.operation_failure.v1`
payload; `CultMeshClient.InvokeAsync` raises
`CultMeshRemoteOperationException` immediately instead of timing out. The
browser client raises the equivalent `CultMeshBrowserOperationError` with the
same status and failure code.
## Browser WebSocket Host
`GameCult.Networking.WebSockets` keeps ASP.NET Core hosting out of the

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

@ -288,6 +288,36 @@ public sealed class CultMeshClientTests
.Which.SourceRuntimeId.Should().Be("pilot-one");
}
[Test]
public async Task InvokeOperation_ThrowsCorrelatedFrameworkFailureWithoutTimingOut()
{
var connector = new DocumentConnector { RejectOperations = true };
using var mesh = new CultMeshClient(new CultMeshClientOptions
{
RendezvousEndpoints = new[] { "rudp://odin:3076" },
Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
Connectors = new[] { connector },
OperationResponseTimeout = TimeSpan.FromSeconds(5)
});
Func<Task> invoke = async () => await mesh.InvokeAsync<ClientOperationRequest, ClientOperationReceipt>(
"aetheria",
"tests.counter",
"tests.counter.increment",
"wrong.request.v1",
"tests.counter.increment_receipt.v1",
new ClientOperationRequest { Amount = 2 },
sourceRuntimeId: "pilot-one",
idempotencyKey: "pilot-command-invalid");
var failure = await invoke.Should().ThrowAsync<CultMeshRemoteOperationException>();
failure.Which.Status.Should().Be("invalid");
failure.Which.Code.Should().Be("request-schema-mismatch");
failure.Which.Message.Should().Contain("expected payload schema");
connector.Clients.Should().ContainSingle();
connector.Clients[0].Operations.Should().ContainSingle();
}
[Test]
public async Task Dispose_TerminatesDocumentWaitingForInitialSnapshot()
{
@ -369,6 +399,7 @@ public sealed class CultMeshClientTests
public List<DocumentClient> Clients { get; } = new();
public bool SuppressFirstSnapshot { get; set; }
public bool SuppressFirstSubscriptionSnapshot { get; set; }
public bool RejectOperations { get; set; }
public string ConnectorId => "document-test";
public int Priority => 0;
public int SubscribeCount => Clients.Sum(client => client.SubscribeCount);
@ -382,7 +413,8 @@ public sealed class CultMeshClientTests
"pilot surface " + (Clients.Count + 1),
snapshotsToSuppress: SuppressFirstSnapshot && Clients.Count == 0
? int.MaxValue
: SuppressFirstSubscriptionSnapshot && Clients.Count == 0 ? 1 : 0);
: SuppressFirstSubscriptionSnapshot && Clients.Count == 0 ? 1 : 0,
rejectOperations: RejectOperations);
Clients.Add(client);
return Task.FromResult<ICultNetSchemaClient>(client);
}
@ -395,11 +427,13 @@ public sealed class CultMeshClientTests
private readonly CultNetDocumentRegistry _documents;
private readonly string _text;
private readonly TaskCompletionSource<Exception> _failure = new(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly bool _rejectOperations;
private int _snapshotsToSuppress;
public DocumentClient(string text, int snapshotsToSuppress = 0)
public DocumentClient(string text, int snapshotsToSuppress = 0, bool rejectOperations = false)
{
_text = text;
_snapshotsToSuppress = snapshotsToSuppress;
_rejectOperations = rejectOperations;
var cache = CultMesh.CreateCultCacheDocumentRegistry(typeof(ClientTestDocument));
_documents = CultMesh.CreateCultNetDocumentRegistry(new[] { typeof(ClientTestDocument) }, cache);
}
@ -420,6 +454,29 @@ public sealed class CultMeshClientTests
if (message is CultNetOperationRequestMessage operation)
{
Operations.Add(operation);
if (_rejectOperations)
{
var failure = new CultNetOperationFailure
{
Code = "request-schema-mismatch",
Message = "CultNet operation expected payload schema tests.counter.increment_request.v1."
};
var rejected = new CultNetOperationResponseMessage
{
MessageId = operation.MessageId,
ServiceId = operation.ServiceId,
Operation = operation.Operation,
Status = "invalid",
PayloadSchema = CultNetOperationServer.FailureSchemaId,
Payload = Convert.ToBase64String(MessagePackSerializer.Serialize(
failure,
CultNetSchemaMessageSerialization.Options)),
Diagnostics = new[] { failure.Message },
SourceRuntimeId = "provider-daemon"
};
foreach (var handler in _operationResponses.ToArray()) handler(rejected);
return;
}
var decodedRequest = MessagePackSerializer.Deserialize<ClientOperationRequest>(
Convert.FromBase64String(operation.Payload),
CultNetSchemaMessageSerialization.Options);

205
tests/GameCult.Networking.Tests/CultNetOperationServerTests.cs Normal file
View file

@ -0,0 +1,205 @@
#nullable enable
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using MessagePack;
using NUnit.Framework;
namespace GameCult.Networking.Tests
{
[TestFixture]
public sealed class CultNetOperationServerTests
{
[Test]
public async Task DispatchesTypedRequestAndCorrelatedReplyWithoutEnvelopePlumbing()
{
var transport = new FakeServer();
var peer = new FakePeer();
CultNetOperationContext<AddRequest>? observed = null;
using var operations = new CultNetOperationServer(transport, "sample.provider")
.Register<AddRequest, AddReceipt>(
"sample.counter",
"sample.counter.add",
"sample.add.v1",
"sample.add_receipt.v1",
context =>
{
observed = context;
return Task.FromResult(new AddReceipt { Count = context.Value.Amount + 4 });
});
await transport.DispatchAsync(Request(new AddRequest { Amount = 3 }), peer);
var response = peer.Messages.OfType<CultNetOperationResponseMessage>().Single();
var receipt = MessagePackSerializer.Deserialize<AddReceipt>(
Convert.FromBase64String(response.Payload),
CultNetSchemaMessageSerialization.Options);
Assert.Multiple(() =>
{
Assert.That(observed!.Value.Amount, Is.EqualTo(3));
Assert.That(observed.IdempotencyKey, Is.EqualTo("command-7"));
Assert.That(observed.SourceRuntimeId, Is.EqualTo("sample.browser"));
Assert.That(response.MessageId, Is.EqualTo("command-7"));
Assert.That(response.ServiceId, Is.EqualTo("sample.counter"));
Assert.That(response.Operation, Is.EqualTo("sample.counter.add"));
Assert.That(response.Status, Is.EqualTo("accepted"));
Assert.That(response.PayloadSchema, Is.EqualTo("sample.add_receipt.v1"));
Assert.That(response.SourceRuntimeId, Is.EqualTo("sample.provider"));
Assert.That(receipt.Count, Is.EqualTo(7));
});
}
[Test]
public async Task PreservesTypedRejectionAndDiagnostics()
{
var transport = new FakeServer();
var peer = new FakePeer();
using var operations = new CultNetOperationServer(transport)
.Register<AddRequest, AddReceipt>(
"sample.counter",
"sample.counter.add",
"sample.add.v1",
"sample.add_receipt.v1",
_ => Task.FromResult(CultNetOperationReply<AddReceipt>.Rejected(
new AddReceipt { Count = 4 },
"counter is locked")));
await transport.DispatchAsync(Request(new AddRequest { Amount = 3 }), peer);
var response = peer.Messages.OfType<CultNetOperationResponseMessage>().Single();
Assert.That(response.Status, Is.EqualTo("rejected"));
Assert.That(response.Diagnostics, Is.EqualTo(new[] { "counter is locked" }));
Assert.That(
MessagePackSerializer.Deserialize<AddReceipt>(
Convert.FromBase64String(response.Payload),
CultNetSchemaMessageSerialization.Options).Count,
Is.EqualTo(4));
}
[Test]
public async Task RejectsWrongSchemaBeforeApplicationHandler()
{
var transport = new FakeServer();
var peer = new FakePeer();
var calls = 0;
using var operations = new CultNetOperationServer(transport)
.Register<AddRequest, AddReceipt>(
"sample.counter",
"sample.counter.add",
"sample.add.v1",
"sample.add_receipt.v1",
context =>
{
calls++;
return Task.FromResult(new AddReceipt { Count = context.Value.Amount });
});
var request = Request(new AddRequest { Amount = 3 });
request.PayloadSchema = "wrong.v1";
await transport.DispatchAsync(request, peer);
Assert.That(calls, Is.Zero);
var response = peer.Messages.OfType<CultNetOperationResponseMessage>().Single();
var failure = MessagePackSerializer.Deserialize<CultNetOperationFailure>(
Convert.FromBase64String(response.Payload),
CultNetSchemaMessageSerialization.Options);
Assert.Multiple(() =>
{
Assert.That(response.MessageId, Is.EqualTo("command-7"));
Assert.That(response.Status, Is.EqualTo("invalid"));
Assert.That(response.PayloadSchema, Is.EqualTo(CultNetOperationServer.FailureSchemaId));
Assert.That(failure.Code, Is.EqualTo("request-schema-mismatch"));
Assert.That(failure.Message, Does.Contain("expected payload schema 'sample.add.v1'"));
});
}
[Test]
public void RejectsDuplicateRouteAndDetachesOnDispose()
{
var transport = new FakeServer();
var operations = new CultNetOperationServer(transport);
operations.Register<AddRequest, AddReceipt>(
"sample.counter",
"sample.counter.add",
"sample.add.v1",
"sample.add_receipt.v1",
context => Task.FromResult(new AddReceipt { Count = context.Value.Amount }));
Assert.That(() => operations.Register<AddRequest, AddReceipt>(
"sample.counter",
"sample.counter.add",
"sample.add.v1",
"sample.add_receipt.v1",
context => Task.FromResult(new AddReceipt { Count = context.Value.Amount })),
Throws.InvalidOperationException);
Assert.That(transport.HasHandler<CultNetOperationRequestMessage>(), Is.True);
operations.Dispose();
Assert.That(transport.HasHandler<CultNetOperationRequestMessage>(), Is.False);
Assert.That(() => operations.Register<AddRequest, AddReceipt>(
"sample.counter",
"sample.counter.add",
"sample.add.v1",
"sample.add_receipt.v1",
context => Task.FromResult(new AddReceipt { Count = context.Value.Amount })),
Throws.TypeOf<ObjectDisposedException>());
}
private static CultNetOperationRequestMessage Request(AddRequest value) => new()
{
MessageId = "command-7",
ServiceId = "sample.counter",
Operation = "sample.counter.add",
PayloadSchema = "sample.add.v1",
PayloadEncoding = "messagepack-base64",
Payload = Convert.ToBase64String(MessagePackSerializer.Serialize(
value,
CultNetSchemaMessageSerialization.Options)),
SourceRuntimeId = "sample.browser"
};
[MessagePackObject]
public sealed class AddRequest
{
[Key(0)] public int Amount { get; set; }
}
[MessagePackObject]
public sealed class AddReceipt
{
[Key(0)] public int Count { get; set; }
}
private sealed class FakeServer : ICultNetSchemaServer
{
private readonly Dictionary<Type, Delegate> _handlers = new();
public void OnCultNet<TMessage>(Func<TMessage, ICultNetSchemaServerPeer, Task> callback)
where TMessage : ICultNetSchemaMessage => _handlers[typeof(TMessage)] = callback;
public void RemoveCultNetMessageListener<TMessage>(Delegate callback)
where TMessage : ICultNetSchemaMessage
{
if (_handlers.TryGetValue(typeof(TMessage), out var current) && current == callback)
_handlers.Remove(typeof(TMessage));
}
public bool HasHandler<TMessage>() where TMessage : ICultNetSchemaMessage =>
_handlers.ContainsKey(typeof(TMessage));
public Task DispatchAsync<TMessage>(TMessage message, ICultNetSchemaServerPeer peer)
where TMessage : ICultNetSchemaMessage =>
((Func<TMessage, ICultNetSchemaServerPeer, Task>)_handlers[typeof(TMessage)])(message, peer);
}
private sealed class FakePeer : ICultNetSchemaServerPeer
{
public List<ICultNetSchemaMessage> Messages { get; } = new();
public void SendCultNet<TMessage>(TMessage message) where TMessage : ICultNetSchemaMessage =>
Messages.Add(message);
}
}
}