From 56b0c1679c4bc07940dcbcd69d773edb6fa27ecc Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 22:06:55 +0000 Subject: [PATCH 1/7] cultmesh-rs: receipt_interop_server example, the real RUDP document server on an ephemeral port for the interop harness Co-Authored-By: Claude Sonnet 5.5 --- .../examples/receipt_interop_server.rs | 77 +++++++++++++++++++ 1 file changed, 77 insertions(+) create mode 100644 packages/cultmesh-rs/examples/receipt_interop_server.rs diff --git a/packages/cultmesh-rs/examples/receipt_interop_server.rs b/packages/cultmesh-rs/examples/receipt_interop_server.rs new file mode 100644 index 00000000..b8c66fd9 --- /dev/null +++ b/packages/cultmesh-rs/examples/receipt_interop_server.rs @@ -0,0 +1,77 @@ +//! The real Rust RUDP document server on an ephemeral loopback port, for the cross-runtime write-receipt interop +//! harness (packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts). It is the production server with a +//! recording sink and nothing else: stdout carries one JSON line per fact the harness asserts on, and the server runs +//! until stdin closes. +//! +//! {"event":"listening","port":N} +//! {"event":"applied","op":"put","messageId":"...","schemaId":"...","recordKey":"..."} +//! +//! Record keys that start with `refuse:` are refused by the sink (authorize and commit), which is how the harness +//! asks for a refused write. +use anyhow::Result; +use cultmesh_rs::{ + CultMeshRudpDocumentServer, CultMeshRudpRawDocumentReceipt, CultMeshRudpRawDocumentSink, + CultMeshRudpSnapshotQuery, CultMeshSystemClock, +}; +use cultnet_rs::CultNetRawDocumentRecord; +use serde_json::json; +use std::io::Read; +use std::net::UdpSocket; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::thread; +use std::time::Duration; + +const REFUSED_PREFIX: &str = "refuse:"; + +struct RecordingSink; + +impl CultMeshRudpRawDocumentSink for RecordingSink { + fn authorize_raw_document(&mut self, receipt: &CultMeshRudpRawDocumentReceipt) -> Result<()> { + if receipt.document.record_key.starts_with(REFUSED_PREFIX) { + anyhow::bail!("not authorized"); + } + Ok(()) + } + + fn accept_raw_document(&mut self, receipt: CultMeshRudpRawDocumentReceipt) -> Result<()> { + if receipt.document.record_key.starts_with(REFUSED_PREFIX) { + anyhow::bail!("not authorized"); + } + println!( + "{}", + json!({ + "event": "applied", + "op": "put", + "messageId": receipt.message_id, + "schemaId": receipt.document.schema_id, + "recordKey": receipt.document.record_key, + }) + ); + Ok(()) + } +} + +fn main() -> Result<()> { + let source = |_: &CultMeshRudpSnapshotQuery| -> Result> { Ok(Vec::new()) }; + let mut server = CultMeshRudpDocumentServer::new( + UdpSocket::bind("127.0.0.1:0")?, + RecordingSink, + source, + CultMeshSystemClock::default(), + Default::default(), + )?; + println!("{}", json!({ "event": "listening", "port": server.local_addr()?.port() })); + let stop = Arc::new(AtomicBool::new(false)); + let watcher = Arc::clone(&stop); + thread::spawn(move || { + let mut sink = Vec::new(); + let _ = std::io::stdin().read_to_end(&mut sink); + watcher.store(true, Ordering::SeqCst); + }); + while !stop.load(Ordering::SeqCst) { + server.poll_once()?; + thread::sleep(Duration::from_millis(1)); + } + Ok(()) +} -- 2.47.3 From e4d2446dbe4dd268ee3b7bd20f686d72e66a53e3 Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 22:06:55 +0000 Subject: [PATCH 2/7] C# InteropPeer: receipt-serve (CultNetDatabaseServer on an ephemeral port) and receipt-relay (LiteNetLib byte shim for the TS client) Co-Authored-By: Claude Sonnet 5.5 --- .../Program.cs | 8 +- .../ReceiptHarness.cs | 118 ++++++++++++++++++ 2 files changed, 125 insertions(+), 1 deletion(-) create mode 100644 tests/GameCult.Networking.InteropPeer/ReceiptHarness.cs diff --git a/tests/GameCult.Networking.InteropPeer/Program.cs b/tests/GameCult.Networking.InteropPeer/Program.cs index 17a1ea89..fcfcf09c 100644 --- a/tests/GameCult.Networking.InteropPeer/Program.cs +++ b/tests/GameCult.Networking.InteropPeer/Program.cs @@ -25,7 +25,7 @@ static async Task ProgramMainAsync(string[] args) { if (args.Length == 0) { - throw new InvalidOperationException("Expected mode: serve | probe | dial | rudp-serve-once | rudp-dial-once | rudp-serve-message-once | rudp-dial-message-once | quic-realtime-serve"); + throw new InvalidOperationException("Expected mode: serve | probe | dial | rudp-serve-once | rudp-dial-once | rudp-serve-message-once | rudp-dial-message-once | quic-realtime-serve | receipt-serve | receipt-relay"); } var mode = args[0]; @@ -53,6 +53,12 @@ static async Task ProgramMainAsync(string[] args) case "rudp-dial-message-once": RudpDialMessageOnce(options); return 0; + case "receipt-serve": + await GameCult.Networking.InteropPeer.ReceiptHarness.ServeAsync(); + return 0; + case "receipt-relay": + await GameCult.Networking.InteropPeer.ReceiptHarness.RelayAsync(int.Parse(options["port"], CultureInfo.InvariantCulture)); + return 0; case "quic-realtime-serve": await QuicRealtimeServeAsync(ParseQuicRealtimeServeConfig(options)); return 0; diff --git a/tests/GameCult.Networking.InteropPeer/ReceiptHarness.cs b/tests/GameCult.Networking.InteropPeer/ReceiptHarness.cs new file mode 100644 index 00000000..75eaf4da --- /dev/null +++ b/tests/GameCult.Networking.InteropPeer/ReceiptHarness.cs @@ -0,0 +1,118 @@ +using System; +using System.IO; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using GameCult.Caching; +using GameCult.Networking; +using LiteNetLib; +using R3; + +// The real C# receipt server (CultNetDatabaseServer on Server) and the one transport shim the TypeScript client needs to reach it, +// for the cross-runtime write-receipt interop harness (packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts). +// TypeScript has no LiteNetLib client, so the shim moves bytes and nothing else: each CultNet frame is a LiteNetLib reliable +// ordered payload, and on the pipe it is a 4-byte big-endian length followed by the frame. +namespace GameCult.Networking.InteropPeer +{ + [CultDocument("receipt.interop_note", ReceiptHarness.SchemaId)] + [MessagePack.MessagePackObject] + public sealed class ReceiptInteropNote + { + [MessagePack.Key(0)] + public string Text { get; set; } = string.Empty; + } + + internal static class ReceiptHarness + { + public const string SchemaId = "receipt.interop_note.v1"; + public const string ShardId = "receipt-interop-shard"; + // A record key with this prefix is refused by the authority: how the harness asks for a refused write. + public const string RefusedPrefix = "refuse:"; + + private sealed class Authority : ICultNetRemoteWriteAuthority + { + public CultNetRemoteWriteDecision Decide(CultNetRemoteWriteRequest request) => + request.RecordKey.ToString().StartsWith(RefusedPrefix, StringComparison.Ordinal) + ? CultNetRemoteWriteDecision.Refuse("denied") + : CultNetRemoteWriteDecision.Grant(); + } + + private static void Emit(object fact) + { + lock (Console.Out) Console.Out.WriteLine(JsonSerializer.Serialize(fact)); + } + + // stdout: {"event":"listening","port":N} then {"event":"applied"|"removed","recordKey":"..."} per committed change; runs until stdin closes. + public static async Task ServeAsync() + { + var cache = new CultCache(); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + Shards = [new CultNetShardDescriptor(ShardId, "primary", epoch: 1, isPrimary: true, schemaIds: [SchemaId])] + }); + using var server = new Server(cache, ServerSecurityOptions.Development()) { Port = 0 }; + using var databaseServer = new CultNetDatabaseServer(server, database, new CultNetDatabaseServerOptions + { + RemoteWriteAuthority = new Authority() + }); + using var watch = cache.Watch().Subscribe(change => + Emit(new { @event = change.Kind == CultCacheDocumentChangeKind.Removed ? "removed" : "applied", recordKey = change.Key.ToString() })); + server.Start(); + Emit(new { @event = "listening", port = server.Port, shardId = ShardId, schemaId = SchemaId }); + await Task.Run(() => Console.In.ReadToEnd()); + } + + // Connects to the server on `port` and relays frames between stdin/stdout and the connection; exits when either ends. + public static async Task RelayAsync(int port) + { + var listener = new EventBasedNetListener(); + var connected = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var ended = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var output = Console.OpenStandardOutput(); + var gate = new object(); + listener.PeerConnectedEvent += _ => connected.TrySetResult(true); + listener.PeerDisconnectedEvent += (_, _) => ended.TrySetResult(true); + listener.NetworkReceiveEvent += (_, reader, _, _) => + { + var frame = reader.GetRemainingBytes(); + var header = new byte[] { (byte)(frame.Length >> 24), (byte)(frame.Length >> 16), (byte)(frame.Length >> 8), (byte)frame.Length }; + lock (gate) + { + output.Write(header); + output.Write(frame); + output.Flush(); + } + }; + var manager = new NetManager(listener) { UnsyncedEvents = true }; + manager.Start(0); + var peer = manager.Connect("127.0.0.1", port, ServerSecurityOptions.Development().ConnectionKey); + if (await Task.WhenAny(connected.Task, Task.Delay(TimeSpan.FromSeconds(10))) != connected.Task) + throw new InvalidOperationException("the relay did not connect"); + var input = Console.OpenStandardInput(); + var pump = Task.Run(() => + { + var header = new byte[4]; + while (ReadExactly(input, header)) + { + var frame = new byte[(header[0] << 24) | (header[1] << 16) | (header[2] << 8) | header[3]]; + if (!ReadExactly(input, frame)) break; + peer.Send(frame, DeliveryMethod.ReliableOrdered); + } + }); + await Task.WhenAny(pump, ended.Task); + manager.Stop(); + } + + private static bool ReadExactly(Stream stream, byte[] buffer) + { + var read = 0; + while (read < buffer.Length) + { + var count = stream.Read(buffer, read, buffer.Length - read); + if (count <= 0) return false; + read += count; + } + return true; + } + } +} -- 2.47.3 From 140eca51384ddac73c61dff68afa2bee5913881c Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 22:06:55 +0000 Subject: [PATCH 3/7] cultmesh-ts: TS write-receipt client against the real Rust and C# receipt servers (test:interop) Co-Authored-By: Claude Sonnet 5.5 --- contracts/cultnet/interop/write-receipt.md | 2 + packages/cultmesh-ts/package.json | 3 +- .../interop/write-receipt-interop.test.ts | 418 ++++++++++++++++++ 3 files changed, 422 insertions(+), 1 deletion(-) create mode 100644 packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts diff --git a/contracts/cultnet/interop/write-receipt.md b/contracts/cultnet/interop/write-receipt.md index 35756ecf..4b13aa72 100644 --- a/contracts/cultnet/interop/write-receipt.md +++ b/contracts/cultnet/interop/write-receipt.md @@ -78,6 +78,8 @@ Those clients complete *successfully* when the window ends with no refusal. Unde Advertising is per server class, not per transport. A class advertises both frames in its hello only if it implements the receipt rules end to end; the C# RUDP schema server stays "no" until it answers writes with receipts. +Cross-runtime check: `packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts` (`npm run test:interop` in `packages/cultmesh-ts`) runs the TypeScript client (`WriteTracker`, the decoder, the one-shot writer) against the real Rust RUDP document server and the real `CultNetDatabaseServer` over loopback, one scenario set against both. Known gap it records as a todo: the Rust RUDP document server refuses every delete (`write_not_authorized: the server does not take deletes`), where this spec has an admitted delete receipted like a put. + ## Conformance follow-ups (not in this PR) C# server and client, Rust, TypeScript, Python, Kotlin: emit and parse the two frames; hello capability; tri-state client outcome; replace the #55 to #58 window-success; re-capture the wire vectors in each runtime. Server write paths include the shard write forwarder (rule 6) and the Odin catalog. diff --git a/packages/cultmesh-ts/package.json b/packages/cultmesh-ts/package.json index 92962cbc..0fef7419 100644 --- a/packages/cultmesh-ts/package.json +++ b/packages/cultmesh-ts/package.json @@ -24,7 +24,8 @@ "build": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json", "prepack": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json", "typecheck": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json --pretty false", - "test": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json && node ../../node_modules/typescript/bin/tsc -p tsconfig.test.json && node --expose-gc --test dist-test/test/*.test.js" + "test": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json && node ../../node_modules/typescript/bin/tsc -p tsconfig.test.json && node --expose-gc --test dist-test/test/*.test.js", + "test:interop": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json && node ../../node_modules/typescript/bin/tsc -p tsconfig.test.json && node --test --test-force-exit dist-test/test/interop/write-receipt-interop.test.js" }, "keywords": [ "cultmesh", diff --git a/packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts b/packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts new file mode 100644 index 00000000..8a646a50 --- /dev/null +++ b/packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts @@ -0,0 +1,418 @@ +// The TypeScript write-receipt client (cultnet-ts WriteTracker and decoder, cultmesh-ts one-shot writer) against the two real +// receipt servers over loopback, as child processes on ephemeral ports: +// +// rust packages/cultmesh-rs/examples/receipt_interop_server.rs, the RUDP document server, reached over RUDP. +// csharp tests/GameCult.Networking.InteropPeer `receipt-serve`, CultNetDatabaseServer on Server (LiteNetLib). TypeScript has no +// LiteNetLib client, so a second child, `receipt-relay`, carries bytes between the test and the server (4-byte +// big-endian length, then the CultNet frame: CultNetPeer's own stream framing). It decodes and judges nothing. +// +// Every scenario runs against both targets; its outcome is read from the WriteTracker (admitted, refused, unconfirmed), and the +// server's own stdout facts say what was committed. Contract: contracts/cultnet/interop/write-receipt.md and its vectors. +// +// Needs cargo and dotnet; run it with `npm run test:interop` in packages/cultmesh-ts (CARGO and DOTNET override the commands; +// RECEIPT_INTEROP_PREBUILT=1 skips the two builds). +import assert from "node:assert/strict"; +import { execFile, spawn, spawnSync, type ChildProcess } from "node:child_process"; +import { readFileSync } from "node:fs"; +import { join, resolve } from "node:path"; +import { Duplex } from "node:stream"; +import { after, before, describe, test } from "node:test"; +import { promisify } from "node:util"; +import { encode } from "@msgpack/msgpack"; +import { z } from "zod"; +import { defineDocumentType } from "@gamecult/cultcache-ts"; +import { + CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS, + CultNetPeer, + WriteTracker, + defineCultNetDocumentBinding, + helloAdvertisesWriteReceipts, + resultWindowMs, + type CultNetErrorV1Message, + type CultNetHelloMessage, + type CultNetMessage, + type WriteSettlement, +} from "@gamecult/cultnet-ts"; +import { CultMesh, CultMeshRudpWriteRefusedError, CultMeshRudpWriteUnconfirmedError } from "../../src/index"; + +const execFileAsync = promisify(execFile); +// dist-test/test/interop -> the repository root +const repoRoot = resolve(__dirname, "..", "..", "..", "..", ".."); +const cargo = process.env.CARGO ?? "cargo"; +const dotnet = process.env.DOTNET ?? "dotnet"; +const rustManifestDir = join(repoRoot, "packages", "cultmesh-rs"); +const csharpProject = join(repoRoot, "tests", "GameCult.Networking.InteropPeer", "GameCult.Networking.InteropPeer.csproj"); +const csharpDll = join(repoRoot, "bin", "GameCult.Networking.InteropPeer", "Debug", "net10.0", "GameCult.Networking.InteropPeer.dll"); +// Set when the server binaries were built beforehand (the builds themselves cannot start under a tight address-space cap). +const prebuilt = process.env.RECEIPT_INTEROP_PREBUILT === "1"; +const vectors = JSON.parse(readFileSync(join(repoRoot, "contracts", "cultnet", "interop", "write-receipt-vectors.json"), "utf8")); + +const SCHEMA_ID = "receipt.interop_note.v1"; +const UNREADABLE = "message_invalid: the message could not be read"; +const sleep = (ms: number) => new Promise((done) => setTimeout(done, ms)); + +async function waitFor(predicate: () => boolean, what: string, ms = 5_000): Promise { + const startedAt = Date.now(); + while (!predicate()) { + if (Date.now() - startedAt > ms) throw new Error(`timed out waiting for ${what}`); + await sleep(5); + } +} + +// ---- the servers ------------------------------------------------------------------------------------------------------- + +interface Fact { event: string; recordKey?: string; messageId?: string; [key: string]: unknown } + +/** A running server child and the JSON facts it prints on stdout. */ +class ServerProcess { + readonly facts: Fact[] = []; + stderr = ""; + port = 0; + constructor(readonly child: ChildProcess) { + let pending = ""; + child.stdout!.on("data", (chunk: Buffer) => { + pending += chunk.toString("utf8"); + for (let at = pending.indexOf("\n"); at >= 0; at = pending.indexOf("\n")) { + const line = pending.slice(0, at).trim(); + pending = pending.slice(at + 1); + if (line) this.facts.push(JSON.parse(line) as Fact); + } + }); + child.stderr!.on("data", (chunk: Buffer) => { this.stderr += chunk.toString("utf8"); }); + } + /** How many times `event` was printed for `recordKey`. */ + count(event: string, recordKey: string): number { + return this.facts.filter((fact) => fact.event === event && fact.recordKey === recordKey).length; + } + async stop(): Promise { + this.child.stdin!.end(); + await new Promise((done) => { + const killer = setTimeout(() => this.child.kill("SIGKILL"), 5_000); + this.child.once("exit", () => { clearTimeout(killer); done(); }); + if (this.child.exitCode !== null) done(); + }); + } +} + +async function launch(command: string, args: string[]): Promise { + const server = new ServerProcess(spawn(command, args, { stdio: ["pipe", "pipe", "pipe"] })); + await waitFor(() => server.facts.some((fact) => fact.event === "listening") || server.child.exitCode !== null, "the server to listen", 60_000); + const listening = server.facts.find((fact) => fact.event === "listening"); + if (!listening) throw new Error(`the server exited before listening:\n${server.stderr}`); + server.port = listening.port as number; + return server; +} + +/** One connection to a server: the peer that speaks CultNet, and how to close it. */ +interface Connection { peer: CultNetPeer; close(): void } + +interface Target { + name: string; + /** Shard fields the server's database requires on a write, if any. */ + shard?: { shardId: string; shardEpoch: number }; + /** A document payload this target's store reads. */ + payload(text: string): Uint8Array; + start(): Promise; + connect(server: ServerProcess): Promise; +} + +let nextConnectionId = 0x20206000; + +const rust: Target = { + name: "rust", + payload: (text) => encode({ text }), + async start() { + if (!prebuilt) await execFileAsync(cargo, ["build", "--example", "receipt_interop_server", "--message-format", "short"], { cwd: rustManifestDir }); + const metadata = spawnSync(cargo, ["metadata", "--format-version", "1", "--no-deps"], { cwd: rustManifestDir, encoding: "utf8" }); + if (metadata.status !== 0) throw new Error(`cargo metadata failed: ${metadata.stderr}`); + const exe = join(JSON.parse(metadata.stdout).target_directory, "debug", "examples", `receipt_interop_server${process.platform === "win32" ? ".exe" : ""}`); + return launch(exe, []); + }, + async connect(server) { + const peer = await CultMesh.createRudpPeer("receipt-interop", nextConnectionId++, { host: "127.0.0.1", port: server.port, uri: `rudp://127.0.0.1:${server.port}` }); + return { peer, close: () => peer.close() }; + }, +}; + +const csharp: Target = { + name: "csharp", + shard: { shardId: "receipt-interop-shard", shardEpoch: 1 }, + payload: (text) => encode([text]), + async start() { + if (!prebuilt) await execFileAsync(dotnet, ["build", csharpProject, "-nologo", "-v", "q"], { cwd: repoRoot, maxBuffer: 64 * 1024 * 1024 }); + return launch(dotnet, [csharpDll, "receipt-serve"]); + }, + async connect(server) { + const relay = spawn(dotnet, [csharpDll, "receipt-relay", "--port", String(server.port)], { stdio: ["pipe", "pipe", "ignore"] }); + const peer = new CultNetPeer(Duplex.from({ readable: relay.stdout!, writable: relay.stdin! }), { wireContract: "cultnet.schema.v0" }); + return { peer, close: () => { relay.stdin!.end(); relay.kill(); } }; + }, +}; + +// ---- a client session: one connection, the frames it saw, one WriteTracker per write ----------------------------------------- + +class Session { + readonly frames: CultNetMessage[] = []; + readonly invalid: Error[] = []; + private readonly trackers = new Set(); + private serverReceipts = false; + private constructor(readonly target: Target, private readonly connection: Connection) { + connection.peer.on("message", (message: CultNetMessage) => { + if (message.schemaVersion === "cultnet.hello.v0") this.serverReceipts = helloAdvertisesWriteReceipts(message); + this.frames.push(message); + for (const tracker of this.trackers) tracker.frame(message, Date.now()); + }); + connection.peer.on("invalidMessage", (error: Error) => this.invalid.push(error)); + connection.peer.on("close", () => { for (const tracker of this.trackers) tracker.sessionEnd(Date.now()); }); + } + + static async open(target: Target, server: ServerProcess): Promise { + return new Session(target, await target.connect(server)); + } + + /** Declares the receipt capability and waits for the server's own hello. */ + async hello(versions: readonly string[] = CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS): Promise { + this.connection.peer.sendHello({ schemaVersion: "cultnet.hello.v0", runtimeId: "receipt-interop", runtimeKind: "cultmesh-ts", supportedMessageVersions: [...versions] }); + await waitFor(() => this.frames.some((frame) => frame.schemaVersion === "cultnet.hello.v0"), "the server's hello"); + return this.frames.find((frame) => frame.schemaVersion === "cultnet.hello.v0") as CultNetHelloMessage; + } + + put(messageId: string, recordKey: string, text: string, document: Record = {}): CultNetMessage { + return this.shard({ + schemaVersion: "cultnet.document_put_raw.v0", + messageId, + document: { + schemaId: SCHEMA_ID, recordKey, storedAt: new Date().toISOString(), payloadEncoding: "messagepack", + payload: this.target.payload(text), ...document, + }, + }); + } + + delete(messageId: string, recordKey: string): CultNetMessage { + return this.shard({ schemaVersion: "cultnet.document_delete.v0", messageId, schemaId: SCHEMA_ID, recordKey }); + } + + private shard(message: unknown): CultNetMessage { + return { ...(message as object), ...this.target.shard } as CultNetMessage; + } + + /** + * Sends one write and settles it with its own tracker, windowed by what the server advertised (5000 ms against a receipt + * server, 500 ms otherwise). Only the tracker decides the outcome. + */ + async write(message: CultNetMessage): Promise { + const messageId = (message as { messageId: string }).messageId; + const tracker = new WriteTracker({ windowMs: resultWindowMs(this.serverReceipts) }); + this.trackers.add(tracker); + tracker.send(messageId, Date.now()); + this.connection.peer.send(message); + for (;;) { + tracker.tick(Date.now()); + const state = tracker.state(messageId)!; + if (state.outcome !== "pending") return state; + await sleep(Math.min(20, tracker.nextDeadlineMs(Date.now()) ?? 20)); + } + } + + /** Frames of `schemaVersion` that name `messageId`. */ + answers(messageId: string, schemaVersion: string): CultNetMessage[] { + return this.frames.filter((frame) => frame.schemaVersion === schemaVersion && (frame as { messageId?: string }).messageId === messageId); + } + + close(): void { this.connection.close(); } +} + +function refusal(settlement: WriteSettlement): CultNetErrorV1Message { + assert.equal(settlement.outcome, "refused", JSON.stringify(settlement)); + return (settlement as Extract).response as CultNetErrorV1Message; +} + +// ---- the scenarios, run against each server ----------------------------------------------------------------------------------- + +for (const target of [rust, csharp]) { + describe(`write receipt, TypeScript client against the real ${target.name} server`, () => { + let server: ServerProcess; + let session: Session | undefined; + before(async () => { server = await target.start(); }, { timeout: 180_000 }); + after(async () => { session?.close(); await server?.stop(); }); + + /** A fresh receipt session (hello sent, the server's hello read) for a scenario. */ + async function receiptSession(): Promise { + session?.close(); + session = await Session.open(target, server); + const hello = await session.hello(); + assert.ok(helloAdvertisesWriteReceipts(hello), "the server's hello lists both receipt frames"); + return session; + } + + test("hello listing both frames: the server answers with its own hello listing both, then a put is receipted after it", { timeout: 30_000 }, async () => { + const s = await receiptSession(); + assert.equal(s.frames[0].schemaVersion, "cultnet.hello.v0"); + const settlement = await s.write(s.put("hello-1", "hello:a", "one")); + assert.deepEqual(settlement, { outcome: "admitted" }); + assert.deepEqual(s.frames.map((frame) => frame.schemaVersion), ["cultnet.hello.v0", "cultnet.write_admitted.v0"]); + }); + + test("a hello listing only one of the two frames does not opt in: no receipt, and the put is applied", { timeout: 30_000 }, async () => { + for (const listed of CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS) { + session?.close(); + session = await Session.open(target, server); + session.hello([listed]).catch(() => undefined); // a server may stay silent to a hello that does not opt in + await sleep(300); + const settlement = await session.write(session.put(`half-${listed}`, `half:${listed}`, "one")); + assert.deepEqual(settlement, { outcome: "unconfirmed", reason: "window_elapsed" }, `listing only ${listed}`); + assert.ok(!session.frames.some((frame) => frame.schemaVersion === "cultnet.write_admitted.v0" || frame.schemaVersion === "cultnet.error.v1")); + await waitFor(() => server.count("applied", `half:${listed}`) === 1, "the put to be applied"); + } + }); + + test("put admitted: the receipt echoes the write and the store has it once", { timeout: 30_000 }, async () => { + const s = await receiptSession(); + const settlement = await s.write(s.put("put-1", "put:a", "one")); + assert.deepEqual(settlement, { outcome: "admitted" }); + assert.deepEqual(s.answers("put-1", "cultnet.write_admitted.v0"), [ + { schemaVersion: "cultnet.write_admitted.v0", messageId: "put-1", operation: "put", schemaId: SCHEMA_ID, recordKey: "put:a" }, + ]); + await waitFor(() => server.count("applied", "put:a") === 1, "the put to be applied"); + }); + + // The Rust RUDP document server has no delete: its sink only accepts documents, and every delete on a receipt session is + // answered with a refusal (rudp_document_server.rs, deletes_on_a_receipt_session_are_refused_one_by_one). The spec + // (Server rules 1 to 3) has a delete admitted and receipted like a put. + test("delete admitted: the receipt says operation delete and the store drops the record", { + timeout: 30_000, + todo: target.name === "rust" ? "the Rust RUDP document server refuses every delete; there is no delete port in CultMeshRudpRawDocumentSink" : false, + }, async () => { + const s = await receiptSession(); + assert.deepEqual(await s.write(s.put("del-put", "del:a", "one")), { outcome: "admitted" }); + assert.deepEqual(await s.write(s.delete("del-1", "del:a")), { outcome: "admitted" }); + assert.deepEqual(s.answers("del-1", "cultnet.write_admitted.v0"), [ + { schemaVersion: "cultnet.write_admitted.v0", messageId: "del-1", operation: "delete", schemaId: SCHEMA_ID, recordKey: "del:a" }, + ]); + await waitFor(() => server.count("removed", "del:a") === 1, "the delete to be applied"); + }); + + test("refused write: an error.v1 carrying the write's id, and nothing is applied", { timeout: 30_000 }, async () => { + const s = await receiptSession(); + const response = refusal(await s.write(s.put("deny-1", "refuse:a", "one"))); + assert.equal(response.schemaVersion, "cultnet.error.v1"); + assert.equal(response.messageId, "deny-1"); + assert.equal(response.code, "write_not_authorized"); + // The session is kept: the next write on it is admitted. + assert.deepEqual(await s.write(s.put("deny-2", "deny:b", "one")), { outcome: "admitted" }); + assert.equal(server.count("applied", "refuse:a"), 0); + }); + + test("resend of the same id and content on the same session: re-receipted, applied once", { timeout: 30_000 }, async () => { + const s = await receiptSession(); + assert.deepEqual(await s.write(s.put("resend-1", "resend:a", "one")), { outcome: "admitted" }); + assert.deepEqual(await s.write(s.put("resend-1", "resend:a", "one")), { outcome: "admitted" }); + const receipts = s.answers("resend-1", "cultnet.write_admitted.v0"); + assert.equal(receipts.length, 2); + assert.deepEqual(receipts[0], receipts[1]); + await sleep(200); // a second application would be printed by now + assert.equal(server.count("applied", "resend:a"), 1); + }); + + test("same id with different content: message_invalid naming the id, never a receipt, the first write untouched", { timeout: 30_000 }, async () => { + const s = await receiptSession(); + assert.deepEqual(await s.write(s.put("clash-1", "clash:a", "one")), { outcome: "admitted" }); + const response = refusal(await s.write(s.put("clash-1", "clash:a", "other"))); + assert.equal(response.code, "message_invalid"); + assert.equal(response.messageId, "clash-1"); + assert.equal(response.details?.field, "messageId"); + assert.equal(s.answers("clash-1", "cultnet.write_admitted.v0").length, 1, "no receipt for the changed write"); + await sleep(200); + assert.equal(server.count("applied", "clash:a"), 1); + }); + + // Each case of the vectors' blankFields section: a document that names nothing is refused as an unreadable frame, with the + // write's id and the fixed text, and never applied; a document that is merely missing optional fields is admitted. + for (const blank of vectors.blankFields.cases as { name: string; set: Record; outcome: "admitted" | "refused" }[]) { + test(`blank-field vector ${blank.name}: ${blank.outcome}`, { timeout: 30_000 }, async () => { + const s = await receiptSession(); + const recordKey = `blank:${blank.name}`; + const { payloadHex, ...fields } = blank.set; + const message = s.put(`blank-${blank.name}`, recordKey, "one", fields); + const sent = (message as unknown as { document: Record }).document; + for (const [key, value] of Object.entries(fields)) if (value === null) delete sent[key]; // nil means absent on the wire here + if (payloadHex === "") sent.payload = new Uint8Array(); + const settlement = await s.write(message); + if (blank.outcome === "admitted") { + assert.deepEqual(settlement, { outcome: "admitted" }); + await waitFor(() => server.count("applied", recordKey) === 1, "the put to be applied"); + return; + } + const response = refusal(settlement); + assert.equal(response.code, "message_invalid"); + assert.equal(response.error, UNREADABLE); + assert.equal(response.details?.field, "message"); + assert.equal(response.messageId, `blank-${blank.name}`); + await sleep(200); + assert.equal(server.count("applied", recordKey), 0); + assert.equal(server.count("applied", "blank:"), 0); + }); + } + + test("a delete that names nothing is refused as unreadable with its id", { timeout: 30_000 }, async () => { + const s = await receiptSession(); + const message = s.delete("blank-del", "blank:del"); + (message as unknown as { schemaId: string }).schemaId = ""; + const response = refusal(await s.write(message)); + assert.equal(response.code, "message_invalid"); + assert.equal(response.error, UNREADABLE); + assert.equal(response.details?.field, "message"); + assert.equal(response.messageId, "blank-del"); + }); + + test("legacy client (no hello): no receipt frames, the write is applied, and its refusal is the old error.v0", { timeout: 30_000 }, async () => { + session?.close(); + session = await Session.open(target, server); + assert.deepEqual(await session.write(session.put("legacy-1", "legacy:a", "one")), { outcome: "unconfirmed", reason: "window_elapsed" }); + await waitFor(() => server.count("applied", "legacy:a") === 1, "the put to be applied"); + const refused = (await session.write(session.put("legacy-2", "refuse:legacy", "one"))) as Extract; + assert.equal(refused.outcome, "refused"); + assert.equal(refused.response.schemaVersion, "cultnet.error.v0"); + assert.deepEqual(session.frames.map((frame) => frame.schemaVersion), ["cultnet.error.v0"]); + assert.equal(session.invalid.length, 0); + }); + }); +} + +// ---- the one-shot writer, which is the real cultmesh-ts client path, against the Rust RUDP server ---------------------------- +// (The C# server speaks LiteNetLib, not RUDP: no RUDP client path reaches it.) + +describe("write receipt, cultmesh-ts publishRudpDocumentOnce against the real rust server", () => { + const note = defineDocumentType({ type: "receipt.interop_note", schemaId: SCHEMA_ID, schema: z.object({ text: z.string() }), name: "text" }); + const binding = defineCultNetDocumentBinding({ definition: note }); + let server: ServerProcess; + before(async () => { server = await rust.start(); }, { timeout: 180_000 }); + after(async () => { await server?.stop(); }); + const endpoint = () => ({ host: "127.0.0.1", port: server.port, uri: `rudp://127.0.0.1:${server.port}` }); + + test("writeReceipts: the put resolves admitted on the receipt, after the server applied it", { timeout: 30_000 }, async () => { + const outcome = await CultMesh.publishRudpDocumentOnce("receipt-interop", nextConnectionId++, endpoint(), binding, "once:a", { text: "one" }, { writeReceipts: true, messageId: "once-1" }); + assert.equal(outcome, "admitted"); + await waitFor(() => server.count("applied", "once:a") === 1, "the put to be applied"); + }); + + test("writeReceipts: a refused put rejects with the server's error.v1 naming the put", { timeout: 30_000 }, async () => { + await assert.rejects( + CultMesh.publishRudpDocumentOnce("receipt-interop", nextConnectionId++, endpoint(), binding, "refuse:once", { text: "one" }, { writeReceipts: true, messageId: "once-2" }), + (error: unknown) => { + assert.ok(error instanceof CultMeshRudpWriteRefusedError); + assert.equal(error.response.schemaVersion, "cultnet.error.v1"); + assert.equal((error.response as CultNetErrorV1Message).messageId, "once-2"); + return true; + }, + ); + }); + + test("without writeReceipts the server stays silent: the put is unconfirmed, not admitted, though the server applied it", { timeout: 30_000 }, async () => { + await assert.rejects( + CultMesh.publishRudpDocumentOnce("receipt-interop", nextConnectionId++, endpoint(), binding, "once:legacy", { text: "one" }, { messageId: "once-3" }), + (error: unknown) => error instanceof CultMeshRudpWriteUnconfirmedError && error.reason === "window_elapsed", + ); + await waitFor(() => server.count("applied", "once:legacy") === 1, "the put to be applied"); + }); +}); -- 2.47.3 From 04207007d6b51d3be27b547300e1e2aac1718244 Mon Sep 17 00:00:00 2001 From: Libby Date: Mon, 9 Nov 2026 07:45:35 +0000 Subject: [PATCH 4/7] interop test: read the store under the record key the blank-field vector set A vector that sets recordKey (zero_width_space_record_key_is_a_name) replaces the harness key, so the admitted case waited for blank: and timed out against the Rust server, and the refused record-key cases counted applications under a key the server never saw. Use the key the document carried. Co-Authored-By: Claude Sonnet 5.5 --- .../cultmesh-ts/test/interop/write-receipt-interop.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts b/packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts index 8a646a50..12bd4394 100644 --- a/packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts +++ b/packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts @@ -331,8 +331,9 @@ for (const target of [rust, csharp]) { for (const blank of vectors.blankFields.cases as { name: string; set: Record; outcome: "admitted" | "refused" }[]) { test(`blank-field vector ${blank.name}: ${blank.outcome}`, { timeout: 30_000 }, async () => { const s = await receiptSession(); - const recordKey = `blank:${blank.name}`; const { payloadHex, ...fields } = blank.set; + // A case that sets recordKey replaces the harness key, so the store is read under the key the server actually saw. + const recordKey = typeof fields.recordKey === "string" ? fields.recordKey : `blank:${blank.name}`; const message = s.put(`blank-${blank.name}`, recordKey, "one", fields); const sent = (message as unknown as { document: Record }).document; for (const [key, value] of Object.entries(fields)) if (value === null) delete sent[key]; // nil means absent on the wire here @@ -350,7 +351,6 @@ for (const target of [rust, csharp]) { assert.equal(response.messageId, `blank-${blank.name}`); await sleep(200); assert.equal(server.count("applied", recordKey), 0); - assert.equal(server.count("applied", "blank:"), 0); }); } -- 2.47.3 From 1704fb5060f392395c9df55522bcf45f943c2f6e Mon Sep 17 00:00:00 2001 From: soul Date: Mon, 9 Nov 2026 14:05:57 +0000 Subject: [PATCH 5/7] cultmesh-ts: test:interop builds cultcache-ts and cultnet-ts (with contracts) first Co-Authored-By: Claude Sonnet 5.5 --- packages/cultmesh-ts/package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/cultmesh-ts/package.json b/packages/cultmesh-ts/package.json index 0fef7419..39e0abba 100644 --- a/packages/cultmesh-ts/package.json +++ b/packages/cultmesh-ts/package.json @@ -25,7 +25,7 @@ "prepack": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json", "typecheck": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json --pretty false", "test": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json && node ../../node_modules/typescript/bin/tsc -p tsconfig.test.json && node --expose-gc --test dist-test/test/*.test.js", - "test:interop": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json && node ../../node_modules/typescript/bin/tsc -p tsconfig.test.json && node --test --test-force-exit dist-test/test/interop/write-receipt-interop.test.js" + "test:interop": "npm --prefix ../.. run build --workspace @gamecult/cultcache-ts --workspace @gamecult/cultnet-ts && node ../../node_modules/typescript/bin/tsc -p tsconfig.json && node ../../node_modules/typescript/bin/tsc -p tsconfig.test.json && node --test --test-force-exit dist-test/test/interop/write-receipt-interop.test.js" }, "keywords": [ "cultmesh", -- 2.47.3 From b8e40dc372660774535f021c39a70aa7b968679e Mon Sep 17 00:00:00 2001 From: soul Date: Mon, 9 Nov 2026 14:10:42 +0000 Subject: [PATCH 6/7] interop test: pace C# relay connections under the server's per-IP per-minute limit Co-Authored-By: Claude Sonnet 5.5 --- .../test/interop/write-receipt-interop.test.ts | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts b/packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts index 12bd4394..092d5ffb 100644 --- a/packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts +++ b/packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts @@ -16,7 +16,7 @@ import { execFile, spawn, spawnSync, type ChildProcess } from "node:child_proces import { readFileSync } from "node:fs"; import { join, resolve } from "node:path"; import { Duplex } from "node:stream"; -import { after, before, describe, test } from "node:test"; +import { after, before, beforeEach, describe, test } from "node:test"; import { promisify } from "node:util"; import { encode } from "@msgpack/msgpack"; import { z } from "zod"; @@ -134,6 +134,20 @@ const rust: Target = { }, }; +// The C# server admits 30 connection requests per IP per minute (Server.MaxConnectionAttemptsPerMinute) and answers the rest with +// no hello. Every scenario wants a fresh session, so before each C# test the harness waits until the last minute holds at most +// CONNECTIONS_PER_MINUTE connections, leaving room for the two a test may open. The server's limit stays as it is. +const CONNECTIONS_PER_MINUTE = 24; +const connectedAt: number[] = []; +async function awaitConnectionBudget(): Promise { + for (;;) { + const now = Date.now(); + while (connectedAt.length > 0 && now - connectedAt[0] >= 61_000) connectedAt.shift(); + if (connectedAt.length <= CONNECTIONS_PER_MINUTE) return; + await sleep(connectedAt[0] + 61_000 - now); + } +} + const csharp: Target = { name: "csharp", shard: { shardId: "receipt-interop-shard", shardEpoch: 1 }, @@ -143,6 +157,7 @@ const csharp: Target = { return launch(dotnet, [csharpDll, "receipt-serve"]); }, async connect(server) { + connectedAt.push(Date.now()); const relay = spawn(dotnet, [csharpDll, "receipt-relay", "--port", String(server.port)], { stdio: ["pipe", "pipe", "ignore"] }); const peer = new CultNetPeer(Duplex.from({ readable: relay.stdout!, writable: relay.stdin! }), { wireContract: "cultnet.schema.v0" }); return { peer, close: () => { relay.stdin!.end(); relay.kill(); } }; @@ -235,6 +250,7 @@ for (const target of [rust, csharp]) { let session: Session | undefined; before(async () => { server = await target.start(); }, { timeout: 180_000 }); after(async () => { session?.close(); await server?.stop(); }); + beforeEach(async () => { if (target === csharp) await awaitConnectionBudget(); }, { timeout: 90_000 }); /** A fresh receipt session (hello sent, the server's hello read) for a scenario. */ async function receiptSession(): Promise { -- 2.47.3 From 96c5fb1ca4dfa1e4523201573af49cb108082a00 Mon Sep 17 00:00:00 2001 From: soul Date: Mon, 9 Nov 2026 14:12:38 +0000 Subject: [PATCH 7/7] interop peer: C# store holds the spec's admitted schema-id names (compatible ids) Co-Authored-By: Claude Sonnet 5.5 --- tests/GameCult.Networking.InteropPeer/ReceiptHarness.cs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/GameCult.Networking.InteropPeer/ReceiptHarness.cs b/tests/GameCult.Networking.InteropPeer/ReceiptHarness.cs index 75eaf4da..6871ea00 100644 --- a/tests/GameCult.Networking.InteropPeer/ReceiptHarness.cs +++ b/tests/GameCult.Networking.InteropPeer/ReceiptHarness.cs @@ -14,7 +14,9 @@ using R3; // ordered payload, and on the pipe it is a 4-byte big-endian length followed by the frame. namespace GameCult.Networking.InteropPeer { - [CultDocument("receipt.interop_note", ReceiptHarness.SchemaId)] + // CompatibleSchemaIds are the names the spec's blankFields vectors admit (U+200B, U+001C, and a name padded with NBSP and EM SPACE): + // a name is any string that is not blank, so the store must hold a record put under one of them. + [CultDocument("receipt.interop_note", ReceiptHarness.SchemaId, CompatibleSchemaIds = ["\u200b", "\u001c", "\u00a0a\u2003"])] [MessagePack.MessagePackObject] public sealed class ReceiptInteropNote { -- 2.47.3