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: admission sizes only the served reply, never a receipt

Browse source 
Admission sized the larger of the snapshot reply and a receipt echoing
the put, so a put servable at the bound was refused, within 89-125 bytes
of it, for a receipt an application that sends none never sends. Whether
a receipt is sent is onDocumentPutRaw's to decide after it stores the
put, so admission cannot know it (ruling: Self, fork A option a).

Admission now sizes only the snapshot reply serving the document alone.
A receipt that cannot be sent takes the existing path: onError hears why
and the peer is told with a cultnet.error.v0 that it is not coming. This
removes the receipt bound of 9757ed54 and its test.

The new test pins the band at its boundary with puts whose echoed receipt
is over the limit: exactly at the limit is admitted and served, with no
receipt and with an echoed one, and one byte over is refused.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 18:37:40 +02:00
commit a1065cabe0
2 changed files with 67 additions and 69 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -4289,22 +4289,21 @@ 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`, or whose receipt could not be, is refused
* before `onDocumentPutRaw` sees it: `onError` receives a
* `CultMeshRudpUnservableDocumentError` and the peer a `cultnet.error.v0`.
* `maxPendingReliablePackets`, is refused before `onDocumentPutRaw` sees it:
* `onError` receives a `CultMeshRudpUnservableDocumentError` and the peer a
* `cultnet.error.v0`.
*
* 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.
* The snapshot reply is sized from the record `documents` would store for the
* put, serving it alone. 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 receipt is not part of the judgement: whether one is sent is the
* handler's to decide after it stores the put.
*
* 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`.
* A reply that cannot be sent all the same, such as a receipt 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;
@ -4333,8 +4332,8 @@ const SHORTEST_MESSAGE_ID = "0";
/**
* A put the document server refused because it could never serve the document:
* the larger of the snapshot reply carrying it alone and the receipt
* acknowledging it (`responseBytes`) would exceed the server's reply limits.
* the snapshot reply carrying it alone (`responseBytes`) would exceed the
* server's reply limits.
*/
export class CultMeshRudpUnservableDocumentError extends Error {
constructor(
@ -6065,7 +6064,7 @@ export class CultMesh {
});
}
/** The one receipt builder, shared by the receipt send and by put admission. */
/** Builds the receipt `onDocumentPutRaw` returned, under the put's message id by default. */
function receiptMessage(receipt: CultMeshRudpDocumentReceipt, putMessageId: string): CultNetMessage {
return options.documents.createRawDocumentPutMessage(
receipt.binding, receipt.messageId ?? putMessageId, receipt.recordKey, receipt.value,
@ -6079,34 +6078,26 @@ export class CultMesh {
}
/**
* 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.
* 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. It is built from the record the
* registry would store, 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;
let replies: CultNetMessage[];
let reply: CultNetMessage;
try {
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,
),
];
reply = options.documents.createRawSnapshotResponseForPut(message, SHORTEST_MESSAGE_ID);
} catch {
replies = [{
reply = {
schemaVersion: "cultnet.snapshot_response_raw.v0",
messageId: SHORTEST_MESSAGE_ID,
documents: [message.document],
}];
};
}
const responseBytes = Math.max(...replies.map((reply) => encodeSchemaMessage(reply).byteLength));
const responseBytes = encodeSchemaMessage(reply).byteLength;
const fragmentCount = Math.max(1, Math.ceil(responseBytes / maxFragmentBytes));
if ((maxPayloadBytes === undefined || responseBytes <= maxPayloadBytes)
&& fragmentCount <= maxReplyFragments) {

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

@ -3246,7 +3246,7 @@ const servedBoundRegistry = () =>
/**
* 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.
* fixtures built with it pin the served bound; `RECEIPT_STORED_AT` puts make the receipt the larger reply.
*/
const SERVED_STORED_AT = `2026-09-30T00:00:00.000Z${"~".repeat(200)}`;
const RECEIPT_STORED_AT = "2026-09-30T00:00:00.000Z";
@ -3320,16 +3320,6 @@ function bodyLengthForReplyBytes(recordKey: string, bytes: number): number {
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;
}
async function withServedBoundRig(
connectionId: number,
serverLimits: { maxFragmentBytes?: number; maxPendingReliablePackets?: number; maxPayloadBytes?: number },
@ -3510,29 +3500,46 @@ 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.
test("CultMesh TS RUDP document server admits a put served at the bound, whatever receipt its handler returns", async () => {
// These puts keep a short storedAt, so the receipt echoing one is larger than its served reply,
// and over the limit. Admission sizes only the served reply: exactly at the limit is admitted
// with or without a receipt, one byte over is refused. An echoed receipt that cannot be sent
// goes the way of any unsendable receipt: the put is stored and the peer is told.
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);
const put = (messageId: string, recordKey: string, body: number) =>
notePut(messageId, recordKey, body, undefined, RECEIPT_STORED_AT);
const bodyServedAt = (messageId: string, recordKey: string, bytes: number): number => {
const probe = 2_000;
const length = probe + bytes - servedAloneBytes(put(messageId, recordKey, probe));
assert.equal(servedAloneBytes(put(messageId, recordKey, length)), bytes, "fixture: reply size is linear in the body");
assert.ok(echoReceiptBytes(put(messageId, recordKey, length)) > limit, "fixture: the echoed receipt is over the limit");
return length;
};
const atBound = bodyServedAt("put-at", "note:at", limit);
const overBound = bodyServedAt("put-ov", "note:ov", limit + 1);
for (const [connectionId, receipt] of [[0x10203066, undefined], [0x10203067, "echo"]] as const) {
await withServedBoundRig(connectionId, { maxFragmentBytes: 1024, maxPayloadBytes: limit }, async (rig) => {
rig.peer.send(put("put-at", "note:at", atBound));
await waitForAdmission(rig, "note:at");
if (receipt === "echo") {
const failure = await waitForError(rig, "the echoed receipt");
assert.match(failure.message, /reply cultnet\.document_put_raw\.v0 is \d+ bytes; maxPayloadBytes is 4000/);
await waitForPeerErrors(rig, 1);
assert.ok(!rig.messages.some(isDocumentPut));
}
assert.deepEqual(servedBodies(await snapshotUnderShortestId(rig, "note:at")), [["note:at", atBound]]);
const errorsBefore = rig.errors.length;
assert.equal(errorsBefore, receipt === "echo" ? 1 : 0, rig.errors.map((error) => error.message).join("; "));
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" });
rig.peer.send(put("put-ov", "note:ov", overBound));
const startedAt = Date.now();
while (rig.errors.length === errorsBefore && Date.now() - startedAt < 3_000) await delay(5);
const refusal = rig.errors[errorsBefore];
assert.ok(refusal instanceof CultMeshRudpUnservableDocumentError, refusal?.message);
assert.equal(refusal.responseBytes, limit + 1);
assert.deepEqual(rig.admitted, ["note:at"]);
}, receipt ? { receipt } : {});
}
});
test("CultMesh TS RUDP document server tells the peer when a stored put's receipt cannot be sent", async () => {