cultmesh-ts: write authority decide timeout counts as a typed refusal

A never-settling decide() held up the later frames of its session. The
server now races it against writeAuthorityTimeoutMs (default 5000) and
refuses with write_not_authorized on expiry. The granted-path test now
asserts the canonical schemaId the authority saw.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

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 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,7 @@ 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 = Math.max(1, options.writeAuthorityTimeoutMs ?? CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_DECIDE_TIMEOUT_MS);
     const wireContract = options.wireContract ?? "cultnet.schema.v0";
     const maxFragmentBytes = options.maxFragmentBytes ?? CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_MAX_FRAGMENT_BYTES;
     if (!(maxFragmentBytes > 0)) {
@@ -5796,20 +5806,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,22 @@ 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("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));