diff --git a/README.md b/README.md index 7e1552f..4ab797b 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,9 @@ renderers such as Nightwing. - Eve deck: `ws://127.0.0.1:8802/eve/deck` - Provider manifest: `http://127.0.0.1:8802/eve/deck/providers` - Health: `http://127.0.0.1:8802/health` +- Idunn daemon health: `idunn.daemon_health` over + `cultnet.transport.rudp.v0`, defaulting to `127.0.0.1:17870` with daemon id + `stonks` and contract `stonks.cultnet-rudp-market-health`. - Snapshot: `http://127.0.0.1:8802/market/state` - CultCache state: `scratch/stonks/stonks-state.cc` - Request events: persisted as keyed `stonks.request_event.v1` CultCache @@ -63,3 +66,14 @@ At the default 15 second poll interval and 48 calls/minute budget, Stonks pulls up to 12 Finnhub equity quotes per refresh. It prioritizes recently mentioned watch symbols, uncached symbols, and then a round-robin pass across the configured watchlist. + +Idunn health publishing is enabled by the launcher with: + +```powershell +--idunn-rudp-health 127.0.0.1:17870 +--idunn-daemon stonks +--idunn-health-contract stonks.cultnet-rudp-market-health +``` + +The `/health` endpoint remains a compatibility/status projection. It is not the +owner of Stonks daemon liveness once Idunn has fresh RUDP health. diff --git a/docs/architecture.md b/docs/architecture.md index 3f3f6e7..f01a95b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -16,6 +16,11 @@ Public market endpoints -> /eve/deck WebSocket + /eve/deck/providers manifest -> Odin interface ingestion -> Eve / Nightwing / agent projection + +Stonks refresh completion + -> idunn.daemon_health raw CultNet document + -> cultnet.transport.rudp.v0 schema channel + -> Idunn keepalive store and decision cycle ``` ## Invariants @@ -27,6 +32,11 @@ Public market endpoints - Odin owns discovery and aggregation, not market truth. - Renderers lower the Stonks surface only. - Stonks does not trade, advise, custody funds, or infer portfolio intent. +- The `/health`, `/market/state`, and `/eve/deck` HTTP/WebSocket surfaces are + compatibility projections. Idunn liveness is daemon-published RUDP health, + not an Odin-owned curl probe. +- Refreshes are serialized because each refresh publishes daemon health after + updating the market snapshot. ## Future Cut diff --git a/package-lock.json b/package-lock.json index b7a39ec..1357731 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9,7 +9,8 @@ "version": "0.1.0", "license": "MIT", "dependencies": { - "cultcache-ts": "file:../CultLib/packages/cultcache-ts" + "cultcache-ts": "file:../CultLib/packages/cultcache-ts", + "cultnet-ts": "file:../CultLib/packages/cultnet-ts" }, "engines": { "node": ">=20" @@ -23,25 +24,32 @@ "zod": "^3.25.76" }, "devDependencies": { - "@gamecult/norn-viewer": "file:../Norn/web/norn-viewer", "@types/node": "^24.0.0", - "@types/react": "^19.2.14", - "@types/react-dom": "^19.2.3", - "@vitejs/plugin-react": "^6.0.2", - "concurrently": "^9.2.1", - "cross-env": "^10.1.0", - "electron": "^42.1.0", - "electron-builder": "^26.8.1", - "react": "^19.2.6", - "react-dom": "^19.2.6", + "typescript": "^5.9.0" + } + }, + "../CultLib/packages/cultnet-ts": { + "version": "0.1.0", + "license": "MIT", + "dependencies": { + "@msgpack/msgpack": "^3.1.3", + "ajv": "^8.17.1", + "cultcache-ts": "file:../cultcache-ts" + }, + "devDependencies": { + "@types/node": "^24.0.0", + "json-schema-to-typescript": "^15.0.4", "typescript": "^5.9.0", - "vite": "^8.0.13", - "wait-on": "^9.0.10" + "zod": "^3.25.76" } }, "node_modules/cultcache-ts": { "resolved": "../CultLib/packages/cultcache-ts", "link": true + }, + "node_modules/cultnet-ts": { + "resolved": "../CultLib/packages/cultnet-ts", + "link": true } } } diff --git a/package.json b/package.json index 12bb5ba..b08fd9b 100644 --- a/package.json +++ b/package.json @@ -9,7 +9,8 @@ "start": "node src/stonks-daemon.cjs" }, "dependencies": { - "cultcache-ts": "file:../CultLib/packages/cultcache-ts" + "cultcache-ts": "file:../CultLib/packages/cultcache-ts", + "cultnet-ts": "file:../CultLib/packages/cultnet-ts" }, "engines": { "node": ">=20" diff --git a/scripts/start-stonks.ps1 b/scripts/start-stonks.ps1 index fdf1b95..853dfce 100644 --- a/scripts/start-stonks.ps1 +++ b/scripts/start-stonks.ps1 @@ -2,6 +2,9 @@ param( [int] $Port = 8802, [string] $StateDir = "E:\Projects\Stonks\scratch\stonks", [string] $FinnhubTokenFile = "E:\Projects\Stonks\finnhub-oauth.txt", + [string] $IdunnRudpHealth = "127.0.0.1:17870", + [string] $IdunnDaemon = "stonks", + [string] $IdunnHealthContract = "stonks.cultnet-rudp-market-health", [switch] $Foreground ) @@ -32,7 +35,10 @@ $args = @( $scriptPath, "--port", "$Port", "--stateDir", $StateDir, - "--finnhubTokenFile", $FinnhubTokenFile + "--finnhubTokenFile", $FinnhubTokenFile, + "--idunn-rudp-health", $IdunnRudpHealth, + "--idunn-daemon", $IdunnDaemon, + "--idunn-health-contract", $IdunnHealthContract ) if ($Foreground) { diff --git a/src/idunn-rudp.cjs b/src/idunn-rudp.cjs new file mode 100644 index 0000000..1982076 --- /dev/null +++ b/src/idunn-rudp.cjs @@ -0,0 +1,232 @@ +"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) 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: `stonks-health:${publisher.daemonId}:${observedAt.replace(/[:.]/g, "-")}`, + document: { + schemaId: "idunn.daemon_health", + recordKey: publisher.daemonId, + storedAt: observedAt, + payloadEncoding: "messagepack", + payload, + sourceRuntimeId: "stonks-daemon", + sourceRole: "daemon-health-publisher", + tags: [CULTNET_RUDP_PROTOCOL_ID], + }, + }; + const wirePayload = encode(encodeCultNetMessageForWire(message, "cultnet.schema.v0")); + const packets = 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 packets) { + 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("Stonks 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("Stonks 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 = { + createIdunnRudpHealthPublisher, + publishIdunnRudpHealth, +}; diff --git a/src/stonks-daemon.cjs b/src/stonks-daemon.cjs index 746f9b6..3f102c9 100644 --- a/src/stonks-daemon.cjs +++ b/src/stonks-daemon.cjs @@ -12,6 +12,7 @@ const { defineDocumentRegistry, defineDocumentType, } = require("cultcache-ts"); +const { createIdunnRudpHealthPublisher, publishIdunnRudpHealth } = require("./idunn-rudp.cjs"); const repoRoot = path.resolve(__dirname, ".."); const args = parseArgs(process.argv.slice(2)); @@ -24,10 +25,19 @@ const finnhubTokenFile = args.finnhubTokenFile || process.env.STONKS_FINNHUB_TOK const finnhubToken = String(args.finnhubToken || process.env.FINNHUB_API_KEY || process.env.STONKS_FINNHUB_TOKEN || readSecretFile(finnhubTokenFile)).trim(); const equityCallsPerMinute = clampNumber(args.equityCallsPerMinute || process.env.STONKS_EQUITY_CALLS_PER_MINUTE || 48, 1, 60); const mentionRefreshMs = clampNumber(args.mentionRefreshMs || process.env.STONKS_MENTION_REFRESH_MS || 600000, 60000, 3600000); +const idunnRudpHealth = args["idunn-rudp-health"] || process.env.STONKS_IDUNN_RUDP_HEALTH; +const idunnDaemon = args["idunn-daemon"] || process.env.STONKS_IDUNN_DAEMON || "stonks"; +const idunnHealthContract = args["idunn-health-contract"] || process.env.STONKS_IDUNN_HEALTH_CONTRACT || "stonks.cultnet-rudp-market-health"; +const idunnRudpHealthPublisher = createIdunnRudpHealthPublisher(idunnRudpHealth ? { + endpoint: idunnRudpHealth, + daemonId: idunnDaemon, + healthContract: idunnHealthContract, +} : null); const providerId = "stonks.market"; const clients = new Set(); const recentRequests = []; const pendingCultCacheWrites = new Set(); +let lastIdunnRudpHealthPublishedAt = null; const equitySymbols = String(args.equities || process.env.STONKS_EQUITIES || "ubi.fr,ea.us,ttwo.us,rblx.us,ntdoy.us,sony.us,msft.us,nvda.us,amd.us,googl.us,meta.us,aapl.us,tsla.us,tsm.us,asml.us,crsr.us,logi.us,se.us") .split(",") @@ -185,8 +195,14 @@ async function main() { }); await refresh(); - setInterval(() => { - refresh().catch((error) => console.error("refresh failed:", error)); + scheduleRefresh(); +} + +function scheduleRefresh() { + setTimeout(() => { + refresh() + .catch((error) => console.error("refresh failed:", error)) + .finally(scheduleRefresh); }, intervalMs); } @@ -215,10 +231,45 @@ async function pullCultCacheOrQuarantine() { } async function refresh() { - latestSnapshot = await marketSnapshot(); - currentState = buildState(latestSnapshot); - persistSnapshot(latestSnapshot, currentState); - broadcast(currentState); + try { + latestSnapshot = await marketSnapshot(); + currentState = buildState(latestSnapshot); + persistSnapshot(latestSnapshot, currentState); + broadcast(currentState); + await publishStonksHealth("active", stonksHealthDetail(latestSnapshot)); + } catch (error) { + await publishStonksHealth("failed", `Stonks refresh failed: ${error.message}`); + throw error; + } +} + +async function publishStonksHealth(state, detail) { + if (!idunnRudpHealthPublisher) return; + try { + await publishIdunnRudpHealth(idunnRudpHealthPublisher, { + state, + detail, + observedAt: new Date().toISOString(), + }); + lastIdunnRudpHealthPublishedAt = Date.now(); + } catch (error) { + const lastPublishedAgeMs = lastIdunnRudpHealthPublishedAt === null + ? Number.POSITIVE_INFINITY + : Date.now() - lastIdunnRudpHealthPublishedAt; + if (lastPublishedAgeMs > Math.max(60_000, intervalMs * 4)) { + console.error("Idunn RUDP health publish failed:", error.message); + } + } +} + +function stonksHealthDetail(snapshot) { + return [ + `Stonks refreshed market snapshot in ${snapshot.durationMs}ms`, + `equities=${snapshot.equities.length}/${equitySymbols.length}`, + `crypto=${snapshot.crypto.length}/${cryptoIds.length}`, + `finnhub=${snapshot.sources.equities.ok ? "ok" : "degraded"}`, + `coingecko=${snapshot.sources.crypto.ok ? "ok" : "degraded"}`, + ].join("; "); } async function marketSnapshot() {