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

cultmesh-ts: judge a put by what serving it sends, and tell the peer every refusal

Browse source 
The document server refused any delivered frame larger than
maxPayloadBytes, reporting it only to onError: the peer's put, already
acknowledged by the transport, vanished. The frame had been reassembled by
then, so the check protected no memory; it only refused puts whose frame
was over the limit because of provenance or a long message id although
their served reply fit. It is gone. maxPayloadBytes bounds what the server
sends; a put is judged by its served reply, and every refusal reaches the
peer as a cultnet.error.v0.

A stored put whose receipt cannot be sent (a receipt from onDocumentPutRaw
larger than maxPayloadBytes) now tells the peer so with a cultnet.error.v0,
as well as onError. The unservable refusal names sizes only, never the
record key, so a long key cannot make the refusal itself too large to send.

The options document states that the served-size judgement is exact only
for an application that stores through the registry it was given.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 15:37:54 +02:00
commit 6cf19b136a
2 changed files with 96 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -4285,12 +4285,23 @@ export interface CultMeshRudpDocumentReceipt<TDefinition extends AnyCultCacheDoc
* Options for `CultMesh.createRudpDocumentServer`.
*
* The server fragments its replies by `maxFragmentBytes` (default
* `CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_MAX_FRAGMENT_BYTES`) and refuses any
* inbound frame or reply larger than `maxPayloadBytes`. A put whose snapshot
* reply could never be sent within those limits, 65535 fragments and
* `CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_MAX_FRAGMENT_BYTES`) and sends no
* 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`. Every failed send reaches `onError`.
* `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.
*
* 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`.
*/
export interface CultMeshRudpDocumentServerOptions extends CultMeshRudpSocketOptions {
documents: CultNetDocumentRegistry;
@ -4330,8 +4341,10 @@ export class CultMeshRudpUnservableDocumentError extends Error {
readonly fragmentCount: number,
readonly maxFragmentCount: number,
) {
// 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 ${recordKey} can never be served: its snapshot reply is ${responseBytes} bytes in `
`Document can never be served: its snapshot reply is ${responseBytes} bytes in `
+ `${fragmentCount} fragments; limits are ${maxPayloadBytes ?? "unbounded"} bytes and `
+ `${maxFragmentCount} fragments.`,
);
@ -5884,13 +5897,6 @@ export class CultMesh {
if (frame.channelId !== "schema") {
continue;
}
if (maxPayloadBytes !== undefined && frame.payload.byteLength > maxPayloadBytes) {
reportError(new Error(
`CultMesh RUDP document server refused a ${frame.payload.byteLength}-byte frame from `
+ `${record.sessionId}; maxPayloadBytes is ${maxPayloadBytes}.`,
));
continue;
}
record.work = record.work
.then(() => handleDocumentServerFrame(record, frame.payload))
.catch(reportError);
@ -5949,15 +5955,24 @@ export class CultMesh {
}
const receipt = await options.onDocumentPutRaw(normalizeRudpDocumentPut(message, record.remote));
if (receipt) {
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"],
},
));
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"],
},
));
} catch (error) {
// The put is stored: the peer must hear that its receipt is not coming.
reportError(error);
sendSchemaMessage(record, {
schemaVersion: "cultnet.error.v0",
error: `CultMesh RUDP document server stored a put but could not send its receipt.`,
});
}
}
}
return;
@ -6058,8 +6073,9 @@ export class CultMesh {
* 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; a document the registry has no
* binding for is sized as received.
* 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.
*/
function unservablePut(message: CultNetMessage): CultMeshRudpUnservableDocumentError | undefined {
if (message.schemaVersion !== "cultnet.document_put_raw.v0") return undefined;

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

@ -3287,6 +3287,7 @@ async function withServedBoundRig(
connectionId: number,
serverLimits: { maxFragmentBytes?: number; maxPendingReliablePackets?: number; maxPayloadBytes?: number },
body: (rig: ServedBoundRig) => Promise<void>,
handler: { echoReceipt?: boolean } = {},
): Promise<void> {
const node = await CultMesh.startNode(
join(await mkdtemp(join(tmpdir(), "cultmesh-ts-served-bound-")), "node.ccmp"),
@ -3318,6 +3319,9 @@ 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;
},
onError: (error) => errors.push(error),
});
@ -3407,12 +3411,10 @@ test("CultMesh TS RUDP document server refuses a put whose reply would overflow
});
});
test("CultMesh TS RUDP document server honours maxPayloadBytes for replies, puts and inbound frames", async () => {
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);
const overBound = bodyLengthForReplyBytes("note:ov", limit + 1);
// A put's own frame is only a few bytes under its reply, so a long message id would push an
// at-bound put over the inbound limit before admission is asked. Keep these ids short.
await withServedBoundRig(0x10203062, { maxFragmentBytes: 1024, maxPayloadBytes: limit }, async (rig) => {
rig.peer.send(notePut("put-over", "note:ov", overBound));
const refusal = await waitForError(rig, "the over-bound put");
@ -3421,20 +3423,24 @@ test("CultMesh TS RUDP document server honours maxPayloadBytes for replies, puts
assert.equal(refusal.maxPayloadBytes, limit);
assert.equal(refusal.fragmentCount, 4);
assert.deepEqual(rig.admitted, []);
// The peer hears the refusal, and it names sizes, never the record.
assert.deepEqual(await waitForPeerErrors(rig, 1), [refusal.message]);
assert.ok(!refusal.message.includes("note:ov"), refusal.message);
rig.peer.send(notePut("put-at", "note:at", atBound));
// 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)}`);
assert.ok(encode(carried).byteLength > limit, "fixture: the put's frame is over the limit");
rig.peer.send(carried);
await waitForAdmission(rig, "note:at");
assert.deepEqual(servedBodies(await snapshotUnderShortestId(rig, "note:at")), [["note:at", atBound]]);
assert.equal(rig.errors.length, 1);
// A frame over the limit is refused on arrival and never offered as a put.
rig.peer.send(notePut("put-huge", "note:huge", 2 * limit));
await delay(50);
const startedAt = Date.now();
while (rig.errors.length < 2 && Date.now() - startedAt < 3_000) await delay(5);
assert.equal(rig.errors.length, 2);
assert.ok(!(rig.errors[1] instanceof CultMeshRudpUnservableDocumentError));
assert.match(rig.errors[1]!.message, /maxPayloadBytes is 4000/);
// A record key longer than the limit: the refusal is still small enough to reach the peer.
const longKey = `note:${"k".repeat(2 * limit)}`;
rig.peer.send(notePut("put-long-key", longKey, 10));
assert.equal((await waitForPeerErrors(rig, 2))[1], rig.errors[1]!.message);
assert.ok(rig.errors[1] instanceof CultMeshRudpUnservableDocumentError, rig.errors[1]!.message);
assert.deepEqual(rig.admitted, ["note:at"]);
// Two servable documents make a reply over the limit: it is not sent, and onError hears why.
@ -3448,6 +3454,44 @@ test("CultMesh TS RUDP document server honours maxPayloadBytes for replies, puts
});
});
test("CultMesh TS RUDP document server tells the peer when a stored put's receipt cannot be sent", async () => {
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");
const [refusal] = await waitForPeerErrors(rig, 1);
assert.match(refusal!, /stored a put but could not send its receipt/);
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 });
});
function isDocumentPut(message: CultNetMessage): boolean {
return message.schemaVersion === "cultnet.document_put_raw.v0";
}
/** Waits until the peer has received `count` cultnet.error.v0 replies, and returns their text. */
async function waitForPeerErrors(rig: ServedBoundRig, count: number): Promise<string[]> {
const errors = () => rig.messages
.filter((message) => message.schemaVersion === "cultnet.error.v0")
.map((message) => (message as { error: string }).error);
const startedAt = Date.now();
while (errors().length < count && Date.now() - startedAt < 3_000) await delay(5);
assert.equal(errors().length, count, `peer errors: ${errors().join("; ")}`);
return errors();
}
test("CultMesh TS RUDP document server reports a datagram that fails to leave", async () => {
// Fragments larger than any UDP datagram: the reply's one packet cannot be sent.
await withServedBoundRig(0x10203063, { maxFragmentBytes: 100_000, maxPendingReliablePackets: 512 }, async (rig) => {