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

TS write-receipt client against the real Rust and C# receipt servers (interop harness) #67

Merged
libby merged 9 commits from libby/write-receipt-ts-interop into main 2026-11-09 14:16:00 +00:00
Conversation 0 Commits 9 Files changed 6 +642 -2
6 changed files with 642 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

2
contracts/cultnet/interop/write-receipt.md
View file

@ -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.

77
packages/cultmesh-rs/examples/receipt_interop_server.rs Normal file
View file

@ -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<Vec<CultNetRawDocumentRecord>> { 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(())
}

3
packages/cultmesh-ts/package.json
View file

@ -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": "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",

434
packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts Normal file
View file

@ -0,0 +1,434 @@
// 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, beforeEach, 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<void>((done) => setTimeout(done, ms));
async function waitFor(predicate: () => boolean, what: string, ms = 5_000): Promise<void> {
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<void> {
this.child.stdin!.end();
await new Promise<void>((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<ServerProcess> {
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<ServerProcess>;
connect(server: ServerProcess): Promise<Connection>;
}
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() };
},
};
// 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<void> {
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 },
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) {
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(); } };
},
};
// ---- 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<WriteTracker>();
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<Session> {
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<CultNetHelloMessage> {
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<string, unknown> = {}): 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<WriteSettlement> {
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<WriteSettlement, { outcome: "refused" }>).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(); });
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<Session> {
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<string, unknown>; outcome: "admitted" | "refused" }[]) {
test(`blank-field vector ${blank.name}: ${blank.outcome}`, { timeout: 30_000 }, async () => {
const s = await receiptSession();
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<string, unknown> }).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);
});
}
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<WriteSettlement, { outcome: "refused" }>;
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");
});
});

8
tests/GameCult.Networking.InteropPeer/Program.cs
View file

@ -25,7 +25,7 @@ static async Task<int> 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<int> 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;

120
tests/GameCult.Networking.InteropPeer/ReceiptHarness.cs Normal file
View file

@ -0,0 +1,120 @@
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
{
// 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
{
[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<ReceiptInteropNote>().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<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
var ended = new TaskCompletionSource<bool>(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;
}
}
}