jschoubben
f0104b7846
Every module that dialled the new bus failed its handshake with "wrong version number": the runtime pinned the server's certificate by a raw TLS connection to a port on which the server speaks first, in the clear. The pin is taken after the INFO line now, on the same socket, and then the real connection verifies against exactly that certificate. And the old transport is deleted — its client, its tests, its dependency — with the wire-compatibility pins that only existed for the move (novox/hq ADR 0131, design 28 task 5.5). A credential names the bus, and there is one.
208 lines
9.1 KiB
TypeScript
208 lines
9.1 KiB
TypeScript
// The runnable entrypoint. Three modes:
|
|
//
|
|
// mesh-tools serve — bind the broker and serve the assigned modules until
|
|
// stopped. A module entrypoint that subscribes to events (on("#"))
|
|
// starts consuming as it is imported, so this also runs consumers.
|
|
// mesh-tools emit TYPE [JSON] emit one event onto the mesh and exit — an operable primitive,
|
|
// and what an events test uses to put a message on the wire.
|
|
// mesh-tools run ENTRYPOINT run one compiled module entrypoint to completion and exit — the
|
|
// runtime side of a run-once step (novox/hq ADR 0052). It imports
|
|
// the given entrypoint, whose top-level code does its work — seed a
|
|
// store, migrate, health-gate — and awaits it. It does NOT connect
|
|
// to the broker: a first-boot step runs offline, before the module
|
|
// has anything to talk to, and the host gates the container that
|
|
// depends on it on this process exiting 0.
|
|
//
|
|
// The broker, in order of preference:
|
|
// MESH_BROKER_FILE a sealed {url, fingerprint} the mesh delivered (novox/hq ADR 0043) — an
|
|
// amqps account scoped to this module. Preferred: a module holds its own.
|
|
// MESH_BROKER_URL a plain URL, for the bootstrap/admin case before a module has an account.
|
|
// MESH_TOOL_MODULES /path/a,/path/b,… compiled module entrypoints (serve mode)
|
|
// MESH_MODULE / MESH_NODE the identity stamped onto emitted events (ADR 0042)
|
|
|
|
import { readFileSync } from "node:fs";
|
|
import { pathToFileURL } from "node:url";
|
|
import { connectNats, fatalBrokerReason as fatalNatsReason, type Credential } from "./broker-nats.js";
|
|
import { runTools } from "./runtime.js";
|
|
import { invokeTool } from "@novox/mesh-sdk/tools";
|
|
import { useBroker } from "@novox/mesh-sdk/messaging";
|
|
import type { Broker } from "@novox/mesh-sdk/messaging";
|
|
import { emit } from "@novox/mesh-sdk/events";
|
|
|
|
/**
|
|
* Connect the way this process is meant to: with its sealed credential if the mesh gave it one, and
|
|
* over the plain bootstrap URL otherwise. A scoped module assumes the foundation's exchanges exist —
|
|
* its account may not declare them (ADR 0043).
|
|
*/
|
|
// fatalBrokerReasonFor is the reason a connection failure is final rather than "not yet", for
|
|
// whichever bus this runtime is on — each transport knows its own refusals.
|
|
function fatalBrokerReasonFor(err: unknown): string | null {
|
|
return fatalNatsReason(err);
|
|
}
|
|
|
|
async function connectBroker(): Promise<Broker> {
|
|
const file = process.env.MESH_BROKER_FILE;
|
|
if (file) {
|
|
let credential: Credential;
|
|
try {
|
|
credential = JSON.parse(readFileSync(file, "utf8")) as Credential;
|
|
} catch (err) {
|
|
console.error(`mesh-tools: cannot read the broker credential at ${file}: ${err}`);
|
|
process.exit(1);
|
|
}
|
|
if (!credential.url) {
|
|
console.error(`mesh-tools: ${file} carries no url — it is not a broker credential`);
|
|
process.exit(1);
|
|
}
|
|
// The mesh scoped this account to a node and module; take the runtime's identity from the
|
|
// credential so its queue and the events it emits match what the mesh authorised, no matter
|
|
// what the environment says.
|
|
if (credential.node) process.env.MESH_NODE = credential.node;
|
|
if (credential.module) process.env.MESH_MODULE = credential.module;
|
|
// **The credential names the bus.** A module moved to the bus being built was handed a
|
|
// credential for it — `nats://…` with user, password and fingerprint beside the address — and
|
|
// nothing else in its environment changed (design 25; novox/hq design 28 task 5.2). The scheme
|
|
// is enough to know which bus to speak; a runtime that always dialled the old one would keep
|
|
// serving and answer nobody.
|
|
// One bus (novox/hq ADR 0131, design 28 task 5.5): the credential names it, and it is this.
|
|
return connectNats(credential);
|
|
}
|
|
const url = process.env.MESH_BROKER_URL;
|
|
if (!url) {
|
|
console.error(
|
|
"mesh-tools: set MESH_BROKER_FILE (a sealed credential) or MESH_BROKER_URL — there is no broker to reach",
|
|
);
|
|
process.exit(1);
|
|
}
|
|
return connectNats({ url });
|
|
}
|
|
|
|
/**
|
|
* Connect for serve mode, retrying while the broker is merely not reachable yet. At startup that
|
|
* is the NORMAL case, not a failure: a container comes up in seconds and the overlay tunnel a
|
|
* moment later (novox/hq issue 058). Exiting instead delegated the retry to the container
|
|
* runtime, which read as a crash-loop to every restart-counting health check and every person
|
|
* watching. Retried indefinitely, aloud: the dependency appears or somebody reads why not.
|
|
*
|
|
* A failure that waiting cannot fix (see fatalBrokerReason) is thrown at once rather than retried —
|
|
* a permanent fault masquerading as "not reachable yet" is the silent non-progress this whole
|
|
* change exists to remove. Missing-file and empty-URL configuration errors exit inside
|
|
* connectBroker before they reach here; a malformed URL and a refused login are caught here.
|
|
*/
|
|
async function connectBrokerPatiently(): Promise<Broker> {
|
|
for (let delay = 2_000; ; delay = Math.min(delay * 2, 30_000)) {
|
|
try {
|
|
return await connectBroker();
|
|
} catch (err) {
|
|
const fatal = fatalBrokerReasonFor(err);
|
|
if (fatal !== null) {
|
|
console.error(`mesh-tools: ${fatal} — waiting will not fix this; giving up`);
|
|
throw err;
|
|
}
|
|
const why = err instanceof Error ? err.message : String(err);
|
|
// A little jitter so every module that was up when the broker bounced does not retry in
|
|
// lockstep and stampede it as it recovers.
|
|
const wait = delay + Math.floor(Math.random() * 1_000);
|
|
console.error(`mesh-tools: the broker is not reachable yet (${why}); retrying in ${Math.round(wait / 1000)}s`);
|
|
await new Promise((r) => setTimeout(r, wait));
|
|
}
|
|
}
|
|
}
|
|
|
|
async function serve(): Promise<void> {
|
|
const moduleEntrypoints = (process.env.MESH_TOOL_MODULES ?? "")
|
|
.split(",")
|
|
.map((s) => s.trim())
|
|
.filter(Boolean);
|
|
|
|
const broker = await connectBrokerPatiently();
|
|
const stop = await runTools({ broker, moduleEntrypoints });
|
|
|
|
const shutdown = async (): Promise<void> => {
|
|
stop();
|
|
await broker.close();
|
|
process.exit(0);
|
|
};
|
|
process.on("SIGTERM", () => void shutdown());
|
|
process.on("SIGINT", () => void shutdown());
|
|
}
|
|
|
|
async function emitOnce(type: string, bodyJson: string): Promise<void> {
|
|
let body: unknown = {};
|
|
if (bodyJson) {
|
|
try {
|
|
body = JSON.parse(bodyJson);
|
|
} catch {
|
|
console.error(`mesh-tools emit: body is not JSON: ${bodyJson}`);
|
|
process.exit(1);
|
|
}
|
|
}
|
|
const broker = await connectBroker();
|
|
useBroker(() => broker);
|
|
// emit awaits the broker's publish confirm (ADR 0042), so the event is accepted before we close.
|
|
await emit(type, body);
|
|
await broker.close();
|
|
}
|
|
|
|
async function invokeOnce(module: string, tool: string, argsJson: string): Promise<void> {
|
|
let args: Record<string, unknown> = {};
|
|
if (argsJson) {
|
|
try {
|
|
args = JSON.parse(argsJson) as Record<string, unknown>;
|
|
} catch {
|
|
console.error(`mesh-tools invoke: args are not JSON: ${argsJson}`);
|
|
process.exit(1);
|
|
}
|
|
}
|
|
const broker = await connectBroker();
|
|
const result = await invokeTool(broker, module, tool, args);
|
|
process.stdout.write(JSON.stringify(result) + "\n");
|
|
await broker.close();
|
|
}
|
|
|
|
/**
|
|
* Run one compiled module entrypoint to completion — the runtime side of a run-once step
|
|
* (novox/hq ADR 0052). Importing it runs its top-level code and awaits any top-level await, so this
|
|
* returns only once the step's own code has finished; a step that throws rejects here and the
|
|
* process exits non-zero, which is how the host knows the step did not complete and must not start
|
|
* the container it gates. No broker is connected — a first-boot seed or migration runs offline.
|
|
*/
|
|
async function runEntry(entrypoint: string): Promise<void> {
|
|
if (!entrypoint) {
|
|
console.error("mesh-tools run <entrypoint> — a compiled module entrypoint path is required");
|
|
process.exit(1);
|
|
}
|
|
// A file URL, not a bare path: dynamic import of an absolute path is not portable, and the
|
|
// entrypoint the manifest names is an absolute path inside the image.
|
|
await import(pathToFileURL(entrypoint).href);
|
|
}
|
|
|
|
async function main(): Promise<void> {
|
|
const [command, ...rest] = process.argv.slice(2);
|
|
if (command === "run") {
|
|
await runEntry(rest[0] ?? "");
|
|
return;
|
|
}
|
|
if (command === "invoke") {
|
|
const [module, tool] = rest;
|
|
if (!module || !tool) {
|
|
console.error("mesh-tools invoke <module> <tool> [json-args] — a module and tool are required");
|
|
process.exit(1);
|
|
}
|
|
await invokeOnce(module, tool, rest[2] ?? "");
|
|
return;
|
|
}
|
|
if (command === "emit") {
|
|
const type = rest[0];
|
|
if (!type) {
|
|
console.error("mesh-tools emit <type> [json-body] — a routing key is required");
|
|
process.exit(1);
|
|
}
|
|
await emitOnce(type, rest[1] ?? "");
|
|
return;
|
|
}
|
|
await serve();
|
|
}
|
|
|
|
void main();
|