Publish Stonks health over Idunn RUDP

This commit is contained in:
Meta Vens 2026-06-16 04:30:56 +01:00
commit 69729cfbab
7 changed files with 343 additions and 21 deletions

View file

@ -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.

View file

@ -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

34
package-lock.json generated
View file

@ -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
}
}
}

View file

@ -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"

View file

@ -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) {

232
src/idunn-rudp.cjs Normal file
View file

@ -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,
};

View file

@ -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() {