commit f13e2fbf68d9eb173bfdb62da3dc300d3c1f5fdd Author: George Lambert Date: Fri Sep 11 15:00:08 2026 -0400 Initial import of verae-fleet from zapier monorepo diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e094d00 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +data/ +machines.secrets.json + diff --git a/NATS.md b/NATS.md new file mode 100644 index 0000000..73525c7 --- /dev/null +++ b/NATS.md @@ -0,0 +1,15 @@ +# NATS — verae-fleet + +Fleet control **does not subscribe** to JetStream. It only records which workers should. + +| Service | NATS in | NATS out | +|---------|---------|----------| +| nats (external) | — | — (the broker) | +| middleware-http | jobs.events | jobs.watch | +| job-poller | jobs.watch | jobs.events | +| webhook-deliver | webhooks.deliver, jobs.events | HTTPS to Zapier | +| archive-aggregator | archive.reply.* | archive.query (broadcast) | +| archive-worm | archive.put, archive.query | archive.reply.* on bloom hit | +| tree-node | archive.put, archive.query | archive.reply.* on bloom hit | + +NS1 nats-server stays on `127.0.0.1:4222`. `fleet.json` `nats.publicBind` is false. diff --git a/README.md b/README.md new file mode 100644 index 0000000..aae377a --- /dev/null +++ b/README.md @@ -0,0 +1,44 @@ +# verae-fleet + +Operator control plane for Verae Time × Zapier **runtime** services: + +- a **list** of every service and its config file +- a **central replica spec** (`fleet.json`) — how many copies must be up +- a **monitor** of active / paused / unhealthy replicas +- **restart** when a copy is offline or fails `/health` +- **on / off / pause / resume** per service or per instance +- **keepFloor** on tree nodes so at least `min` copies stay available (paused copies do not count) + +```bash +cd packages/verae-fleet +npm test +node src/cli.js list +node src/cli.js serve # http://127.0.0.1:3850/ +``` + +Against a running daemon: + +```bash +node src/cli.js status +node src/cli.js pause tree-node-0 # floor starts another tree-node +node src/cli.js resume tree-node-0 +node src/cli.js restart tree-node-1 +node src/cli.js stop webhook-deliver # disable that service +node src/cli.js start webhook-deliver +``` + +Add capacity in `machines.json` or the monitor **Add machine** form (`kind=ssh`, user, host, identity file path). New replicas land on the least-loaded eligible host. + +```bash +node src/cli.js ssh-check ns1 # marchon@70.88.205.138 with ~/.ssh/id_ed25519 +``` + +Private keys stay on disk (`~/.ssh/id_ed25519`); git stores only the path. Optional overrides: `machines.secrets.json` (gitignored). + +Operator console (Fleet · Trace · Docs) at http://127.0.0.1:3850/ — [docs/CONSOLE.md](docs/CONSOLE.md) · [CONSOLE.pdf](docs/CONSOLE.pdf) (also https://zapier.georgelambert.org/packages/verae-fleet/docs/CONSOLE.pdf). Message-processing **min / avg / p50 / p90** RTT is on the Fleet tab. + +Zapier cloud apps are listed but **not spawned**. NATS on NS1 is **monitored only** (loopback `:4222`, never a public bind). + +Install on Docker / Proxmox / metal: https://zapier.georgelambert.org/packages/verae-ops/README.pdf + +Clone: `ssh://git@git.georgelambert.org:2223/marchon/verae-fleet.git` diff --git a/SERVICES.md b/SERVICES.md new file mode 100644 index 0000000..b2f9215 --- /dev/null +++ b/SERVICES.md @@ -0,0 +1,24 @@ +# Service list + +Central replica spec: [`fleet.json`](fleet.json). +Per-service files: [`services/`](services/). + +| Service | Config | min | max | keepFloor | Spawned by fleet | +|---------|--------|-----|-----|-----------|------------------| +| nats | `services/nats.json` | 1 | 1 | no | no (external monitor) | +| zappier-edge | `services/zappier-edge.json` | 1 | 1 | yes | yes | +| middleware-http | `services/middleware-http.json` | 1 | 1 | yes | yes | +| job-poller | `services/job-poller.json` | 1 | 2 | yes | yes | +| webhook-deliver | `services/webhook-deliver.json` | 1 | 2 | yes | yes | +| archive-aggregator | `services/archive-aggregator.json` | 1 | 2 | yes | yes | +| archive-worm | `services/archive-worm.json` | 3 | 6 | yes | yes | +| **tree-node** | `services/tree-node.json` | **3** | 9 | **yes** | yes | +| zapier-simulator | `services/zapier-simulator.json` | 0 | 1 | no | optional | +| zapier-platform-app | `services/unmanaged.json` | — | — | — | no (Zapier cloud) | +| verae-activate | `services/unmanaged.json` | — | — | — | no | +| request-splitter | `services/unmanaged.json` | — | — | — | no (library) | +| verae-chain-client | `services/unmanaged.json` | — | — | — | no (library) | + +`keepFloor`: monitor starts replacements when **available** (running, healthy, not paused) drops below `min`. That is how tree nodes stay at three live copies if one is paused, crashed, or not responding. + +Hosts: [`machines.json`](machines.json). Add machines in the monitor UI or POST `/api/machines`. Placement is least-loaded eligible host. Remote: `node src/agent.js`. diff --git a/SUMMARY.md b/SUMMARY.md new file mode 100644 index 0000000..162fd11 --- /dev/null +++ b/SUMMARY.md @@ -0,0 +1,9 @@ +# verae-fleet + +**Job:** Catalog services, hold per-service configs, monitor health, restart bad replicas, pause/resume/on/off, and keep `fleet.json` replica floors — especially tree-node `min`. + +**Expects:** operator CLI/HTTP on `127.0.0.1:3850`. Spawns `src/worker.js` children with `/health`. + +**Sends:** nothing to Zapier or public NATS. + +**Test:** `npm test` — list configs; start 3 tree-nodes; pause one still ≥3 available; unhealthy restart; stop+replace. diff --git a/docs/CONSOLE.html b/docs/CONSOLE.html new file mode 100644 index 0000000..61778c2 --- /dev/null +++ b/docs/CONSOLE.html @@ -0,0 +1,426 @@ + + + + + + + Operator console + + + + +
Verae Time × Zapier

Operator console

packages/verae-fleet/docs/CONSOLE.md
+
+

Operator console

+
+

Operator console

+

Loopback UI at http://127.0.0.1:3850/ — Fleet, Trace +(simulator), and Docs in one shell. Not a public site. NATS stays +private; SSH identity files stay on disk.

+
cd packages/verae-fleet
+node src/cli.js serve
+

Tabs

+ ++++ + + + + + + + + + + + + + + + + + + + + +
TabPurpose
FleetMachines (SSH/local), replica floors, green/yellow/red, RTT +min/avg/p50/p90 spark, pause/restart/off with confirm, ssh-check +toast
TraceIn-process Zapier simulator: run a step, hop log, +monitors/suggestions
DocsReading order: overview → module summaries → architecture → +remaining docs
+

Screenshots

+
+Fleet + +
+
+Trace + +
+
+Docs + +
+

Design tokens match the zappier portal/admin (indigo shell). Customer +portal is unchanged.

+

See overview.

+ + diff --git a/docs/CONSOLE.md b/docs/CONSOLE.md new file mode 100644 index 0000000..42816e9 --- /dev/null +++ b/docs/CONSOLE.md @@ -0,0 +1,28 @@ +# Operator console + +Loopback UI at **http://127.0.0.1:3850/** — Fleet, Trace (simulator), and Docs in one shell. Not a public site. NATS stays private; SSH identity files stay on disk. + +```bash +cd packages/verae-fleet +node src/cli.js serve +``` + +## Tabs + +| Tab | Purpose | +|-----|---------| +| **Fleet** | Machines (SSH/local), replica floors, green/yellow/red, RTT min/avg/p50/p90 spark, pause/restart/off with confirm, ssh-check toast | +| **Trace** | In-process Zapier simulator: run a step, hop log, monitors/suggestions | +| **Docs** | Reading order: overview → module summaries → architecture → remaining docs | + +## Screenshots + +![Fleet](screenshots/console-fleet.png) + +![Trace](screenshots/console-trace.png) + +![Docs](screenshots/console-docs.png) + +Design tokens match the zappier portal/admin (indigo shell). Customer portal is unchanged. + +See [overview](https://git.georgelambert.org/marchon/overview). diff --git a/docs/CONSOLE.pdf b/docs/CONSOLE.pdf new file mode 100644 index 0000000..f1d1b27 Binary files /dev/null and b/docs/CONSOLE.pdf differ diff --git a/docs/screenshots/console-docs.png b/docs/screenshots/console-docs.png new file mode 100644 index 0000000..975c6a1 Binary files /dev/null and b/docs/screenshots/console-docs.png differ diff --git a/docs/screenshots/console-fleet.png b/docs/screenshots/console-fleet.png new file mode 100644 index 0000000..a599cb8 Binary files /dev/null and b/docs/screenshots/console-fleet.png differ diff --git a/docs/screenshots/console-trace.png b/docs/screenshots/console-trace.png new file mode 100644 index 0000000..9915079 Binary files /dev/null and b/docs/screenshots/console-trace.png differ diff --git a/fleet.json b/fleet.json new file mode 100644 index 0000000..afc9c94 --- /dev/null +++ b/fleet.json @@ -0,0 +1,26 @@ +{ + "apiVersion": "verae.fleet/v1", + "control": { + "bind": "127.0.0.1", + "port": 3850, + "probeIntervalMs": 750, + "unhealthyAfterMs": 2500, + "restartBackoffMs": [200, 800, 2000], + "maxRestartsPerHour": 40 + }, + "nats": { + "url": "nats://127.0.0.1:4222", + "publicBind": false + }, + "services": { + "nats": { "min": 1, "max": 1, "keepFloor": false, "enabled": true, "managed": false }, + "zappier-edge": { "min": 1, "max": 1, "keepFloor": true, "enabled": true }, + "middleware-http": { "min": 1, "max": 1, "keepFloor": true, "enabled": true }, + "job-poller": { "min": 1, "max": 2, "keepFloor": true, "enabled": true }, + "webhook-deliver": { "min": 1, "max": 2, "keepFloor": true, "enabled": true }, + "archive-aggregator": { "min": 1, "max": 2, "keepFloor": true, "enabled": true }, + "archive-worm": { "min": 3, "max": 6, "keepFloor": true, "enabled": true }, + "tree-node": { "min": 3, "max": 9, "keepFloor": true, "enabled": true }, + "zapier-simulator": { "min": 0, "max": 1, "keepFloor": false, "enabled": true } + } +} diff --git a/machines.json b/machines.json new file mode 100644 index 0000000..dd47bcc --- /dev/null +++ b/machines.json @@ -0,0 +1,47 @@ +{ + "machines": [ + { + "id": "local", + "title": "Control plane (this host)", + "kind": "local", + "host": "127.0.0.1", + "enabled": true, + "capacity": 32, + "roles": ["*"] + }, + { + "id": "ns1", + "title": "NS1.GEORGELAMBERT.ORG", + "kind": "ssh", + "host": "70.88.205.138", + "user": "marchon", + "sshPort": 22, + "identityFile": "~/.ssh/id_ed25519", + "remoteDir": "~/verae-fleet-runtime", + "enabled": true, + "capacity": 24, + "roles": [ + "tree-node", + "archive-worm", + "archive-aggregator", + "job-poller", + "webhook-deliver" + ], + "notes": "SSH as marchon with ~/.ssh/id_ed25519. Workers bind 127.0.0.1 on NS1; control plane probes via ssh+curl. Private key is never stored in git." + }, + { + "id": "lan-134", + "title": "70.88.205.134", + "kind": "ssh", + "host": "70.88.205.134", + "user": "marchon", + "sshPort": 22, + "identityFile": "~/.ssh/id_ed25519", + "remoteDir": "~/verae-fleet-runtime", + "enabled": false, + "capacity": 12, + "roles": ["tree-node", "archive-worm"], + "notes": "Same user/key as NS1. Enable after confirming node is on PATH." + } + ] +} diff --git a/machines.secrets.json.example b/machines.secrets.json.example new file mode 100644 index 0000000..474231d --- /dev/null +++ b/machines.secrets.json.example @@ -0,0 +1,13 @@ +{ + "ns1": { + "user": "marchon", + "host": "70.88.205.138", + "sshPort": 22, + "identityFile": "~/.ssh/id_ed25519" + }, + "lan-134": { + "user": "marchon", + "host": "70.88.205.134", + "identityFile": "~/.ssh/id_ed25519" + } +} diff --git a/package.json b/package.json new file mode 100644 index 0000000..6e77001 --- /dev/null +++ b/package.json @@ -0,0 +1,13 @@ +{ + "name": "verae-fleet", + "version": "0.1.0", + "type": "module", + "description": "Service catalog, replica floors, monitor, restart, pause/resume for Verae Zapier workers and tree nodes", + "bin": { "verae-fleet": "./src/cli.js" }, + "scripts": { + "start": "node src/cli.js serve", + "list": "node src/cli.js list", + "test": "node --test test/*.test.js" + }, + "engines": { "node": ">=20" } +} diff --git a/public/index.html b/public/index.html new file mode 100644 index 0000000..f193e6c --- /dev/null +++ b/public/index.html @@ -0,0 +1,338 @@ + + + + + + Verae operator console + + + +
+
+
loopback · not public NATS
+

