packages/cultmesh-ts/CHANGELOG.md

@@ -10,6 +10,7 @@ All notable changes to `@gamecult/cultmesh-ts` are recorded here, following
 - The package is renamed `cultmesh-ts` to `@gamecult/cultmesh-ts` (the repo already scopes `@gamecult/cultcache-ts`), so the unscoped name cannot be claimed by someone else. Import from `@gamecult/cultmesh-ts` and its subpaths; the API is unchanged. Depends on `@gamecult/cultnet-ts`.
 - `asSchemaAlias` is removed from `CultMeshDocumentHandle`, `CultMeshCollectionHandle` and their bound forms. Read the handle by the schema you mean.
 - `CultMeshDocumentCatalog` (`document`, `tryDocument`, `latest`, `replace`, `submitPrediction`, `authoritativeWriter`, `predictionWriter`, `watch`) and `CultMeshCollectionCatalog` (`collection`, `tryCollection`, `latest`, `watchChanges`) take `<TDocument = never, TSchema>` and yield `CultMeshCatalogValue<TDocument, TSchema>`: an explicit type argument wins, a CultCache document definition names its own value type, and a bare schema descriptor yields `unknown`. A value type is no longer inferred from a value or an expected result. The `parse` option is removed from these calls, and a lookup by type answers with that type's handle or none, not an alias of a compatible schema.
+- `createRudpDocumentServer` takes `writeAuthorityTimeoutMs` (default `CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_DECIDE_TIMEOUT_MS`, 5000; a value outside 1..2147483647 throws `RangeError`): a write authority whose `decide()` has not settled by then refuses the write like one that throws (`write_not_authorized`, and `onError` hears of it), so a hung authority no longer holds up the session's later frames.
 - `createRudpDocumentServer` refuses every peer put and delete unless it is given a `writeAuthority`. A refused write is answered with the typed `write_not_authorized` naming the record key (the bytes pinned in `error-vectors.json`), and `onDocumentPutRaw` is never called; pass `writeAuthority: new CultMeshGrantAllWrites()` to keep the old accept-everything behaviour. The authority is asked before the put is sized or decoded and before any handler runs, with the canonical schema id and the peer's address (the schema id is resolved to its canonical form first, which is the timing residual below); one that throws, rejects or answers anything but `granted: true` refuses. A frame the server cannot read is judged as the write it looks like (a put when it does not decode at all), so an ungranted peer gets `write_not_authorized` for it too; a granted peer gets `onError` and a `malformed_document_put` or `malformed_document_delete` refusal. A put carrying a wrong-typed `shardId` or `shardEpoch` is judged with null shard fields and, once granted, refused with `malformed_document_put`. A refusal's record key and reason are clipped (64 and 96 UTF-8 bytes) so the refusal is at most 309 bytes; a `maxPayloadBytes` below that cannot carry it, and the peer then hears nothing (the failure goes to `onError`). Residual: ungranted peers can still distinguish registered from unregistered schemas by timing. A peer `document_delete` now reaches the new `onDocumentDeleteRaw` once granted (without it, a text refusal) instead of the unsupported-request error.
 - Adding a document or collection to a catalog under a schema id or schema name another type already holds is refused with `SchemaConflictError`.
 

packages/cultmesh-ts/src/index.ts

@@ -4096,6 +4096,12 @@ export interface CultMeshRudpDocumentServerOptions extends CultMeshRudpSocketOpt
    * asked before the document is sized, resolved or decoded.
    */
   writeAuthority?: CultMeshRemoteWriteAuthority;
+  /**
+   * How long the write authority has to answer before the write is refused (default
+   * `CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_DECIDE_TIMEOUT_MS`). A `decide()` that never settles is a refusal like any
+   * other, so it cannot hold up the frames that follow it on the session.
+   */
+  writeAuthorityTimeoutMs?: number;
   onDocumentPutRaw?: (document: CultMeshRudpDocumentPut) =>
     | CultMeshRudpDocumentReceipt
     | void
@@ -4111,6 +4117,9 @@ export interface CultMeshRudpDocumentServerOptions extends CultMeshRudpSocketOpt
   sessionTimeoutMs?: number;
 }
 
