Merge pull request 'cultmesh-ts: end a RUDP session after 8 consecutive unreadable frames or authority-refused deletes' (#44) from libby/ts-refused-frames-cap into main

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<TDocument, TSchema>`, exported.
 - `documentById` and `tryDocumentById` on `CultMeshDocumentCatalog`, which address one document by its document id when a catalog holds several documents of one type.
 

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<void>;
     };
     const sessions = new Map<string, SessionRecord>();
@@ -5591,6 +5602,8 @@ export class CultMesh {
             connectPayload,
             closed: false,
             unsendable: false,
+            unreadableInARow: 0,
+            refusedDeletesInARow: 0,
             session: new CultNetRudpSession({
               connectionId,
               initialSequence: options.initialSequence,
@@ -5665,12 +5678,21 @@ 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;
-        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 (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."));
+        }
+        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 +5766,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 +5922,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;

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,87 @@ 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);
+  });
+});
+
+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(); }
+});