From 960b15f0d4b32e4bfa1731e24036305111989cd3 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 22 Aug 2026 02:48:39 +0200 Subject: [PATCH] Complete Idunn RUDP health handshake --- src/idunn-rudp-health.ts | 169 ++++++++++++++++++++++++-------- tests/idunn-rudp-health.test.ts | 94 ++++++++++++++++++ 2 files changed, 222 insertions(+), 41 deletions(-) create mode 100644 tests/idunn-rudp-health.test.ts diff --git a/src/idunn-rudp-health.ts b/src/idunn-rudp-health.ts index de0e5fa..98c9799 100644 --- a/src/idunn-rudp-health.ts +++ b/src/idunn-rudp-health.ts @@ -2,6 +2,8 @@ import dgram from "node:dgram"; import { randomUUID } from "node:crypto"; import { encode } from "@msgpack/msgpack"; import { + CultNetRudpSession, + decodeRudpPacket, encodeCultNetMessageForWire, encodeRudpPacket, type CultNetDocumentPutRawMessage, @@ -16,8 +18,8 @@ import { const CULTNET_RUDP_PROTOCOL_ID = "cultnet.transport.rudp.v0"; const IDUNN_HEALTH_RUDP_CONNECTION_ID = 0x1d0d0001; const RUDP_HEALTH_CONNECT_ATTEMPTS = 3; -const RUDP_PULSE_POST_SEND_GRACE_MS = 1000; -const RUDP_CONNECT_TO_DATA_GRACE_MS = 300; +const RUDP_ACCEPT_TIMEOUT_MS = 2_000; +const RUDP_ACK_TIMEOUT_MS = 1_000; const SIGNED_DAEMON_HEALTH_SCHEMA = "idunn.signed_daemon_health.v1"; const publisherIncarnationId = randomUUID(); let publisherSequence = 0; @@ -61,16 +63,30 @@ async function publishIdunnRudpHealthOnce( health: IdunnHealthInput, ): Promise { const socket = dgram.createSocket(endpointFamily(endpoint.host)); + const receiver = createPacketReceiver(socket); + const session = new CultNetRudpSession({ + connectionId: IDUNN_HEALTH_RUDP_CONNECTION_ID, + initialSequence: 1, + resendDelayMs: 100, + }); try { await bindSocket(socket, endpoint); - const connect = buildConnectPacket(); - await sendPacket(socket, endpoint, connect); - await sleep(RUDP_CONNECT_TO_DATA_GRACE_MS); + await sendPacket(socket, endpoint, session.createConnect(Date.now(), new Uint8Array())); + await receiveUntil(receiver, session, endpoint, "accept", RUDP_ACCEPT_TIMEOUT_MS); + const payload = await buildSignedDocumentPutPayload(config, health); - await sendPacket(socket, endpoint, buildSchemaDataPacket(payload)); - await sleep(RUDP_PULSE_POST_SEND_GRACE_MS); + const ack = receiveUntil(receiver, session, endpoint, "ack", RUDP_ACK_TIMEOUT_MS); + for (const packet of session.sendMany("schema", payload, { + reliable: true, + ordered: true, + nowMs: Date.now(), + })) { + await sendPacket(socket, endpoint, packet); + } + await ack; } finally { + receiver.close(); socket.close(); } } @@ -117,36 +133,6 @@ async function buildSignedDocumentPutPayload(config: HeimdallConfig, health: Idu return encode(encodeCultNetMessageForWire(message, "cultnet.schema.v0")); } -function buildConnectPacket(): CultNetRudpPacket { - return { - packetType: "connect", - connectionId: IDUNN_HEALTH_RUDP_CONNECTION_ID, - sequence: 1, - ack: 0, - ackMask: 0, - channelId: "control", - reliable: true, - ordered: true, - sequenced: false, - payload: new Uint8Array(), - }; -} - -function buildSchemaDataPacket(payload: Uint8Array): CultNetRudpPacket { - return { - packetType: "data", - connectionId: IDUNN_HEALTH_RUDP_CONNECTION_ID, - sequence: 2, - ack: 0, - ackMask: 0, - channelId: "schema", - reliable: true, - ordered: true, - sequenced: false, - payload, - }; -} - async function bindSocket(socket: dgram.Socket, endpoint: Endpoint): Promise { await new Promise((resolve, reject) => { socket.once("error", reject); @@ -180,10 +166,6 @@ async function sendPacket(socket: dgram.Socket, endpoint: Endpoint, input: CultN }); } -function sleep(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)); -} - function endpointFamily(host: string): "udp4" | "udp6" { return host.includes(":") ? "udp6" : "udp4"; } @@ -211,3 +193,108 @@ function parsePort(value: string): number { } return port; } + +type PacketReceiver = { + socket: dgram.Socket; + next(timeoutMs: number, label: string): Promise; + close(): void; +}; + +function createPacketReceiver(socket: dgram.Socket): PacketReceiver { + const packets: CultNetRudpPacket[] = []; + const errors: Error[] = []; + const waiters: Array<{ + resolve: (packet: CultNetRudpPacket) => void; + reject: (error: Error) => void; + timer: NodeJS.Timeout; + }> = []; + + const resolveNext = (): void => { + while (waiters.length > 0 && (packets.length > 0 || errors.length > 0)) { + const waiter = waiters.shift()!; + clearTimeout(waiter.timer); + const error = errors.shift(); + if (error) waiter.reject(error); + else waiter.resolve(packets.shift()!); + } + }; + const onMessage = (wire: Buffer): void => { + try { + packets.push(decodeRudpPacket(wire)); + } catch (error) { + errors.push(asError(error)); + } + resolveNext(); + }; + const onError = (error: Error): void => { + errors.push(error); + resolveNext(); + }; + socket.on("message", onMessage); + socket.on("error", onError); + + return { + socket, + next(timeoutMs, label) { + const packet = packets.shift(); + if (packet) return Promise.resolve(packet); + const error = errors.shift(); + if (error) return Promise.reject(error); + return new Promise((resolve, reject) => { + const waiter = { + resolve, + reject, + timer: setTimeout(() => { + const index = waiters.indexOf(waiter); + if (index >= 0) waiters.splice(index, 1); + const timeout = new Error(`timed out waiting for Idunn RUDP ${label}`) as Error & { code?: string }; + timeout.code = "ETIMEDOUT"; + reject(timeout); + }, Math.max(1, timeoutMs)), + }; + waiters.push(waiter); + }); + }, + close() { + socket.off("message", onMessage); + socket.off("error", onError); + while (waiters.length > 0) { + const waiter = waiters.shift()!; + clearTimeout(waiter.timer); + const error = new Error("Idunn RUDP packet receiver closed") as Error & { code?: string }; + error.code = "ECLOSED"; + waiter.reject(error); + } + }, + }; +} + +async function receiveUntil( + receiver: PacketReceiver, + session: CultNetRudpSession, + endpoint: Endpoint, + packetType: "accept" | "ack", + timeoutMs: number, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + try { + const packet = await receiver.next(Math.min(100, deadline - Date.now()), packetType); + const result = session.receive(packet, Date.now()); + if (result.reply) await sendPacket(receiver.socket, endpoint, result.reply); + if (packet.packetType === packetType) return; + } catch (error) { + if ((error as Error & { code?: string }).code !== "ETIMEDOUT") throw error; + } + for (const packet of session.dueResends(Date.now())) { + await sendPacket(receiver.socket, endpoint, packet); + } + } + const error = new Error(`timed out waiting for Idunn RUDP ${packetType}`) as Error & { code?: string }; + error.code = "ETIMEDOUT"; + throw error; +} + +function asError(value: unknown): Error { + return value instanceof Error ? value : new Error(String(value)); +} diff --git a/tests/idunn-rudp-health.test.ts b/tests/idunn-rudp-health.test.ts new file mode 100644 index 0000000..94d20ba --- /dev/null +++ b/tests/idunn-rudp-health.test.ts @@ -0,0 +1,94 @@ +import dgram from "node:dgram"; +import { mkdtemp } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { decode } from "@msgpack/msgpack"; +import { + CultNetRudpSession, + decodeRudpPacket, + encodeRudpPacket, + parseCultNetMessage, + type CultNetTransportFrame, +} from "cultnet-ts"; +import { describe, expect, it, vi } from "vitest"; +import { loadConfig } from "../src/config.js"; +import { publishIdunnRudpHealth } from "../src/idunn-rudp-health.js"; + +vi.mock("node:fs/promises", async (importOriginal) => { + const original = await importOriginal(); + return { + ...original, + readFile: async (file: Parameters[0], ...args: unknown[]) => { + if (file === "/etc/machine-id") return "heimdall-rudp-test-machine\n"; + return (original.readFile as (...values: unknown[]) => Promise)(file, ...args); + }, + }; +}); + +describe("Heimdall Idunn health publication", () => { + it("completes the canonical RUDP handshake before publishing its signed frame", async () => { + const server = dgram.createSocket("udp4"); + const session = new CultNetRudpSession({ + connectionId: 0x1d0d0001, + initialSequence: 1, + resendDelayMs: 100, + }); + let resolveFrame!: (frame: CultNetTransportFrame) => void; + const frameReceived = new Promise((resolve) => { + resolveFrame = resolve; + }); + + server.on("message", (wire, remote) => { + const packet = decodeRudpPacket(wire); + if (packet.packetType === "connect") { + server.send(encodeRudpPacket(session.acceptConnect(packet, Date.now())), remote.port, remote.address); + return; + } + const result = session.receive(packet, Date.now()); + if (result.reply) server.send(encodeRudpPacket(result.reply), remote.port, remote.address); + for (const frame of result.delivered) resolveFrame(frame); + if (result.delivered.length > 0) { + server.send(encodeRudpPacket(session.createAck()), remote.port, remote.address); + } + }); + await new Promise((resolve) => server.bind(0, "127.0.0.1", resolve)); + + try { + const tmp = await mkdtemp(path.join(os.tmpdir(), "heimdall-idunn-rudp-")); + const address = server.address(); + const config = loadConfig({ + GC_ACCESS_IDUNN_RUDP_HEALTH: `127.0.0.1:${address.port}`, + GC_ACCESS_PROVIDER_HEALTH_IDENTITY_PATH: path.join(tmp, "provider-health-identity.cc"), + }); + const observedAt = "2026-08-22T00:45:00.000Z"; + + await publishIdunnRudpHealth(config, { + daemonId: "yggdrasil-heimdall", + state: "active", + detail: "private-command-plane-ready", + observedAt, + }); + const frame = await frameReceived; + const message = parseCultNetMessage(decode(frame.payload), "cultnet.schema.v0"); + + expect(frame.channelId).toBe("schema"); + expect(message.schemaVersion).toBe("cultnet.document_put_raw.v0"); + if (message.schemaVersion !== "cultnet.document_put_raw.v0") { + throw new Error("Expected raw document publication."); + } + expect(message.document.schemaId).toBe("idunn.signed_daemon_health"); + expect(message.document.recordKey).toBe("yggdrasil-heimdall"); + expect(message.document.sourceRuntimeId).toBe("heimdall-service"); + expect(message.document.sourceRole).toBe("daemon-health-publisher"); + const statement = decode(message.document.payload) as unknown[]; + expect(statement).toHaveLength(17); + expect(statement[0]).toBe("idunn.signed_daemon_health.v1"); + expect(statement[1]).toBe("yggdrasil-heimdall"); + expect(statement[15]).toBeInstanceOf(Uint8Array); + expect((statement[15] as Uint8Array).byteLength).toBe(64); + expect(statement[16]).toBe(false); + } finally { + server.close(); + } + }); +});