diff --git a/packages/cultnet-ts/src/idunn-runtime-authority.ts b/packages/cultnet-ts/src/idunn-runtime-authority.ts index e01be597..2956cc2f 100644 --- a/packages/cultnet-ts/src/idunn-runtime-authority.ts +++ b/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 { 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; diff --git a/packages/cultnet-ts/test/idunn-runtime-authority.test.ts b/packages/cultnet-ts/test/idunn-runtime-authority.test.ts index 4b7b4865..a27499cf 100644 --- a/packages/cultnet-ts/test/idunn-runtime-authority.test.ts +++ b/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((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[][]),