diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index b89cd823..72ce43d6 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -29,6 +29,7 @@ import { type CultNetOperationResponseMessage, type CultNetRawDocumentRecord, type CultNetReconnectPolicy, + type CultNetRudpPacket, type CultNetSchemaCatalogOptions, type CultNetSnapshotResponseRawMessage, type CultNetWireContract, @@ -4280,6 +4281,17 @@ export interface CultMeshRudpDocumentReceipt Promise | CultCache; @@ -4297,6 +4309,36 @@ export interface CultMeshRudpDocumentServerOptions extends CultMeshRudpSocketOpt sessionTimeoutMs?: number; } +/** Replies larger than one fragment are split into fragments of at most this many bytes. */ +export const CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_MAX_FRAGMENT_BYTES = 1200; + +const RUDP_MAX_FRAGMENTS_PER_MESSAGE = 0xffff; + +/** CultNet parses no message id shorter than one character, so no reply is smaller. */ +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. + */ +export class CultMeshRudpUnservableDocumentError extends Error { + constructor( + readonly messageId: string, + readonly recordKey: string, + readonly responseBytes: number, + readonly maxPayloadBytes: number | undefined, + readonly fragmentCount: number, + readonly maxFragmentCount: number, + ) { + super( + `Document ${recordKey} can never be served: its snapshot reply is ${responseBytes} bytes in ` + + `${fragmentCount} fragments; limits are ${maxPayloadBytes ?? "unbounded"} bytes and ` + + `${maxFragmentCount} fragments.`, + ); + this.name = "CultMeshRudpUnservableDocumentError"; + } +} + export interface CultMeshRudpServerSession { readonly sessionId: string; readonly remote: { address: string; family: string; port: number }; @@ -5773,6 +5815,17 @@ export class CultMesh { const resendPollMs = Math.max(10, options.resendPollMs ?? 25); const sessionTimeoutMs = Math.max(1_000, options.sessionTimeoutMs ?? 30_000); const wireContract = options.wireContract ?? "cultnet.schema.v0"; + const maxFragmentBytes = options.maxFragmentBytes ?? CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_MAX_FRAGMENT_BYTES; + if (!(maxFragmentBytes > 0)) { + throw new Error("CultMesh RUDP document server maxFragmentBytes must be greater than zero."); + } + const maxPayloadBytes = options.maxPayloadBytes; + // The most fragments one reply can take: the wire's fragment count, and the + // reliable queue a reply must fit into whole. + const maxReplyFragments = Math.min( + RUDP_MAX_FRAGMENTS_PER_MESSAGE, + options.maxPendingReliablePackets ?? RUDP_MAX_FRAGMENTS_PER_MESSAGE, + ); let resendTimer: NodeJS.Timeout | undefined; function reportError(error: unknown): void { @@ -5794,11 +5847,7 @@ export class CultMesh { if (packet.packetType === "connect") { const connectPayload = Uint8Array.from(packet.payload ?? []); if (record?.session.connectRepeats(packet)) { - socket.send( - encodeRudpPacket(record.session.acceptConnect(packet, nowMs)), - remote.port, - remote.address, - ); + sendPacket(record.session.acceptConnect(packet, nowMs), remote); return; } if (record) closeSession(record); @@ -5816,7 +5865,7 @@ export class CultMesh { work: Promise.resolve(), }; sessions.set(key, record); - socket.send(encodeRudpPacket(record.session.acceptConnect(packet, nowMs)), remote.port, remote.address); + sendPacket(record.session.acceptConnect(packet, nowMs), remote); return; } @@ -5826,15 +5875,22 @@ export class CultMesh { const result = record.session.receive(packet, nowMs); if (result.reply) { - socket.send(encodeRudpPacket(result.reply), record.remote.port, record.remote.address); + sendPacket(result.reply, record.remote); } for (const ready of result.readyToSend ?? []) { - socket.send(encodeRudpPacket(ready), record.remote.port, record.remote.address); + sendPacket(ready, record.remote); } for (const frame of result.delivered) { 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); @@ -5845,11 +5901,7 @@ export class CultMesh { return; } if (packet.reliable) { - socket.send( - encodeRudpPacket(record.session.createAckForReceived(packet.sequence)), - record.remote.port, - record.remote.address, - ); + sendPacket(record.session.createAckForReceived(packet.sequence), record.remote); } } catch (error) { reportError(error); @@ -5876,6 +5928,7 @@ export class CultMesh { const cache = await options.getCache(); sendSchemaMessage(record, options.documents.createRawSnapshotResponse(cache, message.messageId, message)); } catch (error) { + reportError(error); sendSchemaMessage(record, { schemaVersion: "cultnet.error.v0", error: error instanceof Error ? error.message : String(error), @@ -5888,6 +5941,12 @@ export class CultMesh { return; case "cultnet.document_put_raw.v0": if (options.onDocumentPutRaw) { + const unservable = unservablePut(message); + if (unservable) { + reportError(unservable); + sendSchemaMessage(record, { schemaVersion: "cultnet.error.v0", error: unservable.message }); + return; + } const receipt = await options.onDocumentPutRaw(normalizeRudpDocumentPut(message, record.remote)); if (receipt) { sendSchemaMessage(record, options.documents.createRawDocumentPutMessage( @@ -5966,16 +6025,70 @@ export class CultMesh { message: CultNetMessage, ): void { if (record.closed) throw new Error(`CultMesh RUDP session ${record.sessionId} is closed.`); - const payload = encode(encodeCultNetMessageForWire(message, wireContract)); + const payload = encodeSchemaMessage(message); + if (maxPayloadBytes !== undefined && payload.byteLength > maxPayloadBytes) { + throw new Error( + `CultMesh RUDP reply ${message.schemaVersion} is ${payload.byteLength} bytes; ` + + `maxPayloadBytes is ${maxPayloadBytes}.`, + ); + } for (const packet of record.session.sendMany("schema", payload, { reliable: true, ordered: true, nowMs: Date.now(), + maxFragmentBytes, })) { - socket.send(encodeRudpPacket(packet), record.remote.port, record.remote.address); + sendPacket(packet, record.remote); } } + /** The one reply encoder, shared by sends and by put admission. */ + function encodeSchemaMessage(message: CultNetMessage): Uint8Array { + return encode(encodeCultNetMessageForWire(message, wireContract)); + } + + /** A datagram that fails to leave reaches onError; nothing vanishes. */ + function sendPacket(packet: CultNetRudpPacket, remote: { address: string; port: number }): void { + socket.send(encodeRudpPacket(packet), remote.port, remote.address, (error) => { + if (error) reportError(error); + }); + } + + /** + * 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. + */ + function unservablePut(message: CultNetMessage): CultMeshRudpUnservableDocumentError | undefined { + if (message.schemaVersion !== "cultnet.document_put_raw.v0") return undefined; + let reply: CultNetSnapshotResponseRawMessage; + try { + reply = options.documents.createRawSnapshotResponseForPut(message, SHORTEST_MESSAGE_ID); + } catch { + reply = { + schemaVersion: "cultnet.snapshot_response_raw.v0", + messageId: SHORTEST_MESSAGE_ID, + documents: [message.document], + }; + } + const responseBytes = encodeSchemaMessage(reply).byteLength; + const fragmentCount = Math.max(1, Math.ceil(responseBytes / maxFragmentBytes)); + if ((maxPayloadBytes === undefined || responseBytes <= maxPayloadBytes) + && fragmentCount <= maxReplyFragments) { + return undefined; + } + return new CultMeshRudpUnservableDocumentError( + message.messageId, + message.document.recordKey, + responseBytes, + maxPayloadBytes, + fragmentCount, + maxReplyFragments, + ); + } + return { bind, async start(): Promise { @@ -5999,7 +6112,7 @@ export class CultMesh { continue; } for (const packet of record.session.dueResends(nowMs)) { - socket.send(encodeRudpPacket(packet), record.remote.port, record.remote.address); + sendPacket(packet, record.remote); } } }, resendPollMs); diff --git a/packages/cultmesh-ts/test/cultmesh.test.ts b/packages/cultmesh-ts/test/cultmesh.test.ts index 3954c697..f7c4c6c1 100644 --- a/packages/cultmesh-ts/test/cultmesh.test.ts +++ b/packages/cultmesh-ts/test/cultmesh.test.ts @@ -3,7 +3,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import test from "node:test"; import assert from "node:assert/strict"; -import { encode } from "@msgpack/msgpack"; +import { decode, encode } from "@msgpack/msgpack"; import { z } from "zod"; import { defineDocumentType } from "@gamecult/cultcache-ts"; import { @@ -11,8 +11,11 @@ import { CultNetPeer, cultNetBuiltinSchemaRegistry, defineCultNetDocumentBinding, + type CultNetDocumentPutRawMessage, + type CultNetMessage, + type CultNetSnapshotResponseRawMessage, } from "cultnet-ts"; -import { CultMesh } from "../src/index"; +import { CultMesh, CultMeshRudpUnservableDocumentError } from "../src/index"; const noteDocument = defineDocumentType({ type: "cultmesh.note", @@ -3223,6 +3226,212 @@ test("CultMesh TS negotiates streaming frame body transports explicitly", () => assert.deepEqual(frameUpdates, [42n]); }); +// A put the RUDP document server admits must be one it can serve. The server +// fragments its replies as its puts are reassembled, and refuses a put whose +// snapshot reply, carrying the document alone under the shortest message id, +// could never be sent within its limits. +type ServedBoundRig = { + server: ReturnType; + peer: CultNetPeer; + admitted: string[]; + errors: Error[]; + messages: CultNetMessage[]; +}; + +const servedBoundRegistry = () => + new CultNetDocumentRegistry([defineCultNetDocumentBinding({ definition: noteDocument })]); + +function notePut(messageId: string, recordKey: string, bodyLength: number): CultNetDocumentPutRawMessage { + return { + schemaVersion: "cultnet.document_put_raw.v0", + messageId, + document: { + schemaId: "cultmesh.note.v0", + recordKey, + storedAt: "2026-09-30T00:00:00.000Z", + payloadEncoding: "messagepack", + payload: encode({ noteId: recordKey, body: "x".repeat(bodyLength) }), + }, + }; +} + +/** The encoded size of the reply that would serve this put's document alone. */ +function servedAloneBytes(put: CultNetDocumentPutRawMessage): number { + return encode(servedBoundRegistry().createRawSnapshotResponseForPut(put, "0")).byteLength; +} + +/** The body length whose served-alone reply is exactly `bytes`, with a guard. */ +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"); + return length; +} + +async function withServedBoundRig( + connectionId: number, + serverLimits: { maxFragmentBytes?: number; maxPendingReliablePackets?: number; maxPayloadBytes?: number }, + body: (rig: ServedBoundRig) => Promise, +): Promise { + const node = await CultMesh.startNode( + join(await mkdtemp(join(tmpdir(), "cultmesh-ts-served-bound-")), "node.ccmp"), + { documents: [noteDocument] }, + ); + const registry = servedBoundRegistry(); + const admitted: string[] = []; + const errors: Error[] = []; + const messages: CultNetMessage[] = []; + const server = CultMesh.createRudpDocumentServer("cultmesh-ts-served-bound-server", connectionId, { + documents: registry, + getCache: () => node.cache, + bindHost: "127.0.0.1", + bindPort: 0, + resendDelayMs: 25, + resendPollMs: 5, + ...serverLimits, + // Stored as the registry stores a raw put, so the served record is the one it sizes. + onDocumentPutRaw: async (document) => { + admitted.push(document.recordKey); + await registry.applyRawDocumentPutMessage(node.cache, { + schemaVersion: "cultnet.document_put_raw.v0", + messageId: `store:${document.recordKey}`, + document: { + schemaId: document.schemaId, + recordKey: document.recordKey, + storedAt: document.storedAt, + payloadEncoding: "messagepack", + payload: encode(document.payload), + }, + }); + }, + onError: (error) => errors.push(error), + }); + let peer: CultNetPeer | undefined; + try { + await server.start(); + peer = await CultMesh.createRudpPeer( + "cultmesh-ts-served-bound-client", + connectionId, + `rudp://127.0.0.1:${server.bind.port}`, + { resendDelayMs: 25, resendPollMs: 5, maxFragmentBytes: 1024, maxPendingReliablePackets: 512, connectTimeoutMs: 1_000 }, + ); + peer.on("message", (message: CultNetMessage) => messages.push(message)); + await body({ server, peer, admitted, errors, messages }); + } finally { + peer?.close(); + server.close(); + } +} + +/** Requests `recordKey` under the shortest message id and waits for the reply. */ +async function snapshotUnderShortestId(rig: ServedBoundRig, recordKey: string): Promise { + const before = rig.messages.length; + rig.peer.sendSnapshotRequest({ schemaVersion: "cultnet.snapshot_request.v0", messageId: "0", recordKeys: [recordKey] }); + const startedAt = Date.now(); + while (Date.now() - startedAt < 3_000) { + const reply = rig.messages.slice(before).find((message) => + message.schemaVersion === "cultnet.snapshot_response_raw.v0" && message.messageId === "0"); + if (reply) return reply; + await delay(5); + } + throw new Error(`Timed out waiting for the snapshot of ${recordKey}; server errors: ${rig.errors.map(e => e.message).join("; ")}`); +} + +function servedBodies(reply: CultNetMessage): Array<[string, number]> { + assert.equal(reply.schemaVersion, "cultnet.snapshot_response_raw.v0"); + return (reply as CultNetSnapshotResponseRawMessage).documents.map((document) => [ + document.recordKey, + (decode(document.payload) as { body: string }).body.length, + ]); +} + +async function waitForAdmission(rig: ServedBoundRig, recordKey: string): Promise { + const startedAt = Date.now(); + while (!rig.admitted.includes(recordKey) && Date.now() - startedAt < 3_000) await delay(5); + assert.ok(rig.admitted.includes(recordKey), `${recordKey} was not admitted; server errors: ${rig.errors.map(e => e.message).join("; ")}`); +} + +async function waitForError(rig: ServedBoundRig, description: string): Promise { + const startedAt = Date.now(); + while (rig.errors.length === 0 && Date.now() - startedAt < 3_000) await delay(5); + assert.ok(rig.errors.length > 0, `no server error for ${description}`); + return rig.errors[0]!; +} + +test("CultMesh TS RUDP document server serves an 80 KB note put in 1 KiB fragments", async () => { + await withServedBoundRig(0x10203060, { maxFragmentBytes: 1024, maxPendingReliablePackets: 512 }, async (rig) => { + rig.peer.send(notePut("put-large", "note:large", 80_000)); + await waitForAdmission(rig, "note:large"); + assert.deepEqual(servedBodies(await snapshotUnderShortestId(rig, "note:large")), [["note:large", 80_000]]); + assert.deepEqual(rig.errors, []); + }); +}); + +test("CultMesh TS RUDP document server refuses a put whose reply would overflow its reliable queue", async () => { + // Eight 1 KiB fragments fit the queue, nine do not. + const atBound = bodyLengthForReplyBytes("note:at", 8 * 1024); + const overBound = bodyLengthForReplyBytes("note:ov", 8 * 1024 + 1); + await withServedBoundRig(0x10203061, { maxFragmentBytes: 1024, maxPendingReliablePackets: 8 }, async (rig) => { + rig.peer.send(notePut("put-over", "note:ov", overBound)); + const refusal = await waitForError(rig, "the over-bound put"); + assert.ok(refusal instanceof CultMeshRudpUnservableDocumentError, refusal.message); + assert.equal(refusal.messageId, "put-over"); + assert.equal(refusal.recordKey, "note:ov"); + assert.equal(refusal.responseBytes, 8 * 1024 + 1); + assert.equal(refusal.fragmentCount, 9); + assert.equal(refusal.maxFragmentCount, 8); + assert.equal(refusal.maxPayloadBytes, undefined); + assert.match(refusal.message, /8193 bytes in 9 fragments/); + assert.deepEqual(rig.admitted, []); + + rig.peer.send(notePut("put-at", "note:at", atBound)); + await waitForAdmission(rig, "note:at"); + assert.deepEqual(servedBodies(await snapshotUnderShortestId(rig, "note:at")), [["note:at", atBound]]); + assert.equal(rig.errors.length, 1); + }); +}); + +test("CultMesh TS RUDP document server honours maxPayloadBytes for replies, puts and inbound frames", async () => { + const limit = 4_000; + const atBound = bodyLengthForReplyBytes("note:at", limit); + const overBound = bodyLengthForReplyBytes("note:ov", limit + 1); + 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"); + assert.ok(refusal instanceof CultMeshRudpUnservableDocumentError, refusal.message); + assert.equal(refusal.responseBytes, limit + 1); + assert.equal(refusal.maxPayloadBytes, limit); + assert.equal(refusal.fragmentCount, 4); + assert.deepEqual(rig.admitted, []); + + rig.peer.send(notePut("put-at", "note:at", atBound)); + 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/); + assert.deepEqual(rig.admitted, ["note:at"]); + }); +}); + +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) => { + rig.peer.send(notePut("put-large", "note:large", 80_000)); + await waitForAdmission(rig, "note:large"); + rig.peer.sendSnapshotRequest({ schemaVersion: "cultnet.snapshot_request.v0", messageId: "0", recordKeys: ["note:large"] }); + const failure = await waitForError(rig, "the oversized datagram"); + assert.match(failure.message, /EMSGSIZE|message too long/i); + }); +}); + function delay(milliseconds: number): Promise { return new Promise(resolve => setTimeout(resolve, milliseconds)); }