+/** 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;
+
 /** 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;
 
@@ -5534,6 +5543,11 @@ export class CultMesh {
     let sendFailures = 0;
     const resendPollMs = Math.max(10, options.resendPollMs ?? 25);
     const sessionTimeoutMs = Math.max(1_000, options.sessionTimeoutMs ?? 30_000);
+    const decideTimeoutMs = options.writeAuthorityTimeoutMs ?? CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_DECIDE_TIMEOUT_MS;
+    // A timer past 2^31-1 ms (or NaN) fires after 1 ms in Node, which would refuse every write that is not instant.
+    if (!(decideTimeoutMs > 0 && decideTimeoutMs <= 0x7fffffff)) {
+      throw new RangeError("CultMesh RUDP document server writeAuthorityTimeoutMs must be greater than zero and at most 2147483647.");
+    }
     const wireContract = options.wireContract ?? "cultnet.schema.v0";
     const maxFragmentBytes = options.maxFragmentBytes ?? CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_MAX_FRAGMENT_BYTES;
     if (!(maxFragmentBytes > 0)) {
@@ -5796,20 +5810,32 @@ export class CultMesh {
       if (!authority) {
         reason = "no remote write authority is configured";
       } else {
+        let timer: ReturnType<typeof setTimeout> | undefined;
         try {
-          const decision = await authority.decide({
-            peer: { ...record.remote },
-            operation,
-            schemaId: options.documents.canonicalSchemaId(schemaId) ?? schemaId,
-            recordKey,
-            shardId,
-            shardEpoch,
-          });
-          if (decision?.granted === true) return false;
-          reason = typeof decision?.reason === "string" ? decision.reason : undefined;
+          const timedOut = Symbol("decide timeout");
+          const decision = await Promise.race([
+            Promise.resolve().then(() => authority.decide({
+              peer: { ...record.remote },
+              operation,
+              schemaId: options.documents.canonicalSchemaId(schemaId) ?? schemaId,
+              recordKey,
+              shardId,
+              shardEpoch,
+            })),
+            new Promise<typeof timedOut>((resolve) => { timer = setTimeout(() => resolve(timedOut), decideTimeoutMs); }),
+          ]);
+          if (decision === timedOut) {
+            reportError(new Error(`CultMesh remote write authority did not answer within ${decideTimeoutMs} ms; the write was refused.`));
+            reason = "the write authority timed out";
+          } else {
+            if (decision?.granted === true) return false;
+            reason = typeof decision?.reason === "string" ? decision.reason : undefined;
+          }
         } catch (error) {
           reportError(new Error(`CultMesh remote write authority failed (${error instanceof Error ? error.name : typeof error}); the write was refused.`));
           reason = "the write authority failed";
+        } finally {
+          clearTimeout(timer);
         }
       }
       try {

packages/cultmesh-ts/test/remote-write-authority.test.ts

@@ -81,7 +81,7 @@ interface Rig {
 
 let nextConnection = 0x10204000;
 
-async function rig(writeAuthority?: CultMeshRemoteWriteAuthority, extra: { maxPayloadBytes?: number } = {}): Promise<Rig> {
+async function rig(writeAuthority?: CultMeshRemoteWriteAuthority, extra: { maxPayloadBytes?: number; writeAuthorityTimeoutMs?: number } = {}): Promise<Rig> {
   const connectionId = nextConnection++;
   const puts: CultMeshRudpDocumentPut[] = [];
   const deletes: CultMeshRudpDocumentDelete[] = [];
@@ -154,6 +154,7 @@ test("a granted put and delete reach their handlers, send nothing back, and the
     for (const request of authority.requests) {
       assert.equal(request.peer.address, "127.0.0.1");
       assert.ok(request.peer.port > 0);
+      assert.equal(request.schemaId, SCHEMA_ID, "a granted write was judged under the canonical schema id");
     }
     assert.deepEqual(authority.requests[1]!.peer, r.deletes[0]!.remote, "the delete handler sees the peer the authority was asked about");
   });
@@ -234,6 +235,48 @@ test("an authority that throws or rejects refuses, fails closed, and is reported
   }
 });
 
+test("an authority that never settles refuses after the timeout and does not block the frames behind it", async () => {
+  // The first put hangs forever; later frames from the same session must still be judged.
+  const authority = new Recording((request) => request.recordKey === KEY_A ? new Promise(() => {}) : { granted: true });
+  await using(await rig(authority, { writeAuthorityTimeoutMs: 100 }), async (r) => {
+    r.peer.send(put(KEY_A));
+    r.peer.send(put(KEY_B));
+    await sleep(600);
+    assert.deepEqual(r.puts.map((p) => p.recordKey), [KEY_B], "the granted put behind the hung one was applied");
+    assert.equal(r.replies.length, 1);
+    assertRefused(r.replies[0], KEY_A);
+    assert.equal((r.replies[0] as { error: string }).error.includes("timed out"), true);
+    assert.equal(r.errors.length, 1);
+    assert.ok(r.errors[0]!.message.includes("did not answer within 100 ms"));
+  });
+});
+
+test("an authority that settles GRANTED after the timeout is still refused and the write never reaches the handler", async () => {
+  const authority = new Recording((request) => request.recordKey === KEY_A
+    ? new Promise((resolve) => setTimeout(() => resolve({ granted: true }), 300))
+    : { granted: true });
+  await using(await rig(authority, { writeAuthorityTimeoutMs: 100 }), async (r) => {
+    r.peer.send(put(KEY_A));
+    r.peer.send(del(KEY_A));
+    await sleep(700);
+    assert.equal(r.replies.length, 2, "one refusal per frame, no late second reply");
+    assertRefused(r.replies[0], KEY_A);
+    assertRefused(r.replies[1], KEY_A);
+    assert.deepEqual([r.puts.length, r.deletes.length], [0, 0], "the late grant never reached the handler");
+  });
+});
+
+test("a write authority timeout that is not a usable timer delay is refused at construction", () => {
+  for (const bad of [0, -1, Number.NaN, Number.POSITIVE_INFINITY, 2 ** 31]) {
+    assert.throws(() => CultMesh.createRudpDocumentServer("authority-server", 1, {
+      documents: new CultNetDocumentRegistry([defineCultNetDocumentBinding({ definition: note })]),
+      bindHost: "127.0.0.1",
+      bindPort: 0,
+      writeAuthorityTimeoutMs: bad,
+    }), RangeError, String(bad));
+  }
+});
+
 test("a refusal with no reason still says why, and a missing decision refuses", async () => {
   await using(await rig(new Recording(() => ({ granted: false }))), async (r) => {
     r.peer.send(put(KEY_A));