packages/cultnet-ts/src/idunn-odin-presence-publisher.ts

@@ -107,26 +107,36 @@ async function publishDocument(
     const wirePayload = encode(encodeCultNetMessageForWire(message as never, "cultnet.schema.v0"));
     const packets = session.sendMany("schema", wirePayload, { reliable: true, ordered: true, nowMs: Date.now() });
     for (const packet of packets) await sendPacket(socket, endpoint, packet);
-    await receiveUntil(
-      receiver,
-      session,
-      endpoint,
-      () => session.outstandingReliablePacketCount === 0,
-      2000,
-      (frame) => {
-        if (frame.channelId !== "schema") return;
-        const response = parseCultNetMessage(decode(frame.payload));
-        if (response.schemaVersion === "cultnet.error.v0") {
-          throw new Error(`Odin rejected runtime presence: ${response.error}`);
-        }
-      },
-    );
+    // The transport ack is not admission: the put completes on Odin's result frame. A refusal rejects whether
+    // it arrives before the ack, after it, or inside the result window; a success reply ends the wait early.
+    let succeeded = false;
+    const onFrame = (frame: { channelId: string; payload: Uint8Array }) => {
+      if (frame.channelId !== "schema") return;
+      const response = parseCultNetMessage(decode(frame.payload));
+      if (response.schemaVersion === "cultnet.error.v0") throw new IdunnPresenceRefusedError(response);
+      succeeded = true;
+    };
+    await receiveUntil(receiver, session, endpoint, () => session.outstandingReliablePacketCount === 0, 2000, onFrame);
+    if (!succeeded) await receiveUntil(receiver, session, endpoint, () => succeeded, RESULT_WINDOW_MS, onFrame, true);
   } finally {
     receiver.close();
     socket.close();
   }
 }
 
+/** How long after the transport ack Odin's refusal may still arrive; a success reply ends the wait early. */
+const RESULT_WINDOW_MS = 500;
+
+/** Odin refused the presence put: the typed `cultnet.error.v0` it answered with. */
+export class IdunnPresenceRefusedError extends Error {
+  readonly code: string | undefined;
+  constructor(readonly response: { error: string; code?: string | null }) {
+    super(`Odin rejected runtime presence: ${response.error}`);
+    this.name = "IdunnPresenceRefusedError";
+    this.code = response.code ?? undefined;
+  }
+}
+
 type PacketReceiver = {
   socket: dgram.Socket;
   next(timeoutMs: number): Promise<CultNetRudpPacket>;
@@ -190,6 +200,7 @@ async function receiveUntil(
   predicate: (packet: CultNetRudpPacket) => boolean,
   timeoutMs: number,
   onDelivered?: (frame: { channelId: string; payload: Uint8Array; sequence: number }) => void,
+  returnOnTimeout = false,
 ): Promise<void> {
   const deadline = Date.now() + timeoutMs;
   while (Date.now() < deadline) {
@@ -204,6 +215,7 @@ async function receiveUntil(
     }
     for (const packet of session.dueResends(Date.now())) await sendPacket(receiver.socket, endpoint, packet);
   }
+  if (returnOnTimeout) return;
   throw new Error(`Timed out waiting for CultNet RUDP response after ${timeoutMs} ms.`);
 }
 

packages/cultnet-ts/src/peer.ts

@@ -95,6 +95,10 @@ export class CultNetPeer extends EventEmitter {
     return this.#transport?.profile;
   }
 
+  /**
+   * Fire-and-forget: an ack or a return is not admission; read the result frame (a write is refused with
+   * `cultnet.error.v0`, `write_not_authorized` or `message_invalid`, as a `message` event).
+   */
   send(message: CultNetMessage): void {
     const wireValue = encodeCultNetMessageForWire(message, this.#wireContract);
     const payload = encode(wireValue);
@@ -106,6 +110,7 @@ export class CultNetPeer extends EventEmitter {
     this.#stream?.write(encodeFrame(payload));
   }
 
+  /** Resolves when the transport has acknowledged what was sent. An ack is not admission; read the result frame. */
   async flush(timeoutMs?: number): Promise<void> {
     await this.#transport?.flush?.(timeoutMs);
   }

packages/cultnet-ts/test/idunn-runtime-authority.test.ts

@@ -276,7 +276,10 @@ async function startOdinPeer(context: TestContext, endpoint: string) {
     peer.once("error", reject);
     peer.bind(port, "127.0.0.1", resolve);
   });
-  context.after(() => peer.close());
+  // The second packet of a split delivery is a timer; the test's end cancels it rather than racing the closed socket.
+  const timers = new Set<NodeJS.Timeout>();
+  const later = (send: () => void) => { timers.add(setTimeout(send, 120)); };
+  context.after(() => { for (const timer of timers) clearTimeout(timer); peer.close(); });
   const state = {
     reject: false,
     // How a refusal and the ack of the put reach the publisher: in one packet, or the ack first, or the refusal first.
@@ -309,10 +312,10 @@ async function startOdinPeer(context: TestContext, endpoint: string) {
       state.responseAck = response.ack;
       if (state.reject && state.order === "ack-then-refusal") {
         peer.send(encodeRudpPacket(serverSession.createAckForReceived(packet.sequence)), remote.port, remote.address);
-        setTimeout(() => peer.send(encodeRudpPacket(response), remote.port, remote.address), 120);
+        later(() => peer.send(encodeRudpPacket(response), remote.port, remote.address));
       } else if (state.reject && state.order === "refusal-then-ack") {
         peer.send(encodeRudpPacket({ ...response, ack: 0, ackMask: 0 }), remote.port, remote.address);
-        setTimeout(() => peer.send(encodeRudpPacket(serverSession!.createAckForReceived(packet.sequence)), remote.port, remote.address), 120);
+        later(() => peer.send(encodeRudpPacket(serverSession!.createAckForReceived(packet.sequence)), remote.port, remote.address));
       } else {
         peer.send(encodeRudpPacket(response), remote.port, remote.address);
       }
@@ -342,7 +345,7 @@ test("the Odin publisher publishes the signer's document to the endpoint Expecte
   assert.equal(peer.failure, undefined, peer.failure?.message);
 });
 
-for (const order of ["refusal-then-ack"] as const) {
+for (const order of ["ack-then-refusal", "refusal-then-ack"] as const) {
   test(`the Odin publisher does not complete on the ack of a refused presence (${order})`, async (context) => {
     const endpoint = expectedFacts("service").odinEndpoint!;
     const peer = await startOdinPeer(context, endpoint);