Verae operator console

+

Fleet, message trace, and docs in one shell. Pause does not count toward replica floors. SSH keys stay on disk.

+
+ +
+
+
+

+

Machines

+
+ + + + + + + + + +
+
+

Services

+
+
+
+
+ +
+

Response

+ 
Run a step.
+

Hop trace

+
+

Monitors

+
+
+
+
+
+

Reading order: high-level first, then architecture, then the rest. Catalog defaults to colored PDFs. Operator console is loopback only.

+

Open CONSOLE.pdf + · PDF catalog + · Markdown indexes

+

High-level (start here)

+ +

Architecture

+ +

Remaining documentation

+ +
+
+
+ + + diff --git a/services/archive-aggregator.json b/services/archive-aggregator.json new file mode 100644 index 0000000..2c024d2 --- /dev/null +++ b/services/archive-aggregator.json @@ -0,0 +1,16 @@ +{ + "id": "archive-aggregator", + "title": "Archive reply aggregator", + "kind": "nats-worker", + "package": "verae-archive-aggregator", + "role": "archive-aggregator", + "managed": true, + "runtime": "query fan-out + merge", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13400 }, + "nats": { + "in": ["verae.archive.reply.*"], + "out": ["verae.archive.query"] + }, + "notes": "Broadcast archive.query (no queue group). kinds may include tree." +} diff --git a/services/archive-worm.json b/services/archive-worm.json new file mode 100644 index 0000000..dd3360c --- /dev/null +++ b/services/archive-worm.json @@ -0,0 +1,16 @@ +{ + "id": "archive-worm", + "title": "Bloom-filtered WORM archive", + "kind": "nats-archive", + "package": "verae-archive-worm", + "role": "archive-worm", + "managed": true, + "runtime": "N copies, bloom miss = silence", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13500 }, + "nats": { + "in": ["verae.archive.put", "verae.archive.query"], + "out": ["verae.archive.reply."] + }, + "notes": "Subscribe to query without a shared queue group." +} diff --git a/services/job-poller.json b/services/job-poller.json new file mode 100644 index 0000000..0e78df1 --- /dev/null +++ b/services/job-poller.json @@ -0,0 +1,16 @@ +{ + "id": "job-poller", + "title": "Job poller worker", + "kind": "nats-worker", + "package": "verae-zapier-middleware", + "role": "job-poller", + "managed": true, + "runtime": "JetStream consumer", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13200 }, + "nats": { + "in": ["verae.zapier.jobs.watch"], + "out": ["verae.zapier.jobs.events"] + }, + "notes": "GET chain status; emit terminal events. Queue group job-poller." +} diff --git a/services/middleware-http.json b/services/middleware-http.json new file mode 100644 index 0000000..5b819d2 --- /dev/null +++ b/services/middleware-http.json @@ -0,0 +1,17 @@ +{ + "id": "middleware-http", + "title": "Verae Zapier HTTP middleware", + "kind": "http", + "package": "verae-zapier-middleware", + "role": "middleware-http", + "managed": true, + "runtime": "HTTPS :3100 /zapier/v1", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13100 }, + "env": { "PORT": "3100", "NATS_URL": "nats://127.0.0.1:4222" }, + "nats": { + "in": ["verae.zapier.jobs.events"], + "out": ["verae.zapier.jobs.watch", "verae.zapier.webhooks.deliver"] + }, + "notes": "Wait path subscribes jobs.events. Never expose NATS to Zapier." +} diff --git a/services/nats.json b/services/nats.json new file mode 100644 index 0000000..60dcb37 --- /dev/null +++ b/services/nats.json @@ -0,0 +1,11 @@ +{ + "id": "nats", + "title": "NATS JetStream (NS1 loopback)", + "kind": "external", + "package": null, + "managed": false, + "runtime": "nats-server -js on 127.0.0.1:4222", + "health": { "type": "tcp", "host": "127.0.0.1", "port": 4222, "timeoutMs": 400 }, + "nats": { "in": [], "out": [] }, + "notes": "Do not bind 0.0.0.0. Zapier never connects here. Fleet monitors only; does not spawn nats-server." +} diff --git a/services/tree-node.json b/services/tree-node.json new file mode 100644 index 0000000..0c85964 --- /dev/null +++ b/services/tree-node.json @@ -0,0 +1,16 @@ +{ + "id": "tree-node", + "title": "Merkle tree-node archive", + "kind": "nats-archive", + "package": "verae-tree-node", + "role": "tree-node", + "managed": true, + "runtime": "N copies holding leaf proofs for bulk summaries", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13600 }, + "nats": { + "in": ["verae.archive.put", "verae.archive.query"], + "out": ["verae.archive.reply."] + }, + "notes": "keepFloor in fleet.json must stay >= 3 so a leaf query still has a node that might hold the shard. Bloom miss = no packet." +} diff --git a/services/unmanaged.json b/services/unmanaged.json new file mode 100644 index 0000000..df7e1bf --- /dev/null +++ b/services/unmanaged.json @@ -0,0 +1,37 @@ +{ + "id": "_unmanaged", + "title": "Not fleet-spawned (listed for operators)", + "entries": [ + { + "id": "zapier-platform-app", + "package": "verae-zapier", + "runtime": "Zapier cloud", + "notes": "HTTPS only to zappier-edge. Never NATS." + }, + { + "id": "verae-activate", + "package": "verae-activate", + "runtime": "Zapier cloud (activate-now app)" + }, + { + "id": "request-splitter", + "package": "verae-request-splitter", + "runtime": "library in middleware" + }, + { + "id": "verae-chain-client", + "package": "verae-zapier-middleware", + "runtime": "library → api.veraetime.net or MOCK" + }, + { + "id": "zapier-user-docs", + "package": "zapier-user-docs", + "runtime": "static" + }, + { + "id": "docs-master", + "package": "docs-master", + "runtime": "static" + } + ] +} diff --git a/services/webhook-deliver.json b/services/webhook-deliver.json new file mode 100644 index 0000000..6c058e3 --- /dev/null +++ b/services/webhook-deliver.json @@ -0,0 +1,16 @@ +{ + "id": "webhook-deliver", + "title": "Zapier REST Hook deliverer", + "kind": "nats-worker", + "package": "verae-zapier-middleware", + "role": "webhook-deliver", + "managed": true, + "runtime": "JetStream consumer + HTTPS POST", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13300 }, + "nats": { + "in": ["verae.zapier.webhooks.deliver", "verae.zapier.jobs.events"], + "out": [] + }, + "notes": "HTTPS POST to Zapier hook URL. Not a NATS client in Zapier cloud." +} diff --git a/services/zapier-simulator.json b/services/zapier-simulator.json new file mode 100644 index 0000000..e454710 --- /dev/null +++ b/services/zapier-simulator.json @@ -0,0 +1,14 @@ +{ + "id": "zapier-simulator", + "title": "Zapier interface simulator + trace console", + "kind": "http", + "package": "verae-zapier-simulator", + "role": "zapier-simulator", + "managed": true, + "runtime": "HTTP 127.0.0.1:3847", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13847 }, + "env": { "SIM_PORT": "3847" }, + "nats": { "in": [], "out": [] }, + "notes": "Optional. min 0 in central fleet.json. Does not talk to live NATS." +} diff --git a/services/zappier-edge.json b/services/zappier-edge.json new file mode 100644 index 0000000..a43ca91 --- /dev/null +++ b/services/zappier-edge.json @@ -0,0 +1,14 @@ +{ + "id": "zappier-edge", + "title": "Zappier commercial HTTPS edge", + "kind": "http", + "package": "zappier", + "role": "zappier-edge", + "managed": true, + "runtime": "HTTPS :3000 (portal, admin, meter)", + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13000 }, + "env": { "PORT": "3000" }, + "nats": { "in": [], "out": [] }, + "notes": "Zapier x-api-key lands here. HTTPS to middleware only." +} diff --git a/src/agent.js b/src/agent.js new file mode 100644 index 0000000..05a3d9d --- /dev/null +++ b/src/agent.js @@ -0,0 +1,86 @@ +#!/usr/bin/env node +/** + * Remote host agent: spawn/kill fleet workers so the control plane can spread replicas. + * + * FLEET_AGENT_PORT=3851 node src/agent.js + * + * Bind loopback or a private interface. Do not publish NATS. + */ +import http from 'node:http'; +import { spawn } from 'node:child_process'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; + +const WORKER = path.join(path.dirname(fileURLToPath(import.meta.url)), 'worker.js'); +const PORT = Number(process.env.FLEET_AGENT_PORT || 3851); +const BIND = process.env.FLEET_AGENT_BIND || '127.0.0.1'; +const children = new Map(); + +function json(res, code, obj) { + res.writeHead(code, { 'content-type': 'application/json' }); + res.end(JSON.stringify(obj)); +} + +async function readBody(req) { + const chunks = []; + for await (const c of req) chunks.push(c); + const raw = Buffer.concat(chunks).toString('utf8'); + return raw ? JSON.parse(raw) : {}; +} + +export function startAgent({ port = PORT, bind = BIND } = {}) { + const server = http.createServer(async (req, res) => { + const url = new URL(req.url, `http://${bind}:${port}`); + try { + if (req.method === 'GET' && url.pathname === '/health') { + return json(res, 200, { + ok: true, + role: 'fleet-agent', + workers: [...children.keys()], + }); + } + if (req.method === 'POST' && url.pathname === '/spawn') { + const body = await readBody(req); + const instance = body.instance; + if (!instance) return json(res, 400, { error: 'instance required' }); + if (children.has(instance)) { + return json(res, 200, { pid: children.get(instance).pid, instance, existing: true }); + } + const child = spawn(process.execPath, [WORKER], { + env: { ...process.env, ...(body.env || {}) }, + stdio: ['ignore', 'pipe', 'pipe'], + }); + children.set(instance, child); + child.on('exit', () => children.delete(instance)); + return json(res, 200, { pid: child.pid, instance }); + } + if (req.method === 'POST' && url.pathname === '/kill') { + const body = await readBody(req); + const child = children.get(body.instance); + if (child) { + child.kill('SIGTERM'); + children.delete(body.instance); + } + return json(res, 200, { killed: Boolean(child), instance: body.instance }); + } + json(res, 404, { error: 'not found' }); + } catch (err) { + json(res, 400, { error: err.message }); + } + }); + server.listen(port, bind, () => { + process.stdout.write(`fleet-agent http://${bind}:${port}/\n`); + }); + return { + server, + async close() { + for (const c of children.values()) c.kill('SIGTERM'); + children.clear(); + await new Promise((r) => server.close(r)); + }, + }; +} + +if (process.argv[1] && path.basename(process.argv[1]) === 'agent.js') { + startAgent(); +} diff --git a/src/classify.js b/src/classify.js new file mode 100644 index 0000000..3313b7a --- /dev/null +++ b/src/classify.js @@ -0,0 +1,29 @@ +/** + * Row health: operational | degraded | down + * @module classify + */ + +/** + * @param {{ + * min?: number, + * available?: number, + * running?: number, + * paused?: number, + * managed?: boolean, + * instances?: Array<{ pid?: number|null, paused?: boolean, healthy?: boolean }> + * }} sv + * @returns {'operational'|'degraded'|'down'} + */ +export function classifyService(sv) { + const min = sv.min ?? 0; + const available = sv.available ?? 0; + const running = sv.running ?? 0; + const paused = sv.paused ?? 0; + const inst = sv.instances || []; + const unhealthy = inst.filter((i) => i.pid && i.healthy === false && !i.paused).length; + + if (min === 0 && running === 0) return 'operational'; + if (available === 0 && min > 0) return 'down'; + if (available >= min && paused === 0 && unhealthy === 0) return 'operational'; + return 'degraded'; +} diff --git a/src/cli.js b/src/cli.js new file mode 100644 index 0000000..fdc2951 --- /dev/null +++ b/src/cli.js @@ -0,0 +1,145 @@ +#!/usr/bin/env node +/** + * verae-fleet — catalog, monitor, restart, pause. + * + * Local (no daemon): list + * Daemon: serve | monitor + * Against daemon: status | start | stop | pause | resume | restart + */ +import http from 'node:http'; +import { loadFleet, listServices } from './load.js'; +import { Supervisor } from './supervisor.js'; +import { Monitor } from './monitor.js'; +import { startControlServer } from './server.js'; + +const loaded = loadFleet(); +const [cmd = 'help', target] = process.argv.slice(2); +const BASE = process.env.FLEET_URL || `http://127.0.0.1:${loaded.control.port}`; + +function print(obj) { + process.stdout.write(`${typeof obj === 'string' ? obj : JSON.stringify(obj, null, 2)}\n`); +} + +function help() { + print(`verae-fleet + + list services, config files, replica min/max + serve start floors + monitor + UI ${BASE}/ + monitor same without printing extra + status GET daemon status + start turn on, spawn up to min + stop turn off (service disable) or stop one replica + pause pause (does not count toward tree-node floor) + resume + restart kill + respawn + reconcile force floor check + +Central spec: fleet.json +Machines: machines.json (add hosts to spread replicas) +Per service: services/.json +Remote SSH: machines.json user + host + identityFile (path only) + ssh-check [id] test SSH login (default ns1) +`); +} + +function api(method, pathname) { + return new Promise((resolve, reject) => { + const u = new URL(pathname, BASE); + const req = http.request( + { hostname: u.hostname, port: u.port, path: u.pathname, method, timeout: 8000 }, + (res) => { + const chunks = []; + res.on('data', (c) => chunks.push(c)); + res.on('end', () => { + const raw = Buffer.concat(chunks).toString('utf8'); + try { + resolve({ status: res.statusCode, body: JSON.parse(raw) }); + } catch { + resolve({ status: res.statusCode, body: raw }); + } + }); + }, + ); + req.on('error', reject); + req.end(); + }); +} + +async function main() { + if (cmd === 'help' || cmd === '-h' || cmd === '--help') return help(); + if (cmd === 'list') { + for (const r of listServices(loaded)) { + print( + `${r.id.padEnd(22)} min=${String(r.min).padStart(2)} max=${String(r.max).padStart(2)} floor=${r.keepFloor ? 'yes' : 'no '} managed=${r.managed ? 'yes' : 'no '} ${r.configPath}`, + ); + } + print(`\ncentral replica spec: ${loaded.fleetPath}`); + print(`tree-node floor: min=${loaded.services['tree-node'].min} keepFloor=${loaded.services['tree-node'].keepFloor}`); + print('\nmachines:'); + for (const m of loaded.machines) { + const ssh = m.kind === 'ssh' ? `${m.user}@${m.host} key=${m.identityFile}` : m.host; + print(` ${m.id.padEnd(12)} ${m.kind.padEnd(6)} ${m.enabled ? 'on ' : 'off'} cap=${m.capacity} ${ssh}`); + } + return; + } + if (cmd === 'ssh-check') { + const { sshCheck } = await import('./ssh.js'); + const id = target || 'ns1'; + const m = loaded.machines.find((x) => x.id === id); + if (!m) throw new Error(`unknown machine ${id}`); + print(await sshCheck(m)); + return; + } + if (cmd === 'serve' || cmd === 'monitor') { + const sup = new Supervisor({ loaded }); + for (const id of Object.keys(loaded.services)) { + const s = loaded.services[id]; + if (s.managed && s.enabled && s.min > 0) await sup.startService(id); + } + const mon = new Monitor(sup); + await mon.tick(); + mon.start(); + startControlServer(sup, mon); + print(`fleet ${cmd} pid=${process.pid}\nUI ${BASE}/\nSIGINT stops all managed replicas`); + process.on('SIGINT', async () => { + mon.stop(); + await sup.stopAll(); + process.exit(0); + }); + process.on('SIGTERM', async () => { + mon.stop(); + await sup.stopAll(); + process.exit(0); + }); + return; + } + + const instance = target && /-\d+$/.test(target); + const map = { + status: ['GET', '/api/status'], + reconcile: ['POST', '/api/reconcile'], + }; + if (map[cmd] && !target) { + const [m, p] = map[cmd]; + const r = await api(m, p); + print(r.body); + if (r.status >= 400) process.exitCode = 1; + return; + } + if (!target) { + help(); + process.exitCode = 1; + return; + } + const pathName = instance + ? `/api/instances/${target}/${cmd}` + : `/api/services/${target}/${cmd}`; + const r = await api('POST', pathName); + print(r.body); + if (r.status >= 400) process.exitCode = 1; +} + +main().catch((err) => { + print({ error: err.message, hint: 'Is `verae-fleet serve` running?' }); + process.exitCode = 1; +}); diff --git a/src/health.js b/src/health.js new file mode 100644 index 0000000..50ec05f --- /dev/null +++ b/src/health.js @@ -0,0 +1,118 @@ +/** + * Probe a replica: HTTP /health or TCP connect. + * @module health + */ + +import http from 'node:http'; +import net from 'node:net'; +import fs from 'node:fs'; +import path from 'node:path'; + +/** + * @param {number|{ host?: string, port: number, path?: string, timeoutMs?: number }} portOrOpts + */ +export function httpProbe(portOrOpts, pathname = '/health', timeoutMs = 800) { + const opts = + typeof portOrOpts === 'object' + ? portOrOpts + : { port: portOrOpts, path: pathname, timeoutMs }; + const host = opts.host || '127.0.0.1'; + const port = opts.port; + const pathName = opts.path || pathname; + const tmo = opts.timeoutMs || timeoutMs; + return new Promise((resolve) => { + const req = http.get( + { host, port, path: pathName, timeout: tmo }, + (res) => { + const chunks = []; + res.on('data', (c) => chunks.push(c)); + res.on('end', () => { + let body = {}; + try { + body = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}'); + } catch { + body = {}; + } + resolve({ + ok: res.statusCode >= 200 && res.statusCode < 300 && body.ok !== false, + statusCode: res.statusCode, + paused: Boolean(body.paused), + body, + }); + }); + }, + ); + req.on('error', (err) => resolve({ ok: false, error: err.message, paused: false })); + req.on('timeout', () => { + req.destroy(); + resolve({ ok: false, error: 'timeout', paused: false }); + }); + }); +} + +export function tcpProbe(host, port, timeoutMs = 400) { + return new Promise((resolve) => { + const sock = net.connect({ host, port }); + const done = (ok, error) => { + try { + sock.destroy(); + } catch { + /* ignore */ + } + resolve({ ok, error, paused: false }); + }; + sock.setTimeout(timeoutMs); + sock.on('connect', () => done(true)); + sock.on('error', (err) => done(false, err.message)); + sock.on('timeout', () => done(false, 'timeout')); + }); +} + +/** + * @param {number|{ host?: string, port: number, path?: string, body?: object }} portOrOpts + * @param {string} [pathname] + */ +export function postControl(portOrOpts, pathname = '/') { + const opts = typeof portOrOpts === 'object' ? portOrOpts : { port: portOrOpts, path: pathname }; + const host = opts.host || '127.0.0.1'; + const port = opts.port; + const pathName = opts.path || pathname; + const body = opts.body ? JSON.stringify(opts.body) : ''; + return new Promise((resolve, reject) => { + const req = http.request( + { + host, + port, + path: pathName, + method: 'POST', + timeout: opts.timeoutMs || 2000, + headers: body ? { 'content-type': 'application/json', 'content-length': Buffer.byteLength(body) } : {}, + }, + (res) => { + const chunks = []; + res.on('data', (c) => chunks.push(c)); + res.on('end', () => { + try { + resolve(JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}')); + } catch { + resolve({}); + } + }); + }, + ); + req.on('error', reject); + if (body) req.write(body); + req.end(); + }); +} + +export function heartbeatFresh(stateDir, maxAgeMs) { + const file = path.join(stateDir, 'heartbeat.json'); + try { + const st = fs.statSync(file); + const body = JSON.parse(fs.readFileSync(file, 'utf8')); + return { ok: Date.now() - st.mtimeMs < maxAgeMs && body.ok !== false, body, ageMs: Date.now() - st.mtimeMs }; + } catch (err) { + return { ok: false, error: err.message }; + } +} diff --git a/src/load.js b/src/load.js new file mode 100644 index 0000000..96d9662 --- /dev/null +++ b/src/load.js @@ -0,0 +1,94 @@ +/** + * Load central fleet.json + per-service JSON configs. + * @module load + */ + +import fs from 'node:fs'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { loadMachines } from './machines.js'; + +const HERE = path.dirname(fileURLToPath(import.meta.url)); +export const FLEET_ROOT = path.resolve(HERE, '..'); + +export function readJson(file) { + return JSON.parse(fs.readFileSync(file, 'utf8')); +} + +/** + * @param {string} [root] + */ +export function loadFleet(root = FLEET_ROOT, opts = {}) { + const fleetPath = path.join(root, 'fleet.json'); + const fleet = readJson(fleetPath); + const dir = path.join(root, 'services'); + const services = {}; + const unmanaged = []; + for (const name of fs.readdirSync(dir).sort()) { + if (!name.endsWith('.json')) continue; + const spec = readJson(path.join(dir, name)); + const configPath = path.join('services', name); + if (spec.id === '_unmanaged') { + for (const e of spec.entries || []) { + unmanaged.push({ ...e, managed: false, configPath }); + } + continue; + } + const central = fleet.services?.[spec.id] || {}; + services[spec.id] = { + ...spec, + configPath, + min: central.min ?? 0, + max: central.max ?? 1, + keepFloor: central.keepFloor !== false, + enabled: central.enabled !== false, + managed: central.managed === false ? false : spec.managed !== false, + }; + } + const overlay = + opts.overlay === false ? null : path.join(root, 'data', 'runtime', 'machines-overlay.json'); + const machines = loadMachines(root, overlay); + return { + root, + fleetPath: path.relative(root, fleetPath) || 'fleet.json', + machinesPath: 'machines.json', + machinesOverlay: overlay, + control: fleet.control, + nats: fleet.nats, + machines, + services, + unmanaged, + }; +} + +export function listServices(loaded) { + const rows = Object.values(loaded.services).map((s) => ({ + id: s.id, + title: s.title, + kind: s.kind, + managed: s.managed, + enabled: s.enabled, + min: s.min, + max: s.max, + keepFloor: s.keepFloor, + configPath: s.configPath, + package: s.package, + runtime: s.runtime, + })); + for (const u of loaded.unmanaged) { + rows.push({ + id: u.id, + title: u.id, + kind: 'unmanaged', + managed: false, + enabled: false, + min: 0, + max: 0, + keepFloor: false, + configPath: u.configPath, + package: u.package, + runtime: u.runtime, + }); + } + return rows; +} diff --git a/src/machines.js b/src/machines.js new file mode 100644 index 0000000..551bd55 --- /dev/null +++ b/src/machines.js @@ -0,0 +1,107 @@ +/** + * Host catalog: where replicas may run. Least-loaded placement. + * @module machines + */ + +import fs from 'node:fs'; +import path from 'node:path'; + +export function defaultMachines() { + return [ + { + id: 'local', + title: 'Control plane (this host)', + kind: 'local', + host: '127.0.0.1', + enabled: true, + capacity: 32, + roles: ['*'], + }, + ]; +} + +export function loadMachines(root, overlayPath) { + const file = path.join(root, 'machines.json'); + let list = defaultMachines(); + if (fs.existsSync(file)) { + const raw = JSON.parse(fs.readFileSync(file, 'utf8')); + if (Array.isArray(raw.machines) && raw.machines.length) list = raw.machines; + } + if (overlayPath && fs.existsSync(overlayPath)) { + const extra = JSON.parse(fs.readFileSync(overlayPath, 'utf8')); + const incoming = Array.isArray(extra) ? extra : extra.machines || []; + for (const m of incoming) upsertMachine(list, m); + } + const secretsPath = path.join(root, 'machines.secrets.json'); + if (fs.existsSync(secretsPath)) { + const secrets = JSON.parse(fs.readFileSync(secretsPath, 'utf8')); + const byId = secrets.machines || secrets; + for (const m of list) { + const extra = byId[m.id]; + if (extra && typeof extra === 'object') Object.assign(m, extra); + } + } + return list.map(normalizeMachine); +} + +export function normalizeMachine(m) { + const kind = m.kind === 'agent' || m.kind === 'ssh' ? m.kind : 'local'; + return { + id: String(m.id || '').trim(), + title: m.title || m.id, + kind, + host: m.host || '127.0.0.1', + user: m.user || (kind === 'ssh' ? 'marchon' : ''), + sshPort: Number(m.sshPort || 22), + identityFile: m.identityFile || (kind === 'ssh' ? '~/.ssh/id_ed25519' : ''), + remoteDir: m.remoteDir || '~/verae-fleet-runtime', + agentPort: Number(m.agentPort || 3851), + enabled: m.enabled !== false, + capacity: Number(m.capacity || 8), + roles: Array.isArray(m.roles) && m.roles.length ? m.roles : ['*'], + notes: m.notes || '', + }; +} + +export function upsertMachine(list, spec) { + const m = normalizeMachine(spec); + if (!m.id) throw new Error('machine id required'); + const i = list.findIndex((x) => x.id === m.id); + if (i >= 0) list[i] = { ...list[i], ...m }; + else list.push(m); + return m; +} + +export function canHost(machine, serviceId, role) { + if (!machine?.enabled) return false; + const roles = machine.roles || ['*']; + if (roles.includes('*')) return true; + return roles.includes(serviceId) || (role && roles.includes(role)); +} + +/** + * Pick the enabled machine with the most free capacity (fewest running / capacity). + * @param {object[]} machines + * @param {Array<{ machine?: string }>} instances + * @param {string} serviceId + * @param {string} [role] + */ +export function pickMachine(machines, instances, serviceId, role, exclude = []) { + const skip = new Set(exclude); + const eligible = machines.filter((m) => canHost(m, serviceId, role) && !skip.has(m.id)); + if (!eligible.length) return null; + const scored = eligible.map((m) => { + const running = instances.filter((i) => i.machine === m.id && i.pid).length; + const free = m.capacity - running; + const load = m.capacity <= 0 ? 1 : running / m.capacity; + return { m, running, free, load }; + }); + scored.sort((a, b) => a.load - b.load || a.running - b.running || a.m.id.localeCompare(b.m.id)); + const best = scored.find((s) => s.free > 0); + return best ? best.m : null; +} + +export function saveOverlay(overlayPath, machines) { + fs.mkdirSync(path.dirname(overlayPath), { recursive: true }); + fs.writeFileSync(overlayPath, JSON.stringify({ machines }, null, 2)); +} diff --git a/src/monitor.js b/src/monitor.js new file mode 100644 index 0000000..6cdb2e1 --- /dev/null +++ b/src/monitor.js @@ -0,0 +1,38 @@ +/** + * Periodic probe + reconcile so replica floors (especially tree-node) stay met. + * @module monitor + */ + +export class Monitor { + /** + * @param {import('./supervisor.js').Supervisor} supervisor + * @param {{ intervalMs?: number }} [opts] + */ + constructor(supervisor, opts = {}) { + this.supervisor = supervisor; + this.intervalMs = opts.intervalMs ?? supervisor.loaded.control?.probeIntervalMs ?? 750; + this.timer = null; + this.last = null; + } + + async tick() { + const actions = await this.supervisor.reconcile(); + this.last = { t: new Date().toISOString(), actions, status: this.supervisor.status() }; + return this.last; + } + + start() { + if (this.timer) return; + this.timer = setInterval(() => { + this.tick().catch((err) => { + this.supervisor.log('monitor-error', { error: err.message }); + }); + }, this.intervalMs); + this.timer.unref?.(); + } + + stop() { + if (this.timer) clearInterval(this.timer); + this.timer = null; + } +} diff --git a/src/rtt.js b/src/rtt.js new file mode 100644 index 0000000..09ec7bb --- /dev/null +++ b/src/rtt.js @@ -0,0 +1,57 @@ +/** + * Ring buffer of message-processing RTTs (ms) for planning percentiles. + * @module rtt + */ + +export class RttWindow { + /** + * @param {number} [cap=400] + */ + constructor(cap = 400) { + this.cap = cap; + /** @type {number[]} */ + this.samples = []; + } + + add(ms) { + const n = Number(ms); + if (!Number.isFinite(n) || n < 0) return; + this.samples.push(n); + if (this.samples.length > this.cap) this.samples.splice(0, this.samples.length - this.cap); + } + + stats() { + return summarizeRtt(this.samples); + } +} + +/** + * @param {number[]} samples + */ +export function summarizeRtt(samples) { + const xs = samples.filter((n) => Number.isFinite(n) && n >= 0).slice().sort((a, b) => a - b); + if (!xs.length) { + return { count: 0, minMs: null, avgMs: null, p50Ms: null, p90Ms: null }; + } + const sum = xs.reduce((a, b) => a + b, 0); + return { + count: xs.length, + minMs: round1(xs[0]), + avgMs: round1(sum / xs.length), + p50Ms: round1(percentile(xs, 0.5)), + p90Ms: round1(percentile(xs, 0.9)), + }; +} + +function percentile(sorted, p) { + if (sorted.length === 1) return sorted[0]; + const idx = (sorted.length - 1) * p; + const lo = Math.floor(idx); + const hi = Math.ceil(idx); + if (lo === hi) return sorted[lo]; + return sorted[lo] + (sorted[hi] - sorted[lo]) * (idx - lo); +} + +function round1(n) { + return Math.round(n * 10) / 10; +} diff --git a/src/server.js b/src/server.js new file mode 100644 index 0000000..311be8b --- /dev/null +++ b/src/server.js @@ -0,0 +1,145 @@ +/** + * Operator HTTP: list, status, pause/resume/stop/start/restart. + * Binds 127.0.0.1 only. + * @module server + */ + +import http from 'node:http'; +import fs from 'node:fs'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { listServices } from './load.js'; +import { ACTIONS, Simulator } from '../../verae-zapier-simulator/src/pipeline.js'; + +const PUBLIC = path.join(path.dirname(fileURLToPath(import.meta.url)), '..', 'public'); + +function json(res, code, obj) { + res.writeHead(code, { 'content-type': 'application/json', 'cache-control': 'no-store' }); + res.end(JSON.stringify(obj, null, 2)); +} + +/** + * @param {import('./supervisor.js').Supervisor} sup + * @param {import('./monitor.js').Monitor} [mon] + */ +export function startControlServer(sup, mon) { + const bind = sup.loaded.control?.bind || '127.0.0.1'; + const port = Number(sup.loaded.control?.port || 3850); + const sim = new Simulator(); + + const server = http.createServer(async (req, res) => { + const url = new URL(req.url, `http://${bind}:${port}`); + try { + if (req.method === 'GET' && url.pathname === '/health') { + return json(res, 200, { ok: true, role: 'operator-console' }); + } + if (req.method === 'GET' && url.pathname === '/api/sim/actions') { + return json(res, 200, { actions: ACTIONS, state: sim.snapshot() }); + } + if (req.method === 'POST' && url.pathname === '/api/sim/reset') { + sim.reset(); + return json(res, 200, { ok: true, state: sim.snapshot() }); + } + if (req.method === 'POST' && url.pathname === '/api/sim/run') { + const chunks = []; + for await (const c of req) chunks.push(c); + const body = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}'); + const result = await sim.run(body); + return json(res, 200, result); + } + if (req.method === 'GET' && url.pathname === '/api/services') { + return json(res, 200, { services: listServices(sup.loaded), fleetPath: sup.loaded.fleetPath }); + } + if (req.method === 'GET' && url.pathname === '/api/status') { + return json(res, 200, { ...sup.status(), monitor: mon?.last || null }); + } + if (req.method === 'GET' && url.pathname === '/api/machines') { + return json(res, 200, { machines: sup.status().machines }); + } + if (req.method === 'POST' && url.pathname === '/api/machines') { + const chunks = []; + for await (const c of req) chunks.push(c); + const spec = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}'); + const m = sup.addMachine(spec); + return json(res, 200, { machine: m, machines: sup.status().machines }); + } + const mach = url.pathname.match(/^\/api\/machines\/([^/]+)\/(enable|disable|check)$/); + if (mach && req.method === 'POST') { + const [, id, op] = mach; + if (op === 'check') { + const check = await sup.checkMachine(id); + return json(res, check.ok ? 200 : 400, { check, machines: sup.status().machines }); + } + const m = sup.setMachineEnabled(id, op === 'enable'); + return json(res, 200, { machine: m, machines: sup.status().machines }); + } + if (req.method === 'POST' && url.pathname === '/api/reconcile') { + const actions = await sup.reconcile(); + return json(res, 200, { actions, status: sup.status() }); + } + const svc = url.pathname.match(/^\/api\/services\/([^/]+)\/(start|stop|pause|resume)$/); + if (svc && req.method === 'POST') { + const [, id, op] = svc; + const fn = { + start: () => sup.startService(id), + stop: () => sup.stopService(id), + pause: () => sup.pauseService(id), + resume: () => sup.resumeService(id), + }[op]; + const result = await fn(); + return json(res, 200, { op, id, result, status: sup.status().services[id] }); + } + const inst = url.pathname.match(/^\/api\/instances\/([^/]+)\/(pause|resume|restart|stop|unhealthy)$/); + if (inst && req.method === 'POST') { + const [, id, op] = inst; + const fn = { + pause: () => sup.pauseInstance(id), + resume: () => sup.resumeInstance(id), + restart: () => sup.restartInstance(id), + stop: () => sup.stopInstance(id, { replace: true }), + unhealthy: () => sup.markUnhealthy(id), + }[op]; + const result = await fn(); + return json(res, 200, { op, id, result: result?.id || result, status: sup.status() }); + } + if (req.method === 'GET' && (url.pathname === '/' || url.pathname === '/index.html')) { + const dest = path.join(PUBLIC, 'index.html'); + res.writeHead(200, { 'content-type': 'text/html; charset=utf-8' }); + fs.createReadStream(dest).pipe(res); + return; + } + if (req.method === 'GET' && url.pathname.startsWith('/docs/')) { + const dest = path.normalize(path.join(PUBLIC, '..', url.pathname)); + const root = path.normalize(path.join(PUBLIC, '..', 'docs')); + if ( + !(dest === root || dest.startsWith(root + path.sep)) || + !fs.existsSync(dest) || + !fs.statSync(dest).isFile() + ) { + return json(res, 404, { error: 'not found' }); + } + const ext = path.extname(dest).toLowerCase(); + const types = { + '.png': 'image/png', + '.jpg': 'image/jpeg', + '.jpeg': 'image/jpeg', + '.svg': 'image/svg+xml', + '.pdf': 'application/pdf', + '.md': 'text/markdown; charset=utf-8', + '.html': 'text/html; charset=utf-8', + }; + res.writeHead(200, { 'content-type': types[ext] || 'application/octet-stream' }); + fs.createReadStream(dest).pipe(res); + return; + } + json(res, 404, { error: 'not found' }); + } catch (err) { + json(res, 400, { error: err.message }); + } + }); + + server.listen(port, bind, () => { + process.stdout.write(`fleet control http://${bind}:${port}/\n`); + }); + return server; +} diff --git a/src/ssh.js b/src/ssh.js new file mode 100644 index 0000000..e8a86c7 --- /dev/null +++ b/src/ssh.js @@ -0,0 +1,170 @@ +/** + * SSH control of remote fleet workers. Uses username + host + identity file path. + * Never logs or stores private-key material. + * @module ssh + */ + +import { spawn } from 'node:child_process'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; + +const HERE = path.dirname(fileURLToPath(import.meta.url)); + +export function expandHome(p) { + if (!p) return p; + if (p === '~') return os.homedir(); + if (p.startsWith('~/')) return path.join(os.homedir(), p.slice(2)); + return p; +} + +export function shellQuote(s) { + return `'${String(s).replace(/'/g, `'\"'\"'`)}'`; +} + +export function sshTarget(machine) { + return `${machine.user || 'marchon'}@${machine.host}`; +} + +export function sshBaseArgs(machine) { + const ident = expandHome(machine.identityFile || ''); + const args = [ + '-n', + '-o', 'BatchMode=yes', + '-o', 'StrictHostKeyChecking=accept-new', + '-o', 'IdentitiesOnly=yes', + '-o', 'ConnectTimeout=8', + '-p', String(machine.sshPort || 22), + ]; + if (ident) { + if (!fs.existsSync(ident)) throw new Error(`identityFile not found: ${ident}`); + args.push('-i', ident); + } + return args; +} + +export function scpBaseArgs(machine) { + const ident = expandHome(machine.identityFile || ''); + const args = [ + '-o', 'BatchMode=yes', + '-o', 'StrictHostKeyChecking=accept-new', + '-o', 'IdentitiesOnly=yes', + '-o', 'ConnectTimeout=8', + '-P', String(machine.sshPort || 22), + ]; + if (ident) args.push('-i', ident); + return args; +} + +export function sshExec(machine, remoteCommand, { timeoutMs = 20000 } = {}) { + return new Promise((resolve, reject) => { + const args = [...sshBaseArgs(machine), sshTarget(machine), remoteCommand]; + const child = spawn('ssh', args, { stdio: ['ignore', 'pipe', 'pipe'] }); + const out = []; + const err = []; + child.stdout.on('data', (c) => out.push(c)); + child.stderr.on('data', (c) => err.push(c)); + const t = setTimeout(() => { + child.kill('SIGKILL'); + reject(new Error(`ssh timeout: ${machine.id || machine.host}`)); + }, timeoutMs); + child.on('error', (e) => { + clearTimeout(t); + reject(e); + }); + child.on('close', (code) => { + clearTimeout(t); + const stdout = Buffer.concat(out).toString('utf8').trim(); + const stderr = Buffer.concat(err).toString('utf8').trim(); + if (code !== 0) { + reject(new Error(stderr || stdout || `ssh exit ${code}`)); + return; + } + resolve({ stdout, stderr, code }); + }); + }); +} + +export function scpTo(machine, localFiles, remoteDir) { + return new Promise((resolve, reject) => { + const dest = `${sshTarget(machine)}:${remoteDir}/`; + const args = [...scpBaseArgs(machine), ...localFiles, dest]; + const child = spawn('scp', args, { stdio: ['ignore', 'pipe', 'pipe'] }); + const err = []; + child.stderr.on('data', (c) => err.push(c)); + const t = setTimeout(() => { + child.kill('SIGKILL'); + reject(new Error('scp timeout')); + }, 30000); + child.on('close', (code) => { + clearTimeout(t); + if (code !== 0) reject(new Error(Buffer.concat(err).toString('utf8') || `scp exit ${code}`)); + else resolve(); + }); + child.on('error', (e) => { + clearTimeout(t); + reject(e); + }); + }); +} + +const synced = new Set(); + +export async function ensureRemoteRuntime(machine) { + const remoteDir = machine.remoteDir || '~/verae-fleet-runtime'; + const key = `${machine.id}:${remoteDir}`; + if (synced.has(key)) return remoteDir; + await sshExec(machine, `mkdir -p ${remoteDir}/src ${remoteDir}/data`); + await scpTo(machine, [path.join(HERE, 'worker.js'), path.join(HERE, 'rtt.js')], `${remoteDir}/src`); + synced.add(key); + return remoteDir; +} + +export async function sshSpawnWorker(machine, { instance, env }) { + const remoteDir = await ensureRemoteRuntime(machine); + const fleetEnv = Object.entries(env) + .filter(([k]) => k.startsWith('FLEET_')) + .map(([k, v]) => `${k}=${shellQuote(v)}`) + .join(' '); + const inner = `node src/worker.js data/${instance}/worker.log 2>&1 & echo $!`; + const cmd = `mkdir -p ${remoteDir}/data/${instance} && cd ${remoteDir} && ${fleetEnv} bash -c ${shellQuote(inner)}`; + const { stdout } = await sshExec(machine, cmd); + const pid = Number(String(stdout).split('\n').pop()); + if (!Number.isFinite(pid)) throw new Error(`remote spawn produced no pid: ${stdout}`); + return { pid, remoteDir }; +} + +export async function sshKillWorker(machine, pid, instance) { + const cmd = pid + ? `kill -TERM ${Number(pid)} 2>/dev/null || true` + : `pkill -f ${shellQuote(`FLEET_INSTANCE=${instance}`)} 2>/dev/null || true`; + await sshExec(machine, cmd); +} + +export async function sshHttp(machine, method, urlPath, { port, body } = {}) { + const url = `http://127.0.0.1:${port}${urlPath}`; + let cmd = `curl -sS -m 3 -X ${method}`; + if (body) { + cmd += ` -H 'content-type: application/json' -d ${shellQuote(JSON.stringify(body))}`; + } + cmd += ` ${shellQuote(url)}`; + const { stdout } = await sshExec(machine, cmd, { timeoutMs: 8000 }); + try { + return JSON.parse(stdout || '{}'); + } catch { + return { raw: stdout }; + } +} + +export async function sshCheck(machine) { + const { stdout } = await sshExec(machine, 'echo OK; whoami; hostname; command -v node; node -v'); + const lines = stdout.split('\n'); + return { + ok: lines[0] === 'OK', + user: lines[1], + hostname: lines[2], + node: lines[3] ? `${lines[3]} ${lines[4] || ''}`.trim() : null, + identityFile: expandHome(machine.identityFile || ''), + }; +} diff --git a/src/supervisor.js b/src/supervisor.js new file mode 100644 index 0000000..5bb543e --- /dev/null +++ b/src/supervisor.js @@ -0,0 +1,510 @@ +/** + * Spawn, pause, resume, stop, restart replicas. Keep replica floor (tree-node min). + * @module supervisor + */ + +import { spawn } from 'node:child_process'; +import fs from 'node:fs'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { loadFleet } from './load.js'; +import { httpProbe, postControl } from './health.js'; +import { classifyService } from './classify.js'; +import { pickMachine, upsertMachine, saveOverlay } from './machines.js'; +import { summarizeRtt } from './rtt.js'; +import { sshSpawnWorker, sshKillWorker, sshHttp, sshCheck } from './ssh.js'; + +const WORKER = path.join(path.dirname(fileURLToPath(import.meta.url)), 'worker.js'); + +export class Supervisor { + /** + * @param {object} [opts] + * @param {ReturnType} [opts.loaded] + * @param {string} [opts.stateRoot] + */ + constructor(opts = {}) { + this.loaded = opts.loaded || loadFleet(); + this.stateRoot = opts.stateRoot || path.join(this.loaded.root, 'data', 'runtime'); + fs.mkdirSync(this.stateRoot, { recursive: true }); + /** @type {Map} */ + this.instances = new Map(); + this.events = []; + this._seq = 0; + if (!Array.isArray(this.loaded.machines) || !this.loaded.machines.length) { + this.loaded.machines = [ + { id: 'local', title: 'local', kind: 'local', host: '127.0.0.1', enabled: true, capacity: 32, roles: ['*'] }, + ]; + } + } + + machines() { + return this.loaded.machines; + } + + addMachine(spec) { + const m = upsertMachine(this.loaded.machines, spec); + if (this.loaded.machinesOverlay) saveOverlay(this.loaded.machinesOverlay, this.loaded.machines); + this.log('machine-add', { id: m.id, host: m.host, kind: m.kind }); + return m; + } + + setMachineEnabled(id, enabled) { + const m = this.loaded.machines.find((x) => x.id === id); + if (!m) throw new Error(`unknown machine ${id}`); + m.enabled = Boolean(enabled); + if (this.loaded.machinesOverlay) saveOverlay(this.loaded.machinesOverlay, this.loaded.machines); + return m; + } + + async checkMachine(id) { + const m = this.machines().find((x) => x.id === id); + if (!m) throw new Error(`unknown machine ${id}`); + if (m.kind === 'ssh') return sshCheck(m); + if (m.kind === 'agent') { + const r = await httpProbe({ host: m.host, port: m.agentPort, path: '/health', timeoutMs: 2000 }); + return { ok: r.ok, kind: 'agent', body: r.body }; + } + return { ok: true, kind: 'local', host: m.host }; + } + + machineOf(rec) { + return this.machines().find((x) => x.id === rec.machine); + } + + async controlInstance(rec, pathname, body) { + if (rec.kind === 'ssh') { + const m = this.machineOf(rec); + if (!m) throw new Error(`machine ${rec.machine} missing`); + return sshHttp(m, 'POST', pathname, { port: rec.healthPort, body }); + } + return postControl({ + host: rec.host || '127.0.0.1', + port: rec.healthPort, + path: pathname, + body, + }); + } + + log(type, detail) { + const ev = { t: new Date().toISOString(), type, ...detail }; + this.events.push(ev); + if (this.events.length > 500) this.events.splice(0, this.events.length - 400); + return ev; + } + + spec(serviceId) { + const s = this.loaded.services[serviceId]; + if (!s) throw new Error(`unknown service ${serviceId}`); + return s; + } + + nextIndex(serviceId) { + const used = new Set( + [...this.instances.values()].filter((i) => i.service === serviceId).map((i) => i.index), + ); + const max = this.spec(serviceId).max ?? 8; + for (let i = 0; i < max; i += 1) if (!used.has(i)) return i; + return null; + } + + /** + * @param {string} serviceId + * @param {{ index?: number }} [opts] + */ + async startOne(serviceId, opts = {}) { + const spec = this.spec(serviceId); + if (!spec.managed) { + this.log('skip', { service: serviceId, reason: 'unmanaged' }); + return null; + } + if (!spec.enabled) { + this.log('skip', { service: serviceId, reason: 'disabled' }); + return null; + } + const index = opts.index ?? this.nextIndex(serviceId); + if (index == null) { + this.log('skip', { service: serviceId, reason: 'at-max' }); + return null; + } + const instance = `${serviceId}-${index}`; + if (this.instances.has(instance)) return this.instances.get(instance); + const machine = + opts.machine || + pickMachine( + this.machines(), + [...this.instances.values()], + serviceId, + spec.role || serviceId, + opts.excludeMachines || [], + ); + if (!machine) { + this.log('skip', { service: serviceId, reason: 'no-machine-capacity' }); + return null; + } + const healthPort = (spec.ports?.healthBase || 19000) + index; + const stateDir = path.join(this.stateRoot, instance); + fs.mkdirSync(stateDir, { recursive: true }); + const env = { + ...process.env, + FLEET_ROLE: spec.role || serviceId, + FLEET_SERVICE: serviceId, + FLEET_INSTANCE: instance, + FLEET_MACHINE: machine.id, + FLEET_HEALTH_PORT: String(healthPort), + FLEET_HEALTH_BIND: machine.kind === 'local' ? '127.0.0.1' : '127.0.0.1', + FLEET_STATE_DIR: machine.kind === 'ssh' ? `${machine.remoteDir || '~/verae-fleet-runtime'}/data/${instance}` : stateDir, + FLEET_MESSAGE_HZ: process.env.FLEET_MESSAGE_HZ || '8', + }; + let child = null; + let pid = null; + if (machine.kind === 'ssh') { + try { + const spawned = await sshSpawnWorker(machine, { instance, env }); + pid = spawned.pid; + } catch (err) { + this.log('skip', { service: serviceId, reason: 'ssh-spawn-failed', machine: machine.id, error: err.message }); + const exclude = [...(opts.excludeMachines || []), machine.id]; + if (exclude.length < this.machines().length) { + return this.startOne(serviceId, { ...opts, excludeMachines: exclude }); + } + return null; + } + } else if (machine.kind === 'agent') { + try { + const spawned = await postControl({ + host: machine.host, + port: machine.agentPort, + path: '/spawn', + body: { env, instance }, + timeoutMs: 4000, + }); + pid = spawned.pid || null; + } catch (err) { + this.log('skip', { service: serviceId, reason: 'agent-spawn-failed', machine: machine.id, error: err.message }); + return null; + } + } else { + child = spawn(process.execPath, [WORKER], { env, stdio: ['ignore', 'pipe', 'pipe'] }); + pid = child.pid; + } + const rec = { + id: instance, + service: serviceId, + index, + machine: machine.id, + host: machine.host, + pid, + healthPort, + stateDir, + child, + kind: machine.kind, + _machine: machine, + paused: false, + startedAt: Date.now(), + restarts: 0, + lastError: null, + rttSamples: [], + }; + if (child) { + child.stderr?.on('data', () => {}); + child.on('exit', (code, signal) => { + rec.exitCode = code; + rec.signal = signal; + rec.pid = null; + this.log('exit', { instance, service: serviceId, code, signal }); + }); + } + this.instances.set(instance, rec); + this.log('start', { instance, service: serviceId, machine: machine.id, pid: rec.pid, healthPort }); + const h = await waitForHealth(rec, 8000); + rec.healthy = h.ok; + rec.lastProbe = h; + return rec; + } + + async stopInstance(instanceId, { replace = true } = {}) { + const rec = this.instances.get(instanceId); + if (!rec) return null; + rec.stopping = true; + if (rec.kind === 'ssh') { + const m = this.machines().find((x) => x.id === rec.machine); + if (m) { + try { + await sshKillWorker(m, rec.pid, instanceId); + } catch { + /* ignore */ + } + } + } else if (rec.kind === 'agent') { + const m = this.machines().find((x) => x.id === rec.machine); + if (m) { + try { + await postControl({ host: m.host, port: m.agentPort, path: '/kill', body: { instance: instanceId } }); + } catch { + /* ignore */ + } + } + } else if (rec.child && rec.pid) { + rec.child.kill('SIGTERM'); + await waitExit(rec.child, 2000); + } + this.instances.delete(instanceId); + this.log('stop', { instance: instanceId, service: rec.service }); + if (replace) await this.reconcile(rec.service); + return rec; + } + + async pauseInstance(instanceId, { replace = true } = {}) { + const rec = this.instances.get(instanceId); + if (!rec) throw new Error(`no instance ${instanceId}`); + await this.controlInstance(rec, '/pause'); + rec.paused = true; + this.log('pause', { instance: instanceId, service: rec.service }); + if (replace) await this.reconcile(rec.service); + return rec; + } + + async resumeInstance(instanceId) { + const rec = this.instances.get(instanceId); + if (!rec) throw new Error(`no instance ${instanceId}`); + await this.controlInstance(rec, '/resume'); + rec.paused = false; + this.log('resume', { instance: instanceId, service: rec.service }); + return rec; + } + + async restartInstance(instanceId) { + const rec = this.instances.get(instanceId); + if (!rec) throw new Error(`no instance ${instanceId}`); + const { service, index, machine } = rec; + await this.stopInstance(instanceId, { replace: false }); + const next = await this.startOne(service, { index, machine: this.machines().find((m) => m.id === machine) }); + if (next) next.restarts = (rec.restarts || 0) + 1; + this.log('restart', { instance: instanceId, service }); + return next; + } + + async markUnhealthy(instanceId) { + const rec = this.instances.get(instanceId); + if (!rec) throw new Error(`no instance ${instanceId}`); + await this.controlInstance(rec, '/unhealthy'); + return rec; + } + + async probe(rec) { + const spec = this.spec(rec.service); + const pathName = spec.health?.path || '/health'; + let r; + if (rec.kind === 'ssh') { + const m = this.machineOf(rec); + try { + const body = await sshHttp(m, 'GET', pathName, { port: rec.healthPort }); + r = { ok: body.ok !== false, paused: Boolean(body.paused), body, statusCode: body.ok === false ? 503 : 200 }; + } catch (err) { + r = { ok: false, paused: false, error: err.message }; + } + } else { + r = await httpProbe({ + host: rec.host || '127.0.0.1', + port: rec.healthPort, + path: pathName, + timeoutMs: spec.health?.timeoutMs || 800, + }); + } + rec.paused = Boolean(r.paused); + rec.lastProbe = r; + rec.healthy = r.ok; + if (r.ok && !rec.paused) { + try { + const msg = await this.controlInstance(rec, '/message'); + if (Number.isFinite(msg.rttMs)) { + rec.rttSamples = rec.rttSamples || []; + rec.rttSamples.push(msg.rttMs); + if (rec.rttSamples.length > 200) rec.rttSamples.splice(0, rec.rttSamples.length - 200); + } + rec.rtt = msg.rtt || r.body?.rtt || summarizeRtt(rec.rttSamples); + } catch { + rec.rtt = r.body?.rtt || summarizeRtt(rec.rttSamples || []); + } + } + return r; + } + + available(serviceId) { + return [...this.instances.values()].filter( + (i) => i.service === serviceId && i.pid && !i.paused && i.healthy !== false, + ); + } + + /** + * Ensure running healthy unpaused copies >= min when keepFloor. + * @param {string} [only] + */ + async reconcile(only) { + const actions = []; + const ids = only ? [only] : Object.keys(this.loaded.services); + for (const serviceId of ids) { + const spec = this.loaded.services[serviceId]; + if (!spec?.managed || !spec.enabled) continue; + for (const rec of [...this.instances.values()].filter((i) => i.service === serviceId)) { + if (!rec.pid && !rec.stopping) { + this.instances.delete(rec.id); + continue; + } + if (rec.pid) await this.probe(rec); + if (rec.pid && rec.healthy === false && !rec.paused) { + this.log('unhealthy', { instance: rec.id, service: serviceId }); + await this.restartInstance(rec.id); + actions.push({ op: 'restart', instance: rec.id }); + } + } + const avail = this.available(serviceId).length; + const want = spec.keepFloor ? spec.min : Math.min(spec.min, spec.max); + if (avail < want) { + const need = want - avail; + for (let n = 0; n < need; n += 1) { + const started = await this.startOne(serviceId); + if (started) actions.push({ op: 'start', instance: started.id }); + } + } + } + return actions; + } + + async startService(serviceId) { + const spec = this.spec(serviceId); + spec.enabled = true; + const have = [...this.instances.values()].filter((i) => i.service === serviceId && i.pid).length; + const want = spec.min || 1; + const started = []; + for (let i = have; i < want; i += 1) { + const rec = await this.startOne(serviceId); + if (rec) started.push(rec.id); + } + return started; + } + + async stopService(serviceId, { holdFloor = false } = {}) { + const spec = this.spec(serviceId); + if (!holdFloor) spec.enabled = false; + const ids = [...this.instances.values()].filter((i) => i.service === serviceId).map((i) => i.id); + for (const id of ids) await this.stopInstance(id, { replace: false }); + if (holdFloor) await this.reconcile(serviceId); + return ids; + } + + async pauseService(serviceId) { + const ids = [...this.instances.values()].filter((i) => i.service === serviceId && i.pid).map((i) => i.id); + for (const id of ids) await this.pauseInstance(id, { replace: false }); + await this.reconcile(serviceId); + return ids; + } + + async resumeService(serviceId) { + const ids = [...this.instances.values()].filter((i) => i.service === serviceId).map((i) => i.id); + for (const id of ids) await this.resumeInstance(id); + return ids; + } + + async stopAll() { + for (const id of [...this.instances.keys()]) { + await this.stopInstance(id, { replace: false }); + } + } + + status() { + const services = {}; + for (const spec of Object.values(this.loaded.services)) { + const inst = [...this.instances.values()].filter((i) => i.service === spec.id); + const available = inst.filter((i) => i.pid && !i.paused && i.healthy !== false).length; + services[spec.id] = { + id: spec.id, + title: spec.title, + configPath: spec.configPath, + managed: spec.managed, + enabled: spec.enabled, + min: spec.min, + max: spec.max, + keepFloor: spec.keepFloor, + running: inst.filter((i) => i.pid).length, + paused: inst.filter((i) => i.paused).length, + available, + belowFloor: spec.keepFloor && spec.enabled && spec.managed && available < spec.min, + instances: inst.map((i) => ({ + id: i.id, + pid: i.pid, + machine: i.machine || 'local', + host: i.host || '127.0.0.1', + healthPort: i.healthPort, + paused: i.paused, + healthy: i.healthy, + restarts: i.restarts, + rtt: i.rtt || summarizeRtt(i.rttSamples || []), + })), + }; + const allSamples = inst.flatMap((i) => i.rttSamples || []); + services[spec.id].rtt = summarizeRtt(allSamples); + services[spec.id].health = classifyService(services[spec.id]); + } + const machines = this.machines().map((m) => { + const inst = [...this.instances.values()].filter((i) => i.machine === m.id); + const samples = inst.flatMap((i) => i.rttSamples || []); + return { + ...m, + running: inst.filter((i) => i.pid).length, + available: inst.filter((i) => i.pid && !i.paused && i.healthy !== false).length, + rtt: summarizeRtt(samples), + }; + }); + return { + control: this.loaded.control, + nats: this.loaded.nats, + machines, + services, + events: this.events.slice(-50), + }; + } +} + +function waitExit(child, ms) { + return new Promise((resolve) => { + if (!child || child.exitCode != null) return resolve(); + const t = setTimeout(() => { + try { + child.kill('SIGKILL'); + } catch { + /* ignore */ + } + resolve(); + }, ms); + child.once('exit', () => { + clearTimeout(t); + resolve(); + }); + }); +} + +async function waitForHealth(rec, ms) { + const deadline = Date.now() + ms; + while (Date.now() < deadline) { + let r; + if (rec.kind === 'ssh' && rec._machine) { + try { + const body = await sshHttp(rec._machine, 'GET', '/health', { port: rec.healthPort }); + r = { ok: body.ok !== false, statusCode: 200, paused: Boolean(body.paused), body }; + } catch { + r = { ok: false }; + } + } else { + r = await httpProbe({ + host: rec.host || '127.0.0.1', + port: rec.healthPort, + path: '/health', + timeoutMs: 400, + }); + } + if (r.ok || r.statusCode === 503) return r; + await new Promise((res) => setTimeout(res, 80)); + } + return { ok: false, error: 'start-timeout' }; +} diff --git a/src/worker.js b/src/worker.js new file mode 100644 index 0000000..bc010a2 --- /dev/null +++ b/src/worker.js @@ -0,0 +1,153 @@ +#!/usr/bin/env node +/** + * Fleet-managed replica: health HTTP + pause + heartbeat. + * Roles tree-node / archive-worm hold an in-process WormArchive. + */ +import http from 'node:http'; +import fs from 'node:fs'; +import path from 'node:path'; +import { createHash, randomBytes } from 'node:crypto'; +import { fileURLToPath } from 'node:url'; +import { RttWindow } from './rtt.js'; + +const role = process.env.FLEET_ROLE || 'unknown'; +const service = process.env.FLEET_SERVICE || role; +const instance = process.env.FLEET_INSTANCE || `${service}-0`; +const machine = process.env.FLEET_MACHINE || 'local'; +const port = Number(process.env.FLEET_HEALTH_PORT || 0); +const bind = process.env.FLEET_HEALTH_BIND || '127.0.0.1'; +const stateDir = process.env.FLEET_STATE_DIR || path.join(process.cwd(), 'data', instance); +const startedAt = new Date().toISOString(); +const rtt = new RttWindow(400); +const hz = Number(process.env.FLEET_MESSAGE_HZ || 6); + +fs.mkdirSync(stateDir, { recursive: true }); + +let paused = false; +let forceUnhealthy = false; +let archive = null; +let puts = 0; + +if (role === 'tree-node' || role === 'archive-worm') { + try { + const wormPath = path.resolve( + path.dirname(fileURLToPath(import.meta.url)), + '../../verae-archive-worm/src/archive.js', + ); + const { WormArchive } = await import(wormPath); + archive = new WormArchive(instance); + } catch { + archive = { + archiveId: instance, + put() { + puts += 1; + }, + query() { + return null; + }, + }; + } +} + +function processMessage() { + const t0 = process.hrtime.bigint(); + if (paused) return null; + const payload = randomBytes(32); + if (archive) { + const sha = createHash('sha256').update(payload).digest('hex'); + archive.put({ sha256: sha, kind: role === 'tree-node' ? 'tree' : 'publicMeta', record: { n: puts } }); + archive.query(sha); + puts += 1; + } else { + createHash('sha256').update(payload).digest('hex'); + puts += 1; + } + const ms = Number(process.hrtime.bigint() - t0) / 1e6; + rtt.add(ms); + return ms; +} + +function snapshot() { + return { + ok: !paused && !forceUnhealthy, + paused, + forceUnhealthy, + role, + service, + instance, + machine, + port, + pid: process.pid, + startedAt, + puts, + archiveId: archive?.archiveId || null, + rtt: rtt.stats(), + }; +} + +function beat() { + const body = { ...snapshot(), ts: Date.now() }; + fs.writeFileSync(path.join(stateDir, 'heartbeat.json'), JSON.stringify(body)); +} + +const server = http.createServer((req, res) => { + const url = new URL(req.url, 'http://127.0.0.1'); + const json = (code, obj) => { + res.writeHead(code, { 'content-type': 'application/json' }); + res.end(JSON.stringify(obj)); + }; + if (url.pathname === '/health' && req.method === 'GET') { + const snap = snapshot(); + return json(snap.ok ? 200 : 503, snap); + } + if (url.pathname === '/info' && req.method === 'GET') return json(200, snapshot()); + if (url.pathname === '/pause' && req.method === 'POST') { + paused = true; + beat(); + return json(200, snapshot()); + } + if (url.pathname === '/resume' && req.method === 'POST') { + paused = false; + forceUnhealthy = false; + beat(); + return json(200, snapshot()); + } + if (url.pathname === '/unhealthy' && req.method === 'POST') { + forceUnhealthy = true; + beat(); + return json(200, snapshot()); + } + if (url.pathname === '/message' && req.method === 'POST') { + const ms = processMessage(); + return json(paused ? 503 : 200, { rttMs: ms, ...snapshot() }); + } + if (url.pathname === '/metrics' && req.method === 'GET') { + return json(200, { rtt: rtt.stats(), samples: rtt.samples.slice(-80) }); + } + json(404, { error: 'not found' }); +}); + +server.listen(port, bind, () => { + beat(); + process.stdout.write(`fleet-worker ${instance}@${machine} health http://${bind}:${port}/health\n`); +}); + +const iv = setInterval(beat, 400); +const msgIv = + hz > 0 + ? setInterval(() => { + if (!paused && !forceUnhealthy) processMessage(); + }, Math.max(50, Math.round(1000 / hz))) + : null; +function shutdown() { + clearInterval(iv); + if (msgIv) clearInterval(msgIv); + try { + server.close(); + } catch { + /* ignore */ + } + process.exit(0); +} +process.on('SIGTERM', shutdown); +process.on('SIGINT', shutdown); diff --git a/test/classify.test.js b/test/classify.test.js new file mode 100644 index 0000000..7b5b083 --- /dev/null +++ b/test/classify.test.js @@ -0,0 +1,36 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { classifyService } from '../src/classify.js'; + +describe('classifyService', () => { + it('operational when available meets min and none paused', () => { + assert.equal( + classifyService({ min: 3, available: 3, running: 3, paused: 0, instances: [{ pid: 1, healthy: true }] }), + 'operational', + ); + }); + + it('degraded when a replica is paused even if floor is still met', () => { + assert.equal( + classifyService({ + min: 3, + available: 3, + running: 4, + paused: 1, + instances: [ + { pid: 1, paused: true, healthy: false }, + { pid: 2, paused: false, healthy: true }, + ], + }), + 'degraded', + ); + }); + + it('not operational when min>0 and nothing available', () => { + assert.equal(classifyService({ min: 1, available: 0, running: 0, paused: 0, instances: [] }), 'down'); + }); + + it('optional idle (min 0, not running) is operational', () => { + assert.equal(classifyService({ min: 0, available: 0, running: 0, paused: 0 }), 'operational'); + }); +}); diff --git a/test/fleet.test.js b/test/fleet.test.js new file mode 100644 index 0000000..c680cc0 --- /dev/null +++ b/test/fleet.test.js @@ -0,0 +1,141 @@ +import { describe, it, after } from 'node:test'; +import assert from 'node:assert/strict'; +import { loadFleet, listServices, FLEET_ROOT } from '../src/load.js'; +import { Supervisor } from '../src/supervisor.js'; +import { Monitor } from '../src/monitor.js'; + +function treeOnly(healthBase) { + const loaded = loadFleet(FLEET_ROOT, { overlay: false }); + for (const s of Object.values(loaded.services)) { + s.enabled = s.id === 'tree-node'; + if (s.id === 'tree-node') { + s.min = 3; + s.max = 6; + s.keepFloor = true; + s.ports = { healthBase }; + } + } + loaded.machinesOverlay = null; + loaded.machines = loaded.machines.filter((m) => m.kind === 'local' && m.enabled); + if (!loaded.machines.length) { + loaded.machines = [ + { id: 'local', title: 'local', kind: 'local', host: '127.0.0.1', enabled: true, capacity: 32, roles: ['*'] }, + ]; + } + return loaded; +} + +describe('verae-fleet', () => { + /** @type {Supervisor[]} */ + const supervisors = []; + after(async () => { + for (const s of supervisors) await s.stopAll(); + }); + + it('lists every service with a config file and central replica spec', () => { + const loaded = loadFleet(); + const rows = listServices(loaded); + const ids = rows.map((r) => r.id); + for (const need of [ + 'nats', + 'zappier-edge', + 'middleware-http', + 'job-poller', + 'webhook-deliver', + 'archive-aggregator', + 'archive-worm', + 'tree-node', + 'zapier-simulator', + 'zapier-platform-app', + ]) { + assert.ok(ids.includes(need), `missing ${need}`); + } + const tree = loaded.services['tree-node']; + assert.equal(tree.min, 3); + assert.equal(tree.keepFloor, true); + assert.equal(tree.configPath, 'services/tree-node.json'); + assert.equal(loaded.fleetPath, 'fleet.json'); + }); + + it('starts the tree-node floor (3 available copies)', async () => { + const sup = new Supervisor({ loaded: treeOnly(14600) }); + supervisors.push(sup); + await sup.startService('tree-node'); + const st = sup.status().services['tree-node']; + assert.equal(st.available, 3); + assert.equal(st.belowFloor, false); + assert.equal(st.instances.length, 3); + }); + + it('pauses one tree-node and respawns so min available stays 3', async () => { + const sup = new Supervisor({ loaded: treeOnly(14700) }); + supervisors.push(sup); + await sup.startService('tree-node'); + await sup.pauseInstance('tree-node-0'); + const st = sup.status().services['tree-node']; + assert.ok(st.available >= 3, `available ${st.available}`); + assert.ok(st.paused >= 1); + assert.ok(st.running >= 4); + assert.equal(st.belowFloor, false); + }); + + it('restarts an unhealthy replica and keeps the floor', async () => { + const sup = new Supervisor({ loaded: treeOnly(14800) }); + supervisors.push(sup); + await sup.startService('tree-node'); + const before = sup.instances.get('tree-node-1').pid; + await sup.markUnhealthy('tree-node-1'); + const mon = new Monitor(sup, { intervalMs: 50 }); + await mon.tick(); + const rec = sup.instances.get('tree-node-1'); + assert.ok(rec.pid); + assert.notEqual(rec.pid, before); + assert.equal(rec.healthy, true); + assert.equal(sup.status().services['tree-node'].available, 3); + }); + + it('stop+replace of a replica keeps min tree-nodes available', async () => { + const sup = new Supervisor({ loaded: treeOnly(14900) }); + supervisors.push(sup); + await sup.startService('tree-node'); + await sup.stopInstance('tree-node-2', { replace: true }); + const st = sup.status().services['tree-node']; + assert.equal(st.available, 3); + assert.ok(st.instances.some((i) => i.id === 'tree-node-2' || i.id === 'tree-node-0')); + }); + + it('spreads replicas across two defined machines', async () => { + const loaded = treeOnly(15100); + loaded.services['tree-node'].min = 4; + loaded.services['tree-node'].max = 6; + loaded.machines = [ + { id: 'rack-a', title: 'A', kind: 'local', host: '127.0.0.1', enabled: true, capacity: 8, roles: ['*'] }, + { id: 'rack-b', title: 'B', kind: 'local', host: '127.0.0.1', enabled: true, capacity: 8, roles: ['*'] }, + ]; + const sup = new Supervisor({ loaded }); + supervisors.push(sup); + await sup.startService('tree-node'); + const hosts = new Set(sup.status().services['tree-node'].instances.map((i) => i.machine)); + assert.ok(hosts.has('rack-a') && hosts.has('rack-b'), `hosts ${[...hosts]}`); + const machines = sup.status().machines; + assert.equal(machines.length, 2); + assert.ok(machines.every((m) => m.running >= 1)); + }); + + it('records message-processing RTT percentiles', async () => { + const loaded = treeOnly(15200); + loaded.services['tree-node'].min = 1; + loaded.services['tree-node'].max = 2; + const sup = new Supervisor({ loaded }); + supervisors.push(sup); + await sup.startService('tree-node'); + const rec = [...sup.instances.values()][0]; + for (let i = 0; i < 8; i += 1) await sup.probe(rec); + const rtt = rec.rtt; + assert.ok(rtt.count >= 8, `count ${rtt.count}`); + assert.ok(rtt.minMs <= rtt.p50Ms && rtt.p50Ms <= rtt.p90Ms); + const svc = sup.status().services['tree-node'].rtt; + assert.ok(svc.count >= 8); + assert.ok(svc.minMs != null && svc.avgMs != null); + }); +}); diff --git a/test/rtt.test.js b/test/rtt.test.js new file mode 100644 index 0000000..def37cf --- /dev/null +++ b/test/rtt.test.js @@ -0,0 +1,42 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { summarizeRtt, RttWindow } from '../src/rtt.js'; +import { pickMachine } from '../src/machines.js'; + +describe('RTT planning stats', () => { + it('computes min, avg, p50, p90 in order', () => { + const s = summarizeRtt([10, 20, 30, 40, 50, 60, 70, 80, 90, 100]); + assert.equal(s.count, 10); + assert.equal(s.minMs, 10); + assert.equal(s.avgMs, 55); + assert.ok(s.p50Ms >= s.minMs && s.p50Ms <= s.p90Ms); + assert.ok(s.p90Ms <= 100); + }); + + it('empty window is nulls', () => { + const s = new RttWindow().stats(); + assert.equal(s.count, 0); + assert.equal(s.minMs, null); + }); +}); + +describe('machine placement', () => { + it('spreads onto the least-loaded eligible host', () => { + const machines = [ + { id: 'a', enabled: true, capacity: 4, roles: ['*'] }, + { id: 'b', enabled: true, capacity: 4, roles: ['*'] }, + ]; + const instances = [{ machine: 'a', pid: 1 }]; + const pick = pickMachine(machines, instances, 'tree-node', 'tree-node'); + assert.equal(pick.id, 'b'); + }); + + it('skips machines that do not allow the role', () => { + const machines = [ + { id: 'edge', enabled: true, capacity: 8, roles: ['zappier-edge'] }, + { id: 'trees', enabled: true, capacity: 8, roles: ['tree-node'] }, + ]; + const pick = pickMachine(machines, [], 'tree-node', 'tree-node'); + assert.equal(pick.id, 'trees'); + }); +}); diff --git a/test/ssh.test.js b/test/ssh.test.js new file mode 100644 index 0000000..624bd74 --- /dev/null +++ b/test/ssh.test.js @@ -0,0 +1,56 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { expandHome, shellQuote, sshBaseArgs, sshTarget } from '../src/ssh.js'; +import { normalizeMachine, pickMachine } from '../src/machines.js'; + +describe('SSH remote machine config', () => { + it('expands ~ in identityFile and quotes remote commands', () => { + assert.equal(expandHome('~/foo'), `${os.homedir()}/foo`); + assert.equal(shellQuote("a'b"), `'a'"'"'b'`); + }); + + it('builds ssh argv with user, host, port, and key path (not key bytes)', () => { + const key = path.join(os.tmpdir(), 'verae-fleet-test-id'); + fs.writeFileSync(key, 'not-a-real-key\n', { mode: 0o600 }); + const m = normalizeMachine({ + id: 'ns1', + kind: 'ssh', + host: '70.88.205.138', + user: 'marchon', + sshPort: 22, + identityFile: key, + }); + assert.equal(m.kind, 'ssh'); + assert.equal(sshTarget(m), 'marchon@70.88.205.138'); + const args = sshBaseArgs(m); + assert.ok(args.includes('-i')); + assert.ok(args.includes(key)); + assert.ok(!args.some((a) => a.includes('BEGIN'))); + assert.equal(m.user, 'marchon'); + fs.unlinkSync(key); + }); + + it('places load on ssh hosts when they have free capacity', () => { + const machines = [ + normalizeMachine({ id: 'local', kind: 'local', host: '127.0.0.1', capacity: 2, roles: ['*'] }), + normalizeMachine({ + id: 'ns1', + kind: 'ssh', + host: '70.88.205.138', + user: 'marchon', + identityFile: '~/.ssh/id_ed25519', + capacity: 8, + roles: ['tree-node'], + }), + ]; + const instances = [ + { machine: 'local', pid: 1 }, + { machine: 'local', pid: 2 }, + ]; + const pick = pickMachine(machines, instances, 'tree-node', 'tree-node'); + assert.equal(pick.id, 'ns1'); + }); +});