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

cultmesh-ts: a snapshot refusal carries fixed text and acknowledges nothing

Browse source 
The snapshot fallback sent the caught error's message to the peer, and
errors from getCache or the registry went the same way, so a message the
application or a peer supplied could reach the wire. Its ack fields also
covered the refused request. The snapshot path now refuses per case with
fixed text (the snapshot source failed / the snapshot response is too
large / the snapshot response could not be queued) through refuse(),
which sends unreliable and unordered, so it needs no room in a full
queue, with ack fields that acknowledge nothing, as every runtime's
refusal does. onError keeps the detail.

Tests, with a raw peer that acknowledges nothing: a queue of 8 filled
exactly by an unacknowledged 8-fragment reply, then a refusal that still
arrives and acknowledges nothing; and a getCache failure carrying a
canary, refused with "the snapshot source failed".

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 23:06:03 +02:00
commit 8b8f8436cc
2 changed files with 160 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

46
packages/cultmesh-ts/src/index.ts
View file

@ -5965,17 +5965,25 @@ export class CultMesh {
});
return;
}
let reply: CultNetMessage;
try {
const cache = await options.getCache();
sendSchemaMessage(record, options.documents.createRawSnapshotResponse(cache, message.messageId, message));
reply = options.documents.createRawSnapshotResponse(cache, message.messageId, message);
} catch (error) {
reportError(error);
// Unreliable and unordered: when the reply could not be queued, the answer must not
// need room in the queue that just refused it.
sendSchemaMessage(record, {
schemaVersion: "cultnet.error.v0",
error: error instanceof Error ? error.message : String(error),
}, { reliable: false });
refuse(record, "the snapshot source failed");
return;
}
try {
sendSchemaMessage(record, reply);
} catch (error) {
reportError(error);
refuse(
record,
maxPayloadBytes !== undefined && encodeSchemaMessage(reply).byteLength > maxPayloadBytes
? "the snapshot response is too large"
: "the snapshot response could not be queued",
);
}
return;
}
@ -6088,7 +6096,6 @@ export class CultMesh {
function sendSchemaMessage(
record: SessionRecord,
message: CultNetMessage,
{ reliable = true }: { reliable?: boolean } = {},
): void {
if (record.closed) throw new Error(`CultMesh RUDP session ${record.sessionId} is closed.`);
const payload = encodeSchemaMessage(message);
@ -6099,8 +6106,8 @@ export class CultMesh {
);
}
for (const packet of record.session.sendMany("schema", payload, {
reliable,
ordered: reliable,
reliable: true,
ordered: true,
nowMs: Date.now(),
maxFragmentBytes,
})) {
@ -6108,6 +6115,25 @@ export class CultMesh {
}
}
/**
* Answers a request the server cannot serve with fixed text, never an error's message,
* which can quote what a peer or the application supplied. The answer is unreliable and
* unordered, so it needs no room in a reliable queue that may just have refused the reply,
* and its acknowledgement fields acknowledge nothing, as every runtime's refusal does.
*/
function refuse(record: SessionRecord, text: string): void {
if (record.closed) return;
const payload = encodeSchemaMessage({ schemaVersion: "cultnet.error.v0", error: text });
for (const packet of record.session.sendMany("schema", payload, {
reliable: false,
ordered: false,
nowMs: Date.now(),
maxFragmentBytes,
})) {
sendPacket(record, { ...packet, ack: 0, ackMask: 0 });
}
}
/** The one reply encoder, shared by sends and by put admission. */
function encodeSchemaMessage(message: CultNetMessage): Uint8Array {
return encode(encodeCultNetMessageForWire(message, wireContract));

124
packages/cultmesh-ts/test/cultmesh.test.ts
View file

@ -1,3 +1,4 @@
import { createSocket } from "node:dgram";
import { mkdtemp } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
@ -9,9 +10,13 @@ import { defineDocumentType } from "@gamecult/cultcache-ts";
import {
CultNetDocumentRegistry,
CultNetPeer,
CultNetRudpSession,
cultNetBuiltinSchemaRegistry,
decodeRudpPacket,
defineCultNetDocumentBinding,
encodeRudpPacket,
type CultNetDocumentPutRawMessage,
type CultNetRudpPacket,
type CultNetMessage,
type CultNetSnapshotResponseRawMessage,
} from "cultnet-ts";
@ -3512,6 +3517,125 @@ test("CultMesh TS RUDP document server answers a snapshot its reliable queue can
});
});
/**
* A document server over a node that already holds `stored`, and a peer driven packet by packet
* that acknowledges nothing, so a test fills the server's reliable queue at will.
*/
async function withRawPeerRig(
connectionId: number,
serverLimits: { maxFragmentBytes?: number; maxPendingReliablePackets?: number },
stored: CultNetDocumentPutRawMessage[],
getCache: ((cache: unknown) => unknown) | undefined,
body: (rig: {
errors: Error[];
inbox: CultNetRudpPacket[];
request: (messageId: string, recordKeys: string[]) => number[];
}) => Promise<void>,
): Promise<void> {
const node = await CultMesh.startNode(
join(await mkdtemp(join(tmpdir(), "cultmesh-ts-raw-peer-")), "node.ccmp"),
{ documents: [noteDocument] },
);
const registry = servedBoundRegistry();
for (const put of stored) await registry.applyRawDocumentPutMessage(node.cache, put);
const errors: Error[] = [];
const server = CultMesh.createRudpDocumentServer(SERVED_BOUND_SERVER, connectionId, {
documents: registry,
getCache: () => (getCache ? getCache(node.cache) : node.cache) as typeof node.cache,
bindHost: "127.0.0.1",
bindPort: 0,
resendDelayMs: 100_000,
resendPollMs: 5,
...serverLimits,
onError: (error) => errors.push(error),
});
await server.start();
const socket = createSocket("udp4");
try {
await new Promise<void>((resolve) => socket.bind(0, "127.0.0.1", () => resolve()));
const inbox: CultNetRudpPacket[] = [];
socket.on("message", (wire) => inbox.push(decodeRudpPacket(wire)));
const session = new CultNetRudpSession({ connectionId, initialSequence: 100, resendDelayMs: 100_000, maxPendingReliablePackets: 64 });
const send = (packet: CultNetRudpPacket) => socket.send(encodeRudpPacket(packet), server.bind.port, "127.0.0.1");
send(session.createConnect(0));
const startedAt = Date.now();
while (!inbox.some((packet) => packet.packetType === "accept") && Date.now() - startedAt < 2_000) await delay(5);
const accept = inbox.find((packet) => packet.packetType === "accept");
assert.ok(accept, "the server accepts the connect");
session.receive(accept, 1);
inbox.length = 0;
const request = (messageId: string, recordKeys: string[]): number[] => {
const packets = session.sendMany(
"schema",
encode({ schemaVersion: "cultnet.snapshot_request.v0", messageId, recordKeys }),
{ reliable: true, ordered: true, nowMs: 2, maxFragmentBytes: 1200 },
);
packets.forEach(send);
return packets.map((packet) => packet.sequence);
};
await body({ errors, inbox, request });
} finally {
socket.close();
server.close();
}
}
/** The refusal among `packets`: its unreliable fragments, reassembled, and the fragments. */
function refusalIn(packets: CultNetRudpPacket[]): { message: Record<string, unknown>; parts: CultNetRudpPacket[] } | undefined {
const parts = packets
.filter((packet) => packet.packetType === "data" && packet.channelId === "schema" && !packet.reliable)
.sort((a, b) => (a.fragmentIndex ?? 0) - (b.fragmentIndex ?? 0));
if (parts.length === 0) return undefined;
const message = decode(Buffer.concat(parts.map((part) => Buffer.from(part.payload ?? new Uint8Array())))) as Record<string, unknown>;
return { message, parts };
}
function acknowledges(packet: CultNetRudpPacket, sequence: number): boolean {
if (packet.ack === sequence) return true;
for (let bit = 0; bit < 32; bit += 1) {
if ((packet.ackMask & (1 << bit)) !== 0 && packet.ack - bit - 1 === sequence) return true;
}
return false;
}
test("CultMesh TS RUDP document server answers a peer whose reliable queue is full to the brim", async () => {
// An 8-fragment reply the peer never acknowledges fills a queue of 8 exactly. The next reply
// cannot be queued; the peer is still answered, with fixed text that takes no room in that
// queue and acknowledges nothing the peer sent.
const body = bodyLengthForReplyBytes("q1", 800);
await withRawPeerRig(0x10203069, { maxFragmentBytes: 100, maxPendingReliablePackets: 8 }, [notePut("s", "q1", body)], undefined, async (rig) => {
rig.request("0", ["q1"]);
await delay(300);
const first = new Set(rig.inbox.filter((packet) => packet.packetType === "data" && packet.reliable).map((packet) => packet.sequence));
assert.equal(first.size, 8, "fixture: the first reply fills the queue exactly");
rig.inbox.length = 0;
const refused = rig.request("b", ["none"]);
const startedAt = Date.now();
while (!refusalIn(rig.inbox) && Date.now() - startedAt < 3_000) await delay(5);
const refusal = refusalIn(rig.inbox);
assert.ok(refusal, "the refusal reaches the peer");
assert.deepEqual(refusal.message, { schemaVersion: "cultnet.error.v0", error: "the snapshot response could not be queued" });
for (const part of refusal.parts) {
assert.ok(refused.every((sequence) => !acknowledges(part, sequence)), `ack ${part.ack} mask ${part.ackMask}`);
}
assert.equal(rig.errors.length, 1);
});
});
test("CultMesh TS RUDP document server refuses a snapshot with fixed text, never an error's message", async () => {
// The cache the snapshot needs fails with a message a peer must not see: the peer is told
// only that the snapshot source failed; onError keeps the detail.
await withRawPeerRig(0x1020306a, {}, [], () => { throw new Error("no such tenant CANARY-7f3a"); }, async (rig) => {
rig.request("0", ["q1"]);
const startedAt = Date.now();
while (!refusalIn(rig.inbox) && Date.now() - startedAt < 3_000) await delay(5);
const refusal = refusalIn(rig.inbox);
assert.ok(refusal, "the refusal reaches the peer");
assert.deepEqual(refusal.message, { schemaVersion: "cultnet.error.v0", error: "the snapshot source failed" });
assert.match(rig.errors[0]?.message ?? "", /CANARY-7f3a/);
});
});
test("CultMesh TS RUDP document server honours maxPayloadBytes for replies and judges puts by what it would serve", async () => {
const limit = 4_000;
const atBound = bodyLengthForReplyBytes("note:at", limit);