From 621e05d552f0d98786695b4e68c107c420d3285e Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 09:41:18 +0000 Subject: [PATCH 1/2] cultmesh-ts: end a RUDP session after 8 consecutive unreadable frames or authority-refused deletes Heimdall's parity ruling. The refusal is still answered and the session kept below the cap; the frame that reaches CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES is answered too, then the session ends. The write-authority decision is unchanged. Co-Authored-By: Claude Sonnet 5.5 --- packages/cultmesh-ts/CHANGELOG.md | 1 + packages/cultmesh-ts/src/index.ts | 38 ++++++++++-- .../test/remote-write-authority.test.ts | 58 +++++++++++++++++++ 3 files changed, 93 insertions(+), 4 deletions(-) diff --git a/packages/cultmesh-ts/CHANGELOG.md b/packages/cultmesh-ts/CHANGELOG.md index 22d99904..d06e753b 100644 --- a/packages/cultmesh-ts/CHANGELOG.md +++ b/packages/cultmesh-ts/CHANGELOG.md @@ -16,6 +16,7 @@ All notable changes to `@gamecult/cultmesh-ts` are recorded here, following ### Added +- `createRudpDocumentServer` ends a session after `CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES` (8) consecutive undecodable frames or consecutive deletes the write authority refuses; fewer are answered and the session is kept. Any decoded frame resets the undecodable run, any decoded non-refused-delete frame resets the delete run. The same cap holds in every runtime. - `CultMeshCatalogValue`, exported. - `documentById` and `tryDocumentById` on `CultMeshDocumentCatalog`, which address one document by its document id when a catalog holds several documents of one type. diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index e8d526cc..a548e49f 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -4121,6 +4121,14 @@ export interface CultMeshRudpDocumentServerOptions extends CultMeshRudpSocketOpt /** A write authority that has not answered in this long has refused the write. */ export const CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_DECIDE_TIMEOUT_MS = 5_000; +/** + * How many consecutive refused frames of one kind a `createRudpDocumentServer` session may send before it is + * ended: undecodable frames (reset by any frame that decodes) and deletes the write authority refuses (reset by any + * decoded frame that is not a refused delete). Fewer are answered with a typed refusal and the session is kept; the + * frame that reaches this number is answered too, then the session ends. The same number is the cap in every runtime. + */ +export const CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES = 8; + /** 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; @@ -5537,6 +5545,9 @@ export class CultMesh { closed: boolean; /** A packet it owed could never be sent, so the session ended. */ unsendable: boolean; + /** Consecutive undecodable frames, and consecutive deletes the authority refused. */ + unreadableInARow: number; + refusedDeletesInARow: number; work: Promise; }; const sessions = new Map(); @@ -5591,6 +5602,8 @@ export class CultMesh { connectPayload, closed: false, unsendable: false, + unreadableInARow: 0, + refusedDeletesInARow: 0, session: new CultNetRudpSession({ connectionId, initialSequence: options.initialSequence, @@ -5666,11 +5679,16 @@ export class CultMesh { // an ungranted peer hears the same refusal for it as for any other write, not a parse error. const write = readableWrite(decoded); if (decoded !== undefined && !write) throw error; - if (await writeRefused(record, write?.operation ?? "put", write?.schemaId ?? "", write?.recordKey ?? "", null, null)) return; - reportError(error); - sendSchemaMessage(record, cultNetMessageInvalidError("message", "CultMesh RUDP document server could not read the write.")); + record.unreadableInARow += 1; + if (!await writeRefused(record, write?.operation ?? "put", write?.schemaId ?? "", write?.recordKey ?? "", null, null)) { + reportError(error); + sendSchemaMessage(record, cultNetMessageInvalidError("message", "CultMesh RUDP document server could not read the write.")); + } + if (record.unreadableInARow >= CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES) endCappedSession(record); return; } + record.unreadableInARow = 0; + if (message.schemaVersion !== "cultnet.document_delete.v0") record.refusedDeletesInARow = 0; switch (message.schemaVersion) { case "cultnet.snapshot_request.v0": { if (!options.getCache) { @@ -5744,7 +5762,12 @@ export class CultMesh { } return; case "cultnet.document_delete.v0": { - if (await writeRefused(record, "delete", message.schemaId, message.recordKey, null, null)) return; + if (await writeRefused(record, "delete", message.schemaId, message.recordKey, null, null)) { + record.refusedDeletesInARow += 1; + if (record.refusedDeletesInARow >= CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES) endCappedSession(record); + return; + } + record.refusedDeletesInARow = 0; if (!options.onDocumentDeleteRaw) { sendSchemaMessage(record, cultNetTextError("CultMesh RUDP document server has no delete handler.")); return; @@ -5895,6 +5918,13 @@ export class CultMesh { sendRudpDatagram(socket, goodbye, record.remote.port, record.remote.address, () => {}); } + /** A session that reached the cap of consecutive refused frames: its last refusal is sent, then it ends. */ + function endCappedSession(record: SessionRecord): void { + const key = `${record.remote.address}:${record.remote.port}`; + if (sessions.get(key) !== record) return; + endRefusedSession(key, record); + } + /** The session owes its peer a packet that can never be sent as built. */ function endUnsendableSession(record: SessionRecord, error: Error): void { if (record.unsendable) return; diff --git a/packages/cultmesh-ts/test/remote-write-authority.test.ts b/packages/cultmesh-ts/test/remote-write-authority.test.ts index 46fc5268..c6955d23 100644 --- a/packages/cultmesh-ts/test/remote-write-authority.test.ts +++ b/packages/cultmesh-ts/test/remote-write-authority.test.ts @@ -16,6 +16,7 @@ import { } from "@gamecult/cultnet-ts"; import { CultMesh, + CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES, CultMeshGrantAllWrites, type CultMeshRemoteWriteAuthority, type CultMeshRemoteWriteDecision, @@ -505,3 +506,60 @@ test("a granted peer's unreadable write reaches onError and gets a malformed ref } finally { r.close(); } } }); + +const CAP = 8; // the cap every runtime shares; the exported constant must stay this number + +test("the consecutive-refused-frames cap is the number every runtime shares", () => { + assert.equal(CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES, CAP); +}); + +test("unreadable frames are answered and the session kept until eight in a row; any decoded frame resets the run", async () => { + const r = await rawRig(new Recording(grant)); + try { + const batch = async (frames: number) => { + for (let i = 0; i < frames; i++) r.send(Uint8Array.from([0xc1])); + await sleep(400); + }; + await batch(CAP - 1); + assert.equal(r.replies.length, CAP - 1, "each is answered"); + r.send(encodeWireValue(put(KEY_A))); + await sleep(250); + assert.deepEqual(r.puts.map((p) => p.recordKey), [KEY_A], "the session was kept through seven"); + await batch(CAP - 1); + assert.equal(r.replies.length, 2 * (CAP - 1), "the valid frame reset the run"); + r.send(Uint8Array.from([0xc1])); + await sleep(300); + assert.equal(r.replies.length, 2 * (CAP - 1) + 1, "the eighth consecutive one is answered too"); + assert.ok(r.replies.every((reply) => (reply as { code?: string }).code === "message_invalid")); + // The goodbye reached the peer: its client no longer takes a send. + assert.throws(() => r.send(encodeWireValue(put(KEY_B))), /before the session is connected/, "then the session ended"); + assert.deepEqual(r.puts.map((p) => p.recordKey), [KEY_A]); + } finally { r.close(); } +}); + +test("deletes the authority refuses are answered and the session kept until eight in a row; any other decoded frame resets the run", async () => { + const authority = new Recording((request) => request.recordKey === KEY_B ? { granted: true } : { granted: false }); + await using(await rig(authority), async (r) => { + const refuse = async (deletes: number) => { + for (let i = 0; i < deletes; i++) r.peer.send(del(KEY_A)); + await sleep(400); + }; + await refuse(CAP - 1); + assert.equal(r.replies.length, CAP - 1); + r.peer.send(del(KEY_B)); // granted: resets the run + await sleep(250); + assert.deepEqual(r.deletes.map((d) => d.recordKey), [KEY_B]); + await refuse(CAP - 1); + assert.equal(r.replies.length, 2 * (CAP - 1)); + r.peer.send(put(KEY_B)); // a decoded non-delete resets too + await sleep(250); + await refuse(CAP - 1); + assert.equal(r.replies.length, 3 * (CAP - 1), "a put reset the run"); + r.peer.send(del(KEY_A)); + await sleep(300); + assert.equal(r.replies.length, 3 * (CAP - 1) + 1, "the eighth consecutive refused delete is answered too"); + for (const reply of r.replies) assertRefused(reply, KEY_A); + assert.throws(() => r.peer.send(del(KEY_B)), /before the session is connected/, "then the session ended"); + assert.equal(r.deletes.length, 1); + }); +}); From 23c7ce37382c8cb784f11e274881177fa7a59cc5 Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 09:52:22 +0000 Subject: [PATCH 2/2] cultmesh-ts: count and answer a frame that decodes but is no message the parser takes A map with an unknown schemaVersion was rethrown: unanswered, uncounted toward the session cap. It is now answered message_invalid, reported, and counted like any other unreadable frame, so the cap holds for it. Co-Authored-By: Claude Sonnet 5.5 --- packages/cultmesh-ts/src/index.ts | 8 ++++-- .../test/remote-write-authority.test.ts | 27 +++++++++++++++++++ 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index a548e49f..5479f14b 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -5678,9 +5678,13 @@ export class CultMesh { // A frame the server cannot read is judged as the write it looks like (a put when it does not even decode), so // an ungranted peer hears the same refusal for it as for any other write, not a parse error. const write = readableWrite(decoded); - if (decoded !== undefined && !write) throw error; record.unreadableInARow += 1; - if (!await writeRefused(record, write?.operation ?? "put", write?.schemaId ?? "", write?.recordKey ?? "", null, null)) { + if (decoded !== undefined && !write) { + // It decodes but is no message the parser takes (an unknown schemaVersion, a non-map) and is no write the + // authority could judge: answered and counted like any other unreadable frame. + reportError(error); + sendSchemaMessage(record, cultNetMessageInvalidError("message", "CultMesh RUDP document server could not read the message.")); + } else if (!await writeRefused(record, write?.operation ?? "put", write?.schemaId ?? "", write?.recordKey ?? "", null, null)) { reportError(error); sendSchemaMessage(record, cultNetMessageInvalidError("message", "CultMesh RUDP document server could not read the write.")); } diff --git a/packages/cultmesh-ts/test/remote-write-authority.test.ts b/packages/cultmesh-ts/test/remote-write-authority.test.ts index c6955d23..0cebbb8c 100644 --- a/packages/cultmesh-ts/test/remote-write-authority.test.ts +++ b/packages/cultmesh-ts/test/remote-write-authority.test.ts @@ -563,3 +563,30 @@ test("deletes the authority refuses are answered and the session kept until eigh assert.equal(r.deletes.length, 1); }); }); + +test("a frame that decodes but is no message the parser takes is answered, counted, and ended at eight in a row like any unreadable frame", async () => { + const r = await rawRig(new Recording(grant)); + const unknown = () => encodeWireValue({ schemaVersion: "cultnet.not-a-message.v0", messageId: "x" }); + try { + const batch = async (frames: number) => { + for (let i = 0; i < frames; i++) r.send(unknown()); + await sleep(400); + }; + await batch(CAP - 1); + assert.equal(r.replies.length, CAP - 1, "each is answered"); + assert.equal(r.errors.length, CAP - 1, "and reported"); + r.send(encodeWireValue(put(KEY_A))); + await sleep(250); + assert.deepEqual(r.puts.map((p) => p.recordKey), [KEY_A], "the session was kept through seven"); + await batch(CAP - 1); + assert.equal(r.replies.length, 2 * (CAP - 1), "the decoded put reset the run"); + r.send(unknown()); + await sleep(300); + assert.equal(r.replies.length, 2 * (CAP - 1) + 1, "the eighth is answered too"); + for (const reply of r.replies) { + assert.equal((reply as { code?: string }).code, "message_invalid"); + assert.equal((reply as { details: { field: string } }).details.field, "message"); + } + assert.throws(() => r.send(unknown()), /before the session is connected/, "then the session ended"); + } finally { r.close(); } +});