Fix Idunn runtime presence RUDP publication

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

@@ -13,7 +13,7 @@ import {
   encodeRuntimePresenceHealth,
   type RuntimePresenceHealth,
 } from "./runtime-presence-health";
-import { encodeCultNetMessageForWire } from "./contracts";
+import { encodeCultNetMessageForWire, parseCultNetMessage } from "./contracts";
 import { CultNetRudpSession, decodeRudpPacket, encodeRudpPacket } from "./rudp";
 import type { CultNetRudpPacket } from "./rudp";
 import dgram from "node:dgram";
@@ -306,11 +306,23 @@ async function publishDocument(
   try {
     await sendPacket(socket, endpoint, session.createConnect(Date.now(), new Uint8Array()));
     await receiveUntil(receiver, session, endpoint, (packet) => packet.packetType === "accept", 5000);
-    const wirePayload = Uint8Array.from(encodeCultNetMessageForWire(message as never, "cultnet.schema.v0") as Uint8Array);
+    const wirePayload = encode(encodeCultNetMessageForWire(message as never, "cultnet.schema.v0"));
     const packets = session.sendMany("schema", wirePayload, { reliable: true, ordered: true, nowMs: Date.now() });
-    const ack = receiveUntil(receiver, session, endpoint, (packet) => packet.packetType === "ack", 2000);
     for (const packet of packets) await sendPacket(socket, endpoint, packet);
-    await ack;
+    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}`);
+        }
+      },
+    );
   } finally {
     receiver.close();
     socket.close();
@@ -379,6 +391,7 @@ async function receiveUntil(
   endpoint: { host: string; port: number },
   predicate: (packet: CultNetRudpPacket) => boolean,
   timeoutMs: number,
+  onDelivered?: (frame: { channelId: string; payload: Uint8Array; sequence: number }) => void,
 ): Promise<void> {
   const deadline = Date.now() + timeoutMs;
   while (Date.now() < deadline) {
@@ -386,6 +399,7 @@ async function receiveUntil(
       const packet = await receiver.next(Math.min(100, deadline - Date.now()));
       const received = session.receive(packet, Date.now());
       if (received.reply) throw new Error("Runtime presence received an unexpected reply-required packet.");
+      for (const frame of received.delivered) onDelivered?.(frame);
       if (predicate(packet)) return;
     } catch (error) {
       if ((error as NodeJS.ErrnoException)?.code !== "ETIMEDOUT") throw error;

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

@@ -1,6 +1,7 @@
 import { mock, test } from "node:test";
 import assert from "node:assert/strict";
 import crypto from "node:crypto";
+import dgram from "node:dgram";
 import fs from "node:fs";
 import os from "node:os";
 import path from "node:path";
@@ -8,10 +9,12 @@ import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { decode, encode } from "@msgpack/msgpack";
 
 import {
+  createIdunnRuntimePresencePublisher,
   loadIdunnRuntimeAuthorityFromEnvironment,
   signIdunnRuntimePresence,
   systemdListenPidMatches,
 } from "../src/idunn-runtime-authority";
+import { CultNetRudpSession, decodeRudpPacket, encodeRudpPacket } from "../src/rudp";
 import {
   runtimePresenceActivationSigningMessage,
   runtimePresenceProofPayload,
@@ -30,7 +33,7 @@ test("matches systemd's host PID for the private PID namespace init only", () =>
   assert.equal(systemdListenPidMatches(undefined, 1), false);
 });
 
-test("opens Idunn's machine-bound identity and signs the exact Expected incarnation twice", (context) => {
+test("opens Idunn identity and publishes valid CultNet data while surfacing piggybacked application errors", async (context) => {
   const directory = mkdtempSync(path.join(os.tmpdir(), "cultnet-runtime-authority-"));
   context.after(() => rmSync(directory, { recursive: true, force: true }));
 
@@ -126,6 +129,64 @@ test("opens Idunn's machine-bound identity and signs the exact Expected incarnat
   assert.equal(crypto.verify(null, runtimePresenceProviderSigningMessage(proof), crypto.createPublicKey(providerKey), Buffer.from(signedRecord[21] as Uint8Array)), true);
   assert.equal(crypto.verify(null, runtimePresenceActivationSigningMessage(proof), crypto.createPublicKey(activationKey), Buffer.from(signedRecord[23] as Uint8Array)), true);
 
+  const peer = dgram.createSocket("udp4");
+  await new Promise<void>((resolve, reject) => {
+    peer.once("error", reject);
+    peer.bind(0, "127.0.0.1", resolve);
+  });
+  context.after(() => peer.close());
+  const address = peer.address();
+  assert.equal(typeof address, "object");
+  const peerPort = (address as { port: number }).port;
+  let serverSession: CultNetRudpSession | undefined;
+  let receivedMessage: unknown;
+  let responseAck: number | undefined;
+  let peerFailure: Error | undefined;
+  peer.on("message", (wire, remote) => {
+    try {
+      const packet = decodeRudpPacket(wire);
+      if (packet.packetType === "connect") {
+        serverSession = new CultNetRudpSession({ connectionId: packet.connectionId, initialSequence: 100 });
+        const accept = serverSession.acceptConnect(packet, Date.now());
+        peer.send(encodeRudpPacket(accept), remote.port, remote.address);
+        return;
+      }
+      if (!serverSession) throw new Error("Publisher sent data before establishing RUDP.");
+      const result = serverSession.receive(packet, Date.now());
+      const frame = result.delivered.find((candidate) => candidate.channelId === "schema");
+      if (!frame) return;
+      receivedMessage = decode(frame.payload);
+      const errorPayload = encode({
+        schemaVersion: "cultnet.error.v0",
+        error: "test admission denied",
+        routingHint: null,
+        code: null,
+        details: null,
+      });
+      const [response] = serverSession.sendMany("schema", errorPayload, {
+        reliable: true,
+        ordered: true,
+        nowMs: Date.now(),
+      });
+      assert.ok(response);
+      responseAck = response.ack;
+      peer.send(encodeRudpPacket(response), remote.port, remote.address);
+    } catch (error) {
+      peerFailure = error as Error;
+    }
+  });
+  const publisher = createIdunnRuntimePresencePublisher({
+    authority,
+    endpoint: `rudp://127.0.0.1:${peerPort}`,
+    healthContract: contract,
+    capabilities: [{ capability: "streampixels.service.api", schema: "streampixels.api.v1", compatibility: "v1", capacity: 1 }],
+  });
+  await assert.rejects(publisher.publish("warming", "local RUDP integration"), /Odin rejected runtime presence: test admission denied/);
+  assert.deepEqual(receivedMessage && (receivedMessage as { schemaVersion?: string }).schemaVersion, "cultnet.document_put_raw.v0");
+  assert.ok(Buffer.from(encode(receivedMessage)).byteLength > 0);
+  assert.equal(responseAck, 2, "the response data packet must acknowledge the publisher's reliable schema packet");
+  assert.equal(peerFailure, undefined, peerFailure?.message);
+
   descriptorProviderCredential = Buffer.from(encode([
     ...(decode(providerCredential) as unknown[][]),
     ...(decode(providerCredential) as unknown[][]),