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 put is refused unless its receipt could be sent

Browse source 
Admission now sizes two replies built from the record the registry
would store: the snapshot reply serving it alone and the receipt
echoing its value under the put's message id with the server's default
provenance, through the one receipt builder the send path uses. A put
whose larger reply exceeds maxPayloadBytes, 65535 fragments or the
reliable queue is refused before onDocumentPutRaw, so a stored put
always gets its receipt when the application stores through the
registry and echoes that value. The existing fallback (a receipt that
cannot be sent all the same reaches onError, and the peer a
cultnet.error.v0) stays for handlers that acknowledge something larger.

The served-bound fixtures carry a long storedAt, which the served record
keeps and the receipt replaces, so they still pin the served bound; a
new test pins the receipt bound: one byte over is refused, exactly at
the limit is stored and receipted.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 16:04:01 +02:00
commit 9757ed54a5
2 changed files with 142 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -4289,19 +4289,22 @@ export interface CultMeshRudpDocumentReceipt<TDefinition extends AnyCultCacheDoc
* reply larger than `maxPayloadBytes`. A put is judged by what serving it
* would send, not by the size of its own frame: a put whose snapshot reply
* could never be sent within those limits, 65535 fragments and
* `maxPendingReliablePackets`, is refused before `onDocumentPutRaw` sees it:
* `onError` receives a `CultMeshRudpUnservableDocumentError` and the peer a
* `cultnet.error.v0`.
* `maxPendingReliablePackets`, or whose receipt could not be, is refused
* before `onDocumentPutRaw` sees it: `onError` receives a
* `CultMeshRudpUnservableDocumentError` and the peer a `cultnet.error.v0`.
*
* That reply is sized as `documents` would store and serve the put's record.
* The judgement is exact only for an application whose `onDocumentPutRaw`
* stores the put through that registry (`applyRawDocumentPutMessage`) and
* whose `getCache` serves that store; a handler that stores another record
* is judged by the registry's record, not its own.
* Both replies are sized from the record `documents` would store for the put:
* the snapshot reply serving it alone, and the receipt echoing its value back
* under the put's message id with the server's default provenance. The
* judgement is exact only for an application whose `onDocumentPutRaw` stores
* the put through that registry (`applyRawDocumentPutMessage`), whose
* `getCache` serves that store, and whose receipt, if any, echoes the stored
* value; a handler that stores or acknowledges another record is judged by
* the registry's record, not its own.
*
* A reply that cannot be sent, such as a receipt from `onDocumentPutRaw`
* larger than `maxPayloadBytes`, reaches `onError`, and the peer receives a
* `cultnet.error.v0` in its place. Every failed datagram send reaches `onError`.
* A reply that cannot be sent all the same, such as a larger receipt from such
* a handler, reaches `onError`, and the peer receives a `cultnet.error.v0` in
* its place. Every failed datagram send reaches `onError`.
*/
export interface CultMeshRudpDocumentServerOptions extends CultMeshRudpSocketOptions {
documents: CultNetDocumentRegistry;
@ -4330,7 +4333,8 @@ const SHORTEST_MESSAGE_ID = "0";
/**
* A put the document server refused because it could never serve the document:
* the snapshot reply carrying it alone would exceed the server's reply limits.
* the larger of the snapshot reply carrying it alone and the receipt
* acknowledging it (`responseBytes`) would exceed the server's reply limits.
*/
export class CultMeshRudpUnservableDocumentError extends Error {
constructor(
@ -4344,7 +4348,7 @@ export class CultMeshRudpUnservableDocumentError extends Error {
// The sentence names sizes only: it is sent to the peer, and a record key of
// any length would make the refusal itself too large to send.
super(
`Document can never be served: its snapshot reply is ${responseBytes} bytes in `
`Document can never be served: its reply is ${responseBytes} bytes in `
+ `${fragmentCount} fragments; limits are ${maxPayloadBytes ?? "unbounded"} bytes and `
+ `${maxFragmentCount} fragments.`,
);
@ -5956,15 +5960,7 @@ export class CultMesh {
const receipt = await options.onDocumentPutRaw(normalizeRudpDocumentPut(message, record.remote));
if (receipt) {
try {
sendSchemaMessage(record, options.documents.createRawDocumentPutMessage(
receipt.binding, receipt.messageId ?? message.messageId, receipt.recordKey, receipt.value,
{
sourceRuntimeId: receipt.sourceRuntimeId ?? runtimeId,
sourceAgentId: receipt.sourceAgentId,
sourceRole: receipt.sourceRole ?? "provider",
tags: receipt.tags ?? ["receipt"],
},
));
sendSchemaMessage(record, receiptMessage(receipt, message.messageId));
} catch (error) {
// The put is stored: the peer must hear that its receipt is not coming.
reportError(error);
@ -6069,27 +6065,48 @@ export class CultMesh {
});
}
/** The one receipt builder, shared by the receipt send and by put admission. */
function receiptMessage(receipt: CultMeshRudpDocumentReceipt, putMessageId: string): CultNetMessage {
return options.documents.createRawDocumentPutMessage(
receipt.binding, receipt.messageId ?? putMessageId, receipt.recordKey, receipt.value,
{
sourceRuntimeId: receipt.sourceRuntimeId ?? runtimeId,
sourceAgentId: receipt.sourceAgentId,
sourceRole: receipt.sourceRole ?? "provider",
tags: receipt.tags ?? ["receipt"],
},
);
}
/**
* Refuse a put whose document could never be served: the snapshot reply
* carrying it alone, under the shortest message id CultNet parses, sized
* as `sendSchemaMessage` would send it. The document is sized as the
* registry would store and serve it, which is exact only when the
* application stores through the registry; a document the registry has
* no binding for is sized as received.
* Refuse a put whose document could never be served or acknowledged: the
* snapshot reply carrying it alone, under the shortest message id CultNet
* parses, and the receipt echoing it, each sized as `sendSchemaMessage`
* would send it. Both are built from the record the registry would store,
* which is exact only when the application stores through the registry and
* echoes that value; a document the registry has no binding for is sized
* as received, with no receipt.
*/
function unservablePut(message: CultNetMessage): CultMeshRudpUnservableDocumentError | undefined {
if (message.schemaVersion !== "cultnet.document_put_raw.v0") return undefined;
let reply: CultNetSnapshotResponseRawMessage;
let replies: CultNetMessage[];
try {
reply = options.documents.createRawSnapshotResponseForPut(message, SHORTEST_MESSAGE_ID);
const { binding, value } = options.documents.decodeRawDocumentPut(message);
replies = [
options.documents.createRawSnapshotResponseForPut(message, SHORTEST_MESSAGE_ID),
receiptMessage(
{ binding, recordKey: message.document.recordKey, value } as CultMeshRudpDocumentReceipt,
message.messageId,
),
];
} catch {
reply = {
replies = [{
schemaVersion: "cultnet.snapshot_response_raw.v0",
messageId: SHORTEST_MESSAGE_ID,
documents: [message.document],
};
}];
}
const responseBytes = encodeSchemaMessage(reply).byteLength;
const responseBytes = Math.max(...replies.map((reply) => encodeSchemaMessage(reply).byteLength));
const fragmentCount = Math.max(1, Math.ceil(responseBytes / maxFragmentBytes));
if ((maxPayloadBytes === undefined || responseBytes <= maxPayloadBytes)
&& fragmentCount <= maxReplyFragments) {

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

@ -3243,11 +3243,21 @@ type ServedBoundRig = {
const servedBoundRegistry = () =>
new CultNetDocumentRegistry([defineCultNetDocumentBinding({ definition: noteDocument })]);
/**
* A put's own storedAt, kept by the served record and replaced in the receipt by the server's
* 24-character clock. It is long enough that the snapshot reply is always the larger reply, so
* fixtures built with it pin the served bound; `RECEIPT_STORED_AT` puts pin the receipt bound.
*/
const SERVED_STORED_AT = `2026-09-30T00:00:00.000Z${"~".repeat(200)}`;
const RECEIPT_STORED_AT = "2026-09-30T00:00:00.000Z";
const SERVED_BOUND_SERVER = "cultmesh-ts-served-bound-server";
function notePut(
messageId: string,
recordKey: string,
bodyLength: number,
sourceRuntimeId?: string,
storedAt = SERVED_STORED_AT,
): CultNetDocumentPutRawMessage {
return {
schemaVersion: "cultnet.document_put_raw.v0",
@ -3255,7 +3265,7 @@ function notePut(
document: {
schemaId: "cultmesh.note.v0",
recordKey,
storedAt: "2026-09-30T00:00:00.000Z",
storedAt,
payloadEncoding: "messagepack",
payload: encode({ noteId: recordKey, body: "x".repeat(bodyLength) }),
...(sourceRuntimeId ? { sourceRuntimeId } : {}),
@ -3277,11 +3287,46 @@ function servedAloneBytes(put: CultNetDocumentPutRawMessage): number {
}).byteLength;
}
/** The body length whose served-alone reply is exactly `bytes`, with a guard. */
/**
* The encoded size of the receipt that would acknowledge this put by echoing its value: the
* put's message id, the server's clock and its default provenance. Spelled out here rather than
* asked of the server, so a server that sized another receipt is caught.
*/
function echoReceiptBytes(put: CultNetDocumentPutRawMessage): number {
return encode({
schemaVersion: "cultnet.document_put_raw.v0",
messageId: put.messageId,
document: {
schemaId: "cultmesh.note.v0",
recordKey: put.document.recordKey,
storedAt: new Date().toISOString(),
payloadEncoding: "messagepack",
payload: encode(decode(put.document.payload)),
sourceRuntimeId: SERVED_BOUND_SERVER,
sourceAgentId: undefined,
sourceRole: "provider",
tags: ["receipt"],
},
}).byteLength;
}
/** The body length whose served-alone reply is exactly `bytes`, and the larger reply. */
function bodyLengthForReplyBytes(recordKey: string, bytes: number): number {
const probe = 8_000;
const length = probe + bytes - servedAloneBytes(notePut("p", recordKey, probe));
assert.equal(servedAloneBytes(notePut("p", recordKey, length)), bytes, "fixture: reply size is linear in the body");
const put = notePut("p", recordKey, length);
assert.equal(servedAloneBytes(put), bytes, "fixture: reply size is linear in the body");
assert.ok(echoReceiptBytes(put) < bytes, "fixture: the snapshot reply is the larger reply");
return length;
}
/** The body length whose echoed receipt, under `messageId`, is exactly `bytes`, and the larger reply. */
function bodyLengthForReceiptBytes(messageId: string, recordKey: string, bytes: number): number {
const put = (body: number) => notePut(messageId, recordKey, body, undefined, RECEIPT_STORED_AT);
const probe = 2_000;
const length = probe + bytes - echoReceiptBytes(put(probe));
assert.equal(echoReceiptBytes(put(length)), bytes, "fixture: receipt size is linear in the body");
assert.ok(servedAloneBytes(put(length)) < bytes, "fixture: the receipt is the larger reply");
return length;
}
@ -3289,7 +3334,7 @@ async function withServedBoundRig(
connectionId: number,
serverLimits: { maxFragmentBytes?: number; maxPendingReliablePackets?: number; maxPayloadBytes?: number },
body: (rig: ServedBoundRig) => Promise<void>,
handler: { echoReceipt?: boolean } = {},
handler: { receipt?: "echo" | "oversized" } = {},
): Promise<void> {
const node = await CultMesh.startNode(
join(await mkdtemp(join(tmpdir(), "cultmesh-ts-served-bound-")), "node.ccmp"),
@ -3300,7 +3345,7 @@ async function withServedBoundRig(
const errors: Error[] = [];
const messages: CultNetMessage[] = [];
const unparsed: Error[] = [];
const server = CultMesh.createRudpDocumentServer("cultmesh-ts-served-bound-server", connectionId, {
const server = CultMesh.createRudpDocumentServer(SERVED_BOUND_SERVER, connectionId, {
documents: registry,
getCache: () => node.cache,
bindHost: "127.0.0.1",
@ -3322,9 +3367,15 @@ async function withServedBoundRig(
payload: encode(document.payload),
},
});
return handler.echoReceipt
? { binding: defineCultNetDocumentBinding({ definition: noteDocument }), recordKey: document.recordKey, value: document.payload as { noteId: string; body: string } }
: undefined;
if (!handler.receipt) return undefined;
// An echo acknowledges the stored value, as admission sizes it; an oversized receipt
// carries more than admission could know of.
return {
binding: defineCultNetDocumentBinding({ definition: noteDocument }),
recordKey: document.recordKey,
value: document.payload as { noteId: string; body: string },
...(handler.receipt === "oversized" ? { tags: ["x".repeat(8_000)] } : {}),
};
},
onError: (error) => errors.push(error),
});
@ -3431,9 +3482,9 @@ test("CultMesh TS RUDP document server honours maxPayloadBytes for replies and j
await waitForPeerErrors(rig, 1);
assert.ok(!refusal.message.includes("note:ov"), refusal.message);
// A put whose own frame is over the limit only by the provenance and message id it carries
// is served at the bound, so it is admitted and served.
const carried = notePut(`put-at-${"m".repeat(200)}`, "note:at", atBound, `runtime-${"r".repeat(200)}`);
// A put whose own frame is over the limit only by the provenance it carries is served at the
// bound, so it is admitted and served.
const carried = notePut("put-at", "note:at", atBound, `runtime-${"r".repeat(200)}`);
assert.ok(encode(carried).byteLength > limit, "fixture: the put's frame is over the limit");
rig.peer.send(carried);
await waitForAdmission(rig, "note:at");
@ -3459,26 +3510,45 @@ test("CultMesh TS RUDP document server honours maxPayloadBytes for replies and j
});
});
test("CultMesh TS RUDP document server refuses a put whose receipt could not be sent", async () => {
// The receipt is the larger reply here: a put whose receipt is one byte over the limit is
// refused before it is stored; one whose receipt is exactly at the limit is stored and gets it.
const limit = 4_000;
const overBound = bodyLengthForReceiptBytes("put-ov", "note:ro", limit + 1);
const atBound = bodyLengthForReceiptBytes("put-at", "note:ra", limit);
await withServedBoundRig(0x10203065, { maxFragmentBytes: 1024, maxPayloadBytes: limit }, async (rig) => {
rig.peer.send(notePut("put-ov", "note:ro", overBound, undefined, RECEIPT_STORED_AT));
const refusal = await waitForError(rig, "the put whose receipt is over the limit");
assert.ok(refusal instanceof CultMeshRudpUnservableDocumentError, refusal.message);
assert.equal(refusal.responseBytes, limit + 1);
assert.equal(refusal.maxPayloadBytes, limit);
assert.deepEqual(rig.admitted, []);
await waitForPeerErrors(rig, 1);
rig.peer.send(notePut("put-at", "note:ra", atBound, undefined, RECEIPT_STORED_AT));
await waitForAdmission(rig, "note:ra");
const startedAt = Date.now();
while (!rig.messages.some(isDocumentPut) && Date.now() - startedAt < 3_000) await delay(5);
const receipt = rig.messages.find(isDocumentPut) as CultNetDocumentPutRawMessage | undefined;
assert.equal(receipt?.messageId, "put-at", "the stored put gets its receipt");
assert.equal(rig.errors.length, 1);
}, { receipt: "echo" });
});
test("CultMesh TS RUDP document server tells the peer when a stored put's receipt cannot be sent", async () => {
// A handler whose receipt is larger than the echo admission sized: the put is stored, the
// receipt cannot be sent, and the peer hears that it is not coming.
const limit = 4_000;
const small = bodyLengthForReplyBytes("note:small", 1_000);
const atBound = bodyLengthForReplyBytes("note:at", limit);
await withServedBoundRig(0x10203064, { maxFragmentBytes: 1024, maxPayloadBytes: limit }, async (rig) => {
rig.peer.send(notePut("put-small", "note:small", small));
await waitForAdmission(rig, "note:small");
const startedAt = Date.now();
while (!rig.messages.some(isDocumentPut) && Date.now() - startedAt < 3_000) await delay(5);
assert.ok(rig.messages.some(isDocumentPut), "a receipt that fits is sent");
// Served at the bound, the put is admitted and stored; its echoed receipt carries the
// server's provenance and so is over the limit. The peer hears that it is not coming.
rig.peer.send(notePut("put-at", "note:at", atBound));
await waitForAdmission(rig, "note:at");
await waitForPeerErrors(rig, 1);
assert.equal(rig.errors.length, 1);
assert.match(rig.errors[0]!.message, /reply cultnet\.document_put_raw\.v0 is \d+ bytes; maxPayloadBytes is 4000/);
assert.deepEqual(servedBodies(await snapshotUnderShortestId(rig, "note:at")), [["note:at", atBound]]);
}, { echoReceipt: true });
assert.ok(!rig.messages.some(isDocumentPut));
assert.deepEqual(servedBodies(await snapshotUnderShortestId(rig, "note:small")), [["note:small", small]]);
}, { receipt: "oversized" });
});
function isDocumentPut(message: CultNetMessage): boolean {