Merge pull request 'cultmesh-ts: a put completes on its result frame, not on the transport ack' (#55) from libby/ts-client-ack-finding-cultmesh into main

packages/cultmesh-ts/src/index.ts

@@ -4183,7 +4183,29 @@ export interface CultMeshRudpDocumentServer {
   close(): void;
 }
 
+/**
+ * A write the peer refused: the typed `cultnet.error.v0` it answered with (`write_not_authorized`,
+ * `message_invalid`, ...), whichever of the transport ack and the result frame came first.
+ */
+export class CultMeshRudpWriteRefusedError extends Error {
+  readonly code: string | undefined;
+  readonly response: Extract<CultNetMessage, { schemaVersion: "cultnet.error.v0" }>;
+  constructor(response: Extract<CultNetMessage, { schemaVersion: "cultnet.error.v0" }>) {
+    super(response.error);
+    this.name = "CultMeshRudpWriteRefusedError";
+    this.code = response.code ?? undefined;
+    this.response = response;
+  }
+}
+
 export interface CultMeshRudpDocumentPublishOptions extends CultMeshRudpPeerOptions {
+  /**
+   * How long after the put's transport ack a refusal may still arrive. A reliable-transport ack is not
+   * admission; the outcome is the peer's result frame. A server that answers a put only on refusal sends no
+   * success frame, so the put is reported published once this window passes with no refusal; a success reply
+   * ends the wait early. Default 500 ms.
+   */
+  resultWindowMs?: number;
   messageId?: string;
   sourceRuntimeId?: string;
   sourceAgentId?: string;
@@ -6157,11 +6179,23 @@ export class CultMesh {
     requireNonEmpty(runtimeId, "runtimeId");
     requireNonEmpty(recordKey, "recordKey");
     const registry = new CultNetDocumentRegistry([binding]);
+    const messageId = options.messageId ?? `${runtimeId}:${binding.definition.type}:${recordKey}`;
     const peer = await CultMesh.createRudpPeer(runtimeId, connectionId, endpoint, options);
     try {
+      // A transport ack is not admission: the put completes on its result frame. A refusal rejects with the
+      // typed error whether it arrives before the ack, after it, or inside the result window.
+      let refusal: CultMeshRudpWriteRefusedError | undefined;
+      let succeeded = false;
+      let wake: (() => void) | undefined;
+      peer.on("message", (message: CultNetMessage) => {
+        if (message.schemaVersion === "cultnet.error.v0") refusal ??= new CultMeshRudpWriteRefusedError(message);
+        else if ("messageId" in message && message.messageId === messageId) succeeded = true;
+        else return;
+        wake?.();
+      });
       peer.send(registry.createRawDocumentPutMessage(
         binding,
-        options.messageId ?? `${runtimeId}:${binding.definition.type}:${recordKey}`,
+        messageId,
         recordKey,
         value,
         {
@@ -6172,6 +6206,13 @@ export class CultMesh {
         },
       ));
       await peer.flush(options.flushTimeoutMs ?? 30_000);
+      if (!refusal && !succeeded) {
+        await new Promise<void>((resolve) => {
+          const timer = setTimeout(resolve, options.resultWindowMs ?? 500);
+          wake = () => { clearTimeout(timer); resolve(); };
+        });
+      }
+      if (refusal) throw refusal;
     } finally {
       peer.close();
     }
@@ -6190,6 +6231,7 @@ export class CultMesh {
       const receipt = new Promise<CultCacheDocumentValue<TReceiptDefinition>>((resolve, reject) => {
         const timeout = setTimeout(() => { cleanup(); reject(new Error(`Timed out waiting for RUDP application receipt for ${recordKey}.`)); }, options.receiptTimeoutMs ?? 5_000);
         const onMessage = (message: CultNetMessage): void => {
+          if (message.schemaVersion === "cultnet.error.v0") { cleanup(); reject(new CultMeshRudpWriteRefusedError(message)); return; }
           if (message.schemaVersion !== "cultnet.document_put_raw.v0" || message.messageId !== messageId || message.document.schemaId !== receiptBinding.definition.schemaId) return;
           try {
             const decoded = decodeDocumentPayload(receiptBinding.definition, message.document.payload);

packages/cultmesh-ts/test/client-write-completion.test.ts

@@ -16,7 +16,10 @@ import {
   encodeRudpPacket,
 } from "@gamecult/cultnet-ts";
 import { encodeWireValue } from "@gamecult/cultnet-ts/wire";
-import { CultMesh } from "../src/index";
+import { CultMesh, CultMeshGrantAllWrites, CultMeshRudpWriteRefusedError } from "../src/index";
+
+const isTheRefusal = (error: unknown): boolean =>
+  error instanceof CultMeshRudpWriteRefusedError && error.code === "write_not_authorized";
 
 const note = defineDocumentType({
   type: "cultmesh.completion_note",
@@ -34,8 +37,12 @@ async function refusingPeer(connectionId: number, order: Order) {
   const socket: Socket = createSocket("udp4");
   await new Promise<void>((resolve) => socket.bind(0, "127.0.0.1", resolve));
   const session = new CultNetRudpSession({ connectionId, resendDelayMs: 50 });
-  const send = (packet: Parameters<typeof encodeRudpPacket>[0], port: number, address: string) =>
-    socket.send(encodeRudpPacket(packet), port, address);
+  const timers = new Set<NodeJS.Timeout>();
+  let closed = false;
+  const send = (packet: Parameters<typeof encodeRudpPacket>[0], port: number, address: string) => {
+    if (!closed) socket.send(encodeRudpPacket(packet), port, address);
+  };
+  const later = (action: () => void) => { const timer = setTimeout(() => { timers.delete(timer); action(); }, 120); timers.add(timer); };
   socket.on("message", (wire, remote) => {
     const packet = decodeRudpPacket(wire);
     const now = Date.now();
@@ -55,29 +62,56 @@ async function refusingPeer(connectionId: number, order: Order) {
       );
       if (order === "ack-then-refusal") {
         send(ack, remote.port, remote.address);
-        setTimeout(() => refusal.forEach((p) => send(p, remote.port, remote.address)), 120);
+        later(() => refusal.forEach((p) => send(p, remote.port, remote.address)));
       } else {
         // The refusal carries no acknowledgement of its own; the ack follows it.
         refusal.forEach((p) => send({ ...p, ack: 0, ackMask: 0 }, remote.port, remote.address));
-        setTimeout(() => send(ack, remote.port, remote.address), 120);
+        later(() => send(ack, remote.port, remote.address));
       }
     }
     if (packet.reliable) send(session.createAckForReceived(packet.sequence), remote.port, remote.address);
   });
-  return { endpoint: `rudp://127.0.0.1:${socket.address().port}`, close: () => socket.close() };
+  return { endpoint: `rudp://127.0.0.1:${socket.address().port}`, close: () => { closed = true; timers.forEach(clearTimeout); socket.close(); } };
 }
 
 const orders: Order[] = ["ack-then-refusal", "refusal-then-ack"];
 let nextConnection = 0x10205000;
 
 for (const order of orders) {
+  test(`publishRudpDocumentOnce does not complete on the ack of a refused put (${order})`, async () => {
+    const peer = await refusingPeer(nextConnection, order);
+    try {
+      await assert.rejects(
+        CultMesh.publishRudpDocumentOnce("completion-once", nextConnection++, peer.endpoint, binding, "tests:completion:a", { noteId: "a", text: "x" }, { resendDelayMs: 50, resendPollMs: 5, flushTimeoutMs: 2_000 }),
+        isTheRefusal,
+      );
+    } finally { peer.close(); }
+  });
+
   test(`publishRudpDocumentAndWaitForReceipt reports a refused put as refused, never as success (${order})`, async () => {
     const peer = await refusingPeer(nextConnection, order);
     try {
       await assert.rejects(
         CultMesh.publishRudpDocumentAndWaitForReceipt("completion-receipt", nextConnection++, peer.endpoint, binding, "tests:completion:a", { noteId: "a", text: "x" }, binding, { resendDelayMs: 50, resendPollMs: 5, receiptTimeoutMs: 800 }),
+        isTheRefusal,
       );
     } finally { peer.close(); }
     await sleep(10);
   });
 }
+
+test("publishRudpDocumentOnce still publishes a put the server admitted", async () => {
+  const connectionId = nextConnection++;
+  const puts: string[] = [];
+  const server = CultMesh.createRudpDocumentServer("completion-admit", connectionId, {
+    documents: new CultNetDocumentRegistry([binding]),
+    writeAuthority: new CultMeshGrantAllWrites(),
+    bindHost: "127.0.0.1", bindPort: 0, resendDelayMs: 50, resendPollMs: 5,
+    onDocumentPutRaw: (put) => { puts.push(put.recordKey); },
+  });
+  await server.start();
+  try {
+    await CultMesh.publishRudpDocumentOnce("completion-admit-client", connectionId, `rudp://127.0.0.1:${server.bind.port}`, binding, "tests:completion:a", { noteId: "a", text: "x" }, { resendDelayMs: 50, resendPollMs: 5, resultWindowMs: 200 });
+    assert.deepEqual(puts, ["tests:completion:a"]);
+  } finally { server.close(); }
+});