cultmesh-ts: the RUDP document server fragments its replies and refuses unservable puts
The server reassembled fragmented puts but sent every reply as one UDP datagram, ignoring its own maxFragmentBytes and maxPayloadBytes, and a datagram that failed to leave vanished. An 80 KB note put in 1 KiB fragments was admitted and its snapshot never arrived, with no error. Replies now fragment by maxFragmentBytes (default CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_MAX_FRAGMENT_BYTES, 1200), and maxPayloadBytes bounds both inbound frames and replies. A put whose served-alone reply (sized by the registry, under the shortest message id, with the send path's encoder) would exceed maxPayloadBytes, 65535 fragments or maxPendingReliablePackets is refused before onDocumentPutRaw: onError receives a CultMeshRudpUnservableDocumentError naming the sizes, and the peer a cultnet.error.v0. Every datagram send reports failure to onError, as does a snapshot reply that cannot be sent. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
2cf1e07bac
commit
0eb7434ab1
2 changed files with 340 additions and 18 deletions
|
|
@ -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<TDefinition extends AnyCultCacheDoc
|
|||
tags?: string[];
|
||||
}
|
||||
|
||||
/**
|
||||
* 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
|
||||
* `maxPendingReliablePackets`, is refused before `onDocumentPutRaw` sees it:
|
||||
* `onError` receives a `CultMeshRudpUnservableDocumentError` and the peer a
|
||||
* `cultnet.error.v0`. Every failed send reaches `onError`.
|
||||
*/
|
||||
export interface CultMeshRudpDocumentServerOptions extends CultMeshRudpSocketOptions {
|
||||
documents: CultNetDocumentRegistry;
|
||||
getCache?: () => Promise<CultCache> | 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<void> {
|
||||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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<typeof CultMesh.createRudpDocumentServer>;
|
||||
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<void>,
|
||||
): Promise<void> {
|
||||
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<CultNetMessage> {
|
||||
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<void> {
|
||||
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<Error> {
|
||||
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<void> {
|
||||
return new Promise(resolve => setTimeout(resolve, milliseconds));
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue