From 3c2445718c4ef0b6dbd4a5ded1f5716036f1879f Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 16 Jun 2026 05:18:26 +0100 Subject: [PATCH] Publish Weksa health over Idunn RUDP --- docs/verse-service-contract.md | 11 +- scripts/idunn-rudp-health.cjs | 233 +++++++++++++++++++++++++++++++ scripts/restart-weksa-daemon.ps1 | 7 +- scripts/start-weksa-daemon.ps1 | 12 +- scripts/weksa-daemon.mjs | 86 ++++++++++-- 5 files changed, 330 insertions(+), 19 deletions(-) create mode 100644 scripts/idunn-rudp-health.cjs diff --git a/docs/verse-service-contract.md b/docs/verse-service-contract.md index d0bb994..900a8e3 100644 --- a/docs/verse-service-contract.md +++ b/docs/verse-service-contract.md @@ -68,10 +68,13 @@ commit when the runtime lands. The local Starfire daemon is launched by `scripts/start-weksa-daemon.ps1`, checked by `scripts/health-weksa-daemon.cmd`, and restarted by -`scripts/restart-weksa-daemon.cmd`. Its compatibility HTTP surface listens on -`http://127.0.0.1:8813` with `/health`, `/provider-advertisement`, -`/operator-state`, `/eve/operator`, `/cultmesh/publications`, and -`POST /speech-provider/mimo/voicedesign`. +`scripts/restart-weksa-daemon.cmd`. Its daemon-owned health path publishes +`weksa.cultnet-rudp-provider-health` to Idunn over +`cultnet.transport.rudp.v0` after each serialized witness refresh. Its +compatibility HTTP surface still listens on `http://127.0.0.1:8813` with +`/health`, `/provider-advertisement`, `/operator-state`, `/eve/operator`, +`/cultmesh/publications`, and `POST /speech-provider/mimo/voicedesign`, but +those endpoints are operator/debug projections, not daemon liveness owners. ## CultMesh Verses diff --git a/scripts/idunn-rudp-health.cjs b/scripts/idunn-rudp-health.cjs new file mode 100644 index 0000000..c9e6761 --- /dev/null +++ b/scripts/idunn-rudp-health.cjs @@ -0,0 +1,233 @@ +"use strict"; + +const dgram = require("dgram"); +const path = require("path"); + +const { + CultNetRudpSession, + decodeRudpPacket, + encodeCultNetMessageForWire, + encodeRudpPacket, +} = require("cultnet-ts"); + +const { encode } = require(path.resolve( + __dirname, + "..", + "..", + "CultLib", + "node_modules", + "@msgpack", + "msgpack", +)); + +const CULTNET_RUDP_PROTOCOL_ID = "cultnet.transport.rudp.v0"; +const IDUNN_HEALTH_RUDP_CONNECTION_ID = 0x1d0d0001; + +function createIdunnRudpHealthPublisher(options) { + if (!options?.endpoint) return null; + return { + daemonId: options.daemonId, + endpoint: parseEndpoint(options.endpoint), + healthContract: options.healthContract, + }; +} + +async function publishIdunnRudpHealth(publisher, health) { + if (!publisher) return; + + const socket = dgram.createSocket(endpointFamily(publisher.endpoint.host)); + await bindSocket(socket, publisher.endpoint); + const receiver = createPacketReceiver(socket); + const session = new CultNetRudpSession({ + connectionId: IDUNN_HEALTH_RUDP_CONNECTION_ID, + initialSequence: 1, + resendDelayMs: 100, + }); + + try { + const connect = session.createConnect(Date.now(), new Uint8Array()); + await sendPacket(socket, publisher.endpoint, connect); + await receiveUntil(receiver, session, publisher.endpoint, (packet) => packet.packetType === "accept", 5000, "accept"); + + const observedAt = health.observedAt || new Date().toISOString(); + const record = { + daemonId: publisher.daemonId, + state: health.state, + detail: health.detail, + observedAt, + healthContract: publisher.healthContract, + publicationSource: "daemon-published", + transport: CULTNET_RUDP_PROTOCOL_ID, + }; + const payload = encode([ + record.daemonId, + record.state, + record.detail, + record.observedAt, + record.healthContract, + record.publicationSource, + record.transport, + ]); + const message = { + schemaVersion: "cultnet.document_put_raw.v0", + messageId: `weksa-health:${publisher.daemonId}:${observedAt.replace(/[:.]/g, "-")}`, + document: { + schemaId: "idunn.daemon_health", + recordKey: publisher.daemonId, + storedAt: observedAt, + payloadEncoding: "messagepack", + payload, + sourceRuntimeId: "weksa-daemon", + sourceRole: "daemon-health-publisher", + tags: [CULTNET_RUDP_PROTOCOL_ID], + }, + }; + const wirePayload = encode(encodeCultNetMessageForWire(message, "cultnet.schema.v0")); + const dataPackets = session.sendMany("schema", wirePayload, { + reliable: true, + ordered: true, + nowMs: Date.now(), + }); + const ack = receiveUntil(receiver, session, publisher.endpoint, (packet) => packet.packetType === "ack", 500, "ack") + .catch(() => undefined); + for (const packet of dataPackets) { + await sendPacket(socket, publisher.endpoint, packet); + } + await ack; + } finally { + receiver.close(); + socket.close(); + } +} + +async function bindSocket(socket, endpoint) { + await new Promise((resolve, reject) => { + socket.once("error", reject); + socket.bind(0, endpoint.host.includes(":") ? "::" : "0.0.0.0", () => { + socket.off("error", reject); + resolve(); + }); + }); +} + +function parseEndpoint(value) { + const text = String(value || "").trim(); + const ipv6 = text.match(/^\[([^\]]+)\]:(\d+)$/); + if (ipv6) return { host: ipv6[1], port: parsePort(ipv6[2]) }; + const index = text.lastIndexOf(":"); + if (index <= 0) { + throw new Error(`Idunn RUDP endpoint must be host:port, got "${value}".`); + } + return { host: text.slice(0, index), port: parsePort(text.slice(index + 1)) }; +} + +function parsePort(value) { + const port = Number(value); + if (!Number.isInteger(port) || port <= 0 || port > 65535) { + throw new Error(`Idunn RUDP endpoint port is invalid: ${value}`); + } + return port; +} + +function endpointFamily(host) { + return host.includes(":") ? "udp6" : "udp4"; +} + +async function receiveUntil(receiver, session, endpoint, predicate, timeoutMs, label) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + try { + const packet = await receiver.next(Math.min(100, deadline - Date.now()), label); + const result = session.receive(packet, Date.now()); + if (result.reply) { + throw new Error("Weksa RUDP health publisher received an unexpected reply-required packet."); + } + if (predicate(packet)) return; + } catch (error) { + if (error.code !== "ETIMEDOUT") throw error; + } + for (const packet of session.dueResends(Date.now())) { + await sendPacket(receiver.socket, endpoint, packet); + } + } + throw new Error(`timed out waiting for Idunn RUDP ${label} response after ${timeoutMs}ms`); +} + +function createPacketReceiver(socket) { + const packets = []; + const waiters = []; + const errors = []; + + const resolveNext = () => { + while (waiters.length > 0 && (packets.length > 0 || errors.length > 0)) { + const waiter = waiters.shift(); + clearTimeout(waiter.timer); + if (errors.length > 0) waiter.reject(errors.shift()); + else waiter.resolve(packets.shift()); + } + }; + const onMessage = (wire) => { + try { + packets.push(decodeRudpPacket(wire)); + } catch (error) { + errors.push(error); + } + resolveNext(); + }; + const onError = (error) => { + errors.push(error); + resolveNext(); + }; + + socket.on("message", onMessage); + socket.on("error", onError); + + return { + socket, + next(timeoutMs, label = "packet") { + if (packets.length > 0) return Promise.resolve(packets.shift()); + if (errors.length > 0) return Promise.reject(errors.shift()); + return new Promise((resolve, reject) => { + const waiter = { + resolve, + reject, + timer: setTimeout(() => { + const index = waiters.indexOf(waiter); + if (index >= 0) waiters.splice(index, 1); + const error = new Error(`timed out waiting for Idunn RUDP ${label}`); + error.code = "ETIMEDOUT"; + reject(error); + }, 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("Weksa RUDP health publisher closed."); + error.code = "ECLOSED"; + waiter.reject(error); + } + }, + }; +} + +async function sendPacket(socket, endpoint, packet) { + const wire = encodeRudpPacket(packet); + await new Promise((resolve, reject) => { + socket.send(wire, endpoint.port, endpoint.host, (error) => { + if (error) reject(error); + else resolve(); + }); + }); +} + +module.exports = { + CULTNET_RUDP_PROTOCOL_ID, + createIdunnRudpHealthPublisher, + publishIdunnRudpHealth, +}; diff --git a/scripts/restart-weksa-daemon.ps1 b/scripts/restart-weksa-daemon.ps1 index a1db803..17ff850 100644 --- a/scripts/restart-weksa-daemon.ps1 +++ b/scripts/restart-weksa-daemon.ps1 @@ -2,7 +2,10 @@ param( [int] $Port = 8813, [string] $HostName = "127.0.0.1", [string] $StateRoot = "E:\Projects\weksa\.weksa", - [string] $MiMoApiKeyPath = "E:\Projects\gamecult-ops\mimo-api-Weksa.txt" + [string] $MiMoApiKeyPath = "E:\Projects\gamecult-ops\mimo-api-Weksa.txt", + [string] $IdunnRudpHealth = "127.0.0.1:17870", + [string] $IdunnDaemon = "weksa", + [string] $IdunnHealthContract = "weksa.cultnet-rudp-provider-health" ) $ErrorActionPreference = "Stop" @@ -18,5 +21,5 @@ if (Test-Path -LiteralPath $pidPath) { Remove-Item -LiteralPath $pidPath -Force -ErrorAction SilentlyContinue } -& powershell.exe -NoProfile -ExecutionPolicy Bypass -File (Join-Path $repoRoot "scripts\start-weksa-daemon.ps1") -Port $Port -HostName $HostName -StateRoot $StateRoot -MiMoApiKeyPath $MiMoApiKeyPath +& powershell.exe -NoProfile -ExecutionPolicy Bypass -File (Join-Path $repoRoot "scripts\start-weksa-daemon.ps1") -Port $Port -HostName $HostName -StateRoot $StateRoot -MiMoApiKeyPath $MiMoApiKeyPath -IdunnRudpHealth $IdunnRudpHealth -IdunnDaemon $IdunnDaemon -IdunnHealthContract $IdunnHealthContract exit $LASTEXITCODE diff --git a/scripts/start-weksa-daemon.ps1 b/scripts/start-weksa-daemon.ps1 index fb228a7..b5ff2d8 100644 --- a/scripts/start-weksa-daemon.ps1 +++ b/scripts/start-weksa-daemon.ps1 @@ -2,7 +2,10 @@ param( [int] $Port = 8813, [string] $HostName = "127.0.0.1", [string] $StateRoot = "E:\Projects\weksa\.weksa", - [string] $MiMoApiKeyPath = "E:\Projects\gamecult-ops\mimo-api-Weksa.txt" + [string] $MiMoApiKeyPath = "E:\Projects\gamecult-ops\mimo-api-Weksa.txt", + [string] $IdunnRudpHealth = "127.0.0.1:17870", + [string] $IdunnDaemon = "weksa", + [string] $IdunnHealthContract = "weksa.cultnet-rudp-provider-health" ) $ErrorActionPreference = "Stop" @@ -13,6 +16,8 @@ $outLog = Join-Path $StateRoot "weksa-daemon.out.log" $errLog = Join-Path $StateRoot "weksa-daemon.err.log" New-Item -ItemType Directory -Force -Path $StateRoot | Out-Null +$cultLibPackages = Resolve-Path -LiteralPath (Join-Path $repoRoot "..\CultLib\packages") +$env:NODE_PATH = (@($cultLibPackages.Path, $env:NODE_PATH) | Where-Object { $_ }) -join [IO.Path]::PathSeparator if (Test-Path -LiteralPath $pidPath) { $pidText = (Get-Content -LiteralPath $pidPath -Raw).Trim() @@ -38,7 +43,10 @@ $arguments = @( "--host", $HostName, "--port", [string] $Port, "--state-root", $StateRoot, - "--mimo-api-key-path", $MiMoApiKeyPath + "--mimo-api-key-path", $MiMoApiKeyPath, + "--idunn-rudp-health", $IdunnRudpHealth, + "--idunn-daemon", $IdunnDaemon, + "--idunn-health-contract", $IdunnHealthContract ) $process = Start-Process -FilePath "node" ` diff --git a/scripts/weksa-daemon.mjs b/scripts/weksa-daemon.mjs index b0fb01f..0e0a6bf 100644 --- a/scripts/weksa-daemon.mjs +++ b/scripts/weksa-daemon.mjs @@ -7,6 +7,13 @@ import { basename, delimiter, dirname, extname, resolve } from "node:path"; import { fileURLToPath } from "node:url"; const repoRoot = resolve(dirname(fileURLToPath(import.meta.url)), ".."); +configureCultLibModulePath(); +const require = Module.createRequire(import.meta.url); +const { + CULTNET_RUDP_PROTOCOL_ID, + createIdunnRudpHealthPublisher, + publishIdunnRudpHealth, +} = require("./idunn-rudp-health.cjs"); const options = parseArgs(process.argv.slice(2)); const port = Number(options.port ?? process.env.WEKSA_DAEMON_PORT ?? 8813); const host = options.host ?? process.env.WEKSA_DAEMON_HOST ?? "127.0.0.1"; @@ -15,11 +22,18 @@ const cultMeshStorePath = resolve(stateRoot, "provider-advertisement-store.cc"); const pidPath = resolve(stateRoot, "weksa-daemon.pid"); const logPath = resolve(stateRoot, "weksa-daemon.log"); const mimoApiKeyPath = options.mimoApiKeyPath ?? process.env.WEKSA_MIMO_API_KEY_FILE ?? "E:/Projects/gamecult-ops/mimo-api-Weksa.txt"; +const idunnHealthPublisher = createIdunnRudpHealthPublisher({ + endpoint: options.idunnRudpHealth ?? process.env.WEKSA_IDUNN_RUDP_HEALTH, + daemonId: options.idunnDaemon ?? process.env.WEKSA_IDUNN_DAEMON ?? "weksa", + healthContract: options.idunnHealthContract ?? process.env.WEKSA_IDUNN_HEALTH_CONTRACT ?? "weksa.cultnet-rudp-provider-health", +}); const startedAt = new Date().toISOString(); let tick = 0; let lastPublishedAt = ""; let lastError = undefined; let cachedSnapshot = undefined; +let witnessRefreshRunning = false; +let lastRudpHealthSuccessMs = 0; if (options.health) { await runHealthCheck(); @@ -29,7 +43,7 @@ if (options.health) { await mkdir(stateRoot, { recursive: true }); await writeFile(pidPath, `${process.pid}\n`, "ascii"); await appendLog(`started pid=${process.pid} host=${host} port=${port}`); -await publishWitnesses(); +await refreshWitnesses("startup"); const server = http.createServer(async (request, response) => { try { @@ -61,7 +75,7 @@ const server = http.createServer(async (request, response) => { if (url.pathname === "/speech-provider/mimo/voicedesign" && request.method === "POST") { const body = await readJsonBody(request); const result = await handleMimoVoiceDesign(body); - await publishWitnesses(); + await refreshWitnesses("mimo-voicedesign-command"); return sendJson(response, 200, result); } sendJson(response, 404, { ok: false, error: "unknown endpoint" }); @@ -77,9 +91,10 @@ server.listen(port, host, () => { }); const interval = setInterval(() => { - publishWitnesses().catch(async (error) => { + refreshWitnesses("interval").catch(async (error) => { lastError = String(error?.stack ?? error); await appendLog(`publish error: ${lastError}`); + await publishWeksaRudpHealth("failed", `Weksa witness refresh failed: ${String(error?.message ?? error)}`); }); }, 30_000); @@ -131,15 +146,48 @@ async function publishWitnesses() { cachedSnapshot = snapshot; lastPublishedAt = snapshot.operatorState.updated_at; tick += 1; + return snapshot; } async function currentSnapshot() { if (!cachedSnapshot) { - await publishWitnesses(); + await refreshWitnesses("snapshot-demand"); } return cachedSnapshot; } +async function refreshWitnesses(reason) { + if (witnessRefreshRunning) { + await appendLog(`skipped overlapping witness refresh reason=${reason}`); + return cachedSnapshot; + } + witnessRefreshRunning = true; + try { + const snapshot = await publishWitnesses(); + await publishWeksaRudpHealth( + "healthy", + `Weksa refreshed witnesses reason=${reason} tick=${tick} provider=${snapshot.providerAdvertisement.providerId}`, + snapshot.operatorState.updated_at, + ); + return snapshot; + } finally { + witnessRefreshRunning = false; + } +} + +async function publishWeksaRudpHealth(state, detail, observedAt = new Date().toISOString()) { + if (!idunnHealthPublisher) return; + try { + await publishIdunnRudpHealth(idunnHealthPublisher, { state, detail, observedAt }); + lastRudpHealthSuccessMs = Date.now(); + } catch (error) { + const hasRecentSuccess = lastRudpHealthSuccessMs > 0 && Date.now() - lastRudpHealthSuccessMs < 300_000; + if (!hasRecentSuccess) { + await appendLog(`Idunn RUDP health publish failed: ${String(error?.message ?? error)}`); + } + } +} + async function buildSnapshot() { const updatedAt = new Date().toISOString(); const providerAdvertisement = loadProviderAdvertisement(updatedAt); @@ -182,7 +230,8 @@ async function buildSnapshot() { updated_at: updatedAt, transport: { cultmesh_store: ".weksa/provider-advertisement-store.cc", - native_cultnet_peer: "pending", + native_cultnet_peer: idunnHealthPublisher ? `${CULTNET_RUDP_PROTOCOL_ID}:idunn-health` : "pending", + idunn_rudp_health: idunnHealthPublisher ? `${idunnHealthPublisher.endpoint.host}:${idunnHealthPublisher.endpoint.port}` : "unconfigured", compatibility_http: `http://${host}:${port}`, }, publications: [ @@ -251,12 +300,7 @@ async function createCultMeshNodeWithOwnedStoreReset(CultMesh, storePath, docume function loadCultRuntime() { try { - process.env.NODE_PATH = [ - resolve(repoRoot, "..", "CultLib", "packages"), - process.env.NODE_PATH || "", - ].filter(Boolean).join(delimiter); - Module._initPaths(); - const require = Module.createRequire(import.meta.url); + configureCultLibModulePath(); const { CultMesh } = require("cultmesh-ts"); const { defineDocumentType } = require("cultcache-ts"); return { CultMesh, defineDocumentType, error: null }; @@ -265,6 +309,14 @@ function loadCultRuntime() { } } +function configureCultLibModulePath() { + process.env.NODE_PATH = [ + resolve(repoRoot, "..", "CultLib", "packages"), + process.env.NODE_PATH || "", + ].filter(Boolean).join(delimiter); + Module._initPaths(); +} + function defineCultMeshDocuments(defineDocumentType) { const surfaceDefinition = defineDocumentType({ type: "gamecult.eve.surface_state", @@ -928,6 +980,18 @@ function parseArgs(args) { parsed.mimoApiKeyPath = args[index + 1]; index += 1; break; + case "--idunn-rudp-health": + parsed.idunnRudpHealth = args[index + 1]; + index += 1; + break; + case "--idunn-daemon": + parsed.idunnDaemon = args[index + 1]; + index += 1; + break; + case "--idunn-health-contract": + parsed.idunnHealthContract = args[index + 1]; + index += 1; + break; case "--health": parsed.health = true; break;