From 9757ed54a58e46b3e58eb62ff505f1e2dde52aab Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 16:04:01 +0200 Subject: [PATCH] cultmesh-ts: a put is refused unless its receipt could be sent 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 --- packages/cultmesh-ts/src/index.ts | 83 +++++++++------ packages/cultmesh-ts/test/cultmesh.test.ts | 114 +++++++++++++++++---- 2 files changed, 142 insertions(+), 55 deletions(-) diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index 6da497e7..d3972cac 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -4289,19 +4289,22 @@ export interface CultMeshRudpDocumentReceipt encodeSchemaMessage(reply).byteLength)); const fragmentCount = Math.max(1, Math.ceil(responseBytes / maxFragmentBytes)); if ((maxPayloadBytes === undefined || responseBytes <= maxPayloadBytes) && fragmentCount <= maxReplyFragments) { diff --git a/packages/cultmesh-ts/test/cultmesh.test.ts b/packages/cultmesh-ts/test/cultmesh.test.ts index 006f6d83..8167e41d 100644 --- a/packages/cultmesh-ts/test/cultmesh.test.ts +++ b/packages/cultmesh-ts/test/cultmesh.test.ts @@ -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, - handler: { echoReceipt?: boolean } = {}, + handler: { receipt?: "echo" | "oversized" } = {}, ): Promise { 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 {