diff --git a/packages/cultmesh-browser/README.md b/packages/cultmesh-browser/README.md index acf349e6..62f70585 100644 --- a/packages/cultmesh-browser/README.md +++ b/packages/cultmesh-browser/README.md @@ -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 diff --git a/packages/cultmesh-browser/src/index.ts b/packages/cultmesh-browser/src/index.ts index eb2420e3..b14e05ec 100644 --- a/packages/cultmesh-browser/src/index.ts +++ b/packages/cultmesh-browser/src/index.ts @@ -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(message), + )); + } catch (error) { + pending.reject(error instanceof Error + ? error + : new Error("CultMesh operation failure payload was malformed.")); + } + return; + } pending.resolve(message); return; } diff --git a/packages/cultmesh-browser/test/cultmesh-browser.test.ts b/packages/cultmesh-browser/test/cultmesh-browser.test.ts index b05b184b..23f6f5a6 100644 --- a/packages/cultmesh-browser/test/cultmesh-browser.test.ts +++ b/packages/cultmesh-browser/test/cultmesh-browser.test.ts @@ -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>(); + #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) { diff --git a/samples/eve-browser-network/Program.cs b/samples/eve-browser-network/Program.cs index 5b8e132f..82c09c19 100644 --- a/samples/eve-browser-network/Program.cs +++ b/samples/eve-browser-network/Program.cs @@ -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(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( - 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(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( + "sample.counter", + "sample.counter.ping", + "sample.ping_request.v1", + "sample.ping_receipt.v1", + context => Task.FromResult(new PingReceipt { Sequence = context.Value.Sequence })) + .Register( + "sample.counter", + "sample.counter.increment", + "sample.increment.v1", + "gamecult.eve.command_receipt.v1", + async context => { - var input = MessagePackSerializer.Deserialize( - Convert.FromBase64String(request.Payload), - CultNetSchemaMessageSerialization.Options); - receipt = IncrementReceipt.Accepted(request.MessageId, current.Count + input.Amount); - var receipts = new Dictionary(current.Receipts, StringComparer.Ordinal) + await operationGate.WaitAsync(); + try { - [request.MessageId] = receipt - }; - await database.PutAsync(new CultRecordKey(counterKey), new CounterState + var current = cache.Get(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(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( - 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 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(StringComparer.Ordinal), - [ - new EveSurfaceComponent( - "counter.value", - "metric", - new Dictionary(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(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 diff --git a/samples/eve-browser-network/README.md b/samples/eve-browser-network/README.md index c039cd42..bbcc95ca 100644 --- a/samples/eve-browser-network/README.md +++ b/samples/eve-browser-network/README.md @@ -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` 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 diff --git a/scripts/verify-eve-getting-started.ps1 b/scripts/verify-eve-getting-started.ps1 index 57dd0a61..be49aafe 100644 --- a/scripts/verify-eve-getting-started.ps1 +++ b/scripts/verify-eve-getting-started.ps1 @@ -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 ` diff --git a/src/GameCult.Mesh/CultMeshClient.cs b/src/GameCult.Mesh/CultMeshClient.cs index 970e0f2f..1d33c3ec 100644 --- a/src/GameCult.Mesh/CultMeshClient.cs +++ b/src/GameCult.Mesh/CultMeshClient.cs @@ -82,6 +82,29 @@ namespace GameCult.Mesh public TResponse Value { get; } } + /// Correlated framework-level failure returned before a typed domain reply exists. + 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(); + } + + /// Gets the provider-authored failure status. + public string Status { get; } + + /// Gets the stable machine-readable failure code. + public string Code { get; } + + /// Gets provider diagnostics carried by the response. + public IReadOnlyList Diagnostics { get; } + } + /// Owns one shared live document subscription until disposed. public sealed class CultMeshDocumentLease : 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( + 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}'."); diff --git a/src/GameCult.Mesh/README.md b/src/GameCult.Mesh/README.md index 6b9be603..b0ff9e06 100644 --- a/src/GameCult.Mesh/README.md +++ b/src/GameCult.Mesh/README.md @@ -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") diff --git a/src/GameCult.Mesh/docs/getting-started/03-publish-an-eve-surface.md b/src/GameCult.Mesh/docs/getting-started/03-publish-an-eve-surface.md index a9126cd7..a910cf97 100644 --- a/src/GameCult.Mesh/docs/getting-started/03-publish-an-eve-surface.md +++ b/src/GameCult.Mesh/docs/getting-started/03-publish-an-eve-surface.md @@ -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( + "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 diff --git a/src/GameCult.Networking/CultNetOperationServer.cs b/src/GameCult.Networking/CultNetOperationServer.cs new file mode 100644 index 00000000..6261328e --- /dev/null +++ b/src/GameCult.Networking/CultNetOperationServer.cs @@ -0,0 +1,348 @@ +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using MessagePack; + +namespace GameCult.Networking +{ + /// Portable failure payload for operation routing and envelope validation. + [MessagePackObject] + public sealed class CultNetOperationFailure + { + /// Gets or sets the stable machine-readable failure code. + [Key("code")] public string Code { get; set; } = "operation-error"; + + /// Gets or sets the human-readable failure description. + [Key("message")] public string Message { get; set; } = "CultNet operation failed."; + } + + /// + /// 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. + /// + public sealed class CultNetOperationContext + { + internal CultNetOperationContext(CultNetOperationRequestMessage envelope, TRequest value) + { + Envelope = envelope ?? throw new ArgumentNullException(nameof(envelope)); + Value = value; + } + + /// Gets the decoded application request. + public TRequest Value { get; } + + /// Gets the durable caller-supplied idempotency key. + public string IdempotencyKey => Envelope.MessageId; + + /// Gets the calling runtime identity, when supplied. + public string? SourceRuntimeId => Envelope.SourceRuntimeId; + + /// Gets the optional target runtime identity. + public string? TargetRuntimeId => Envelope.TargetRuntimeId; + + /// Gets the original transport-neutral envelope for diagnostics. + public CultNetOperationRequestMessage Envelope { get; } + } + + /// Provider-authored typed result for one CultNet operation. + public sealed class CultNetOperationReply + { + /// Creates one provider-authored typed reply. + public CultNetOperationReply( + string status, + TResponse value, + IEnumerable? diagnostics = null) + { + Status = RequireNonEmpty(status, nameof(status)); + Value = value; + Diagnostics = diagnostics == null ? Array.Empty() : new List(diagnostics).ToArray(); + } + + /// Gets the provider-authored operation status. + public string Status { get; } + + /// Gets the typed response payload. + public TResponse Value { get; } + + /// Gets provider-authored diagnostics. + public IReadOnlyList Diagnostics { get; } + + /// Creates an accepted typed reply. + public static CultNetOperationReply Accepted(TResponse value) => new("accepted", value); + + /// Creates a rejected typed reply whose payload remains schema-valid. + public static CultNetOperationReply 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(); + } + + /// + /// 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. + /// + public sealed class CultNetOperationServer : IDisposable + { + /// Schema used by correlated framework-level operation failures. + 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 _bindings = new(); + private readonly Func _handler; + private bool _disposed; + + /// Attaches one typed operation dispatcher to a schema server. + 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); + } + + /// + /// Registers one typed operation. Duplicate service/operation routes are rejected. + /// + public CultNetOperationServer Register( + string serviceId, + string operation, + string requestSchema, + string responseSchema, + Func, Task>> 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( + route, + requestSchema, + responseSchema, + handler ?? throw new ArgumentNullException(nameof(handler)), + _sourceRuntimeId)); + return this; + } + + /// + /// Registers one typed operation whose successful handler result is always accepted. + /// + public CultNetOperationServer Register( + string serviceId, + string operation, + string requestSchema, + string responseSchema, + Func, Task> handler) + { + if (handler == null) throw new ArgumentNullException(nameof(handler)); + return Register( + serviceId, + operation, + requestSchema, + responseSchema, + async context => CultNetOperationReply.Accepted(await handler(context).ConfigureAwait(false))); + } + + /// + public void Dispose() + { + if (_disposed) return; + _disposed = true; + _server.RemoveCultNetMessageListener(_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 InvokeAsync(CultNetOperationRequestMessage request); + } + + private sealed class Binding : IBinding + { + private readonly Route _route; + private readonly string _requestSchema; + private readonly string _responseSchema; + private readonly Func, Task>> _handler; + private readonly string? _sourceRuntimeId; + + public Binding( + Route route, + string requestSchema, + string responseSchema, + Func, Task>> handler, + string? sourceRuntimeId) + { + _route = route; + _requestSchema = RequireNonEmpty(requestSchema, nameof(requestSchema)); + _responseSchema = RequireNonEmpty(responseSchema, nameof(responseSchema)); + _handler = handler; + _sourceRuntimeId = sourceRuntimeId; + } + + public async Task 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( + 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(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(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 + { + 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(); + } +} diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md index cf463e72..de8496d3 100644 --- a/src/GameCult.Networking/README.md +++ b/src/GameCult.Networking/README.md @@ -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( + "counter", + "counter.increment", + "counter.increment_request.v1", + "counter.increment_receipt.v1", + command => IncrementExactlyOnceAsync( + command.IdempotencyKey, + command.Value)); +``` + +Returning `CultNetOperationReply.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 diff --git a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs index 10af2417..82934a54 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs @@ -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 invoke = async () => await mesh.InvokeAsync( + "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(); + 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 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(client); } @@ -395,11 +427,13 @@ public sealed class CultMeshClientTests private readonly CultNetDocumentRegistry _documents; private readonly string _text; private readonly TaskCompletionSource _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( Convert.FromBase64String(operation.Payload), CultNetSchemaMessageSerialization.Options); diff --git a/tests/GameCult.Networking.Tests/CultNetOperationServerTests.cs b/tests/GameCult.Networking.Tests/CultNetOperationServerTests.cs new file mode 100644 index 00000000..f82456fb --- /dev/null +++ b/tests/GameCult.Networking.Tests/CultNetOperationServerTests.cs @@ -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? observed = null; + using var operations = new CultNetOperationServer(transport, "sample.provider") + .Register( + "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().Single(); + var receipt = MessagePackSerializer.Deserialize( + 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( + "sample.counter", + "sample.counter.add", + "sample.add.v1", + "sample.add_receipt.v1", + _ => Task.FromResult(CultNetOperationReply.Rejected( + new AddReceipt { Count = 4 }, + "counter is locked"))); + + await transport.DispatchAsync(Request(new AddRequest { Amount = 3 }), peer); + + var response = peer.Messages.OfType().Single(); + Assert.That(response.Status, Is.EqualTo("rejected")); + Assert.That(response.Diagnostics, Is.EqualTo(new[] { "counter is locked" })); + Assert.That( + MessagePackSerializer.Deserialize( + 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( + "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().Single(); + var failure = MessagePackSerializer.Deserialize( + 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( + "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( + "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(), Is.True); + + operations.Dispose(); + + Assert.That(transport.HasHandler(), Is.False); + Assert.That(() => operations.Register( + "sample.counter", + "sample.counter.add", + "sample.add.v1", + "sample.add_receipt.v1", + context => Task.FromResult(new AddReceipt { Count = context.Value.Amount })), + Throws.TypeOf()); + } + + 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 _handlers = new(); + + public void OnCultNet(Func callback) + where TMessage : ICultNetSchemaMessage => _handlers[typeof(TMessage)] = callback; + + public void RemoveCultNetMessageListener(Delegate callback) + where TMessage : ICultNetSchemaMessage + { + if (_handlers.TryGetValue(typeof(TMessage), out var current) && current == callback) + _handlers.Remove(typeof(TMessage)); + } + + public bool HasHandler() where TMessage : ICultNetSchemaMessage => + _handlers.ContainsKey(typeof(TMessage)); + + public Task DispatchAsync(TMessage message, ICultNetSchemaServerPeer peer) + where TMessage : ICultNetSchemaMessage => + ((Func)_handlers[typeof(TMessage)])(message, peer); + } + + private sealed class FakePeer : ICultNetSchemaServerPeer + { + public List Messages { get; } = new(); + + public void SendCultNet(TMessage message) where TMessage : ICultNetSchemaMessage => + Messages.Add(message); + } + } +}