From 100a35d5895011ac2b956483a51e8f57f1708bd7 Mon Sep 17 00:00:00 2001 From: George Lambert Date: Fri, 11 Sep 2026 19:55:20 -0400 Subject: [PATCH] Initial import of verae-keep from zapier monorepo --- NATS.md | 3 + README.md | 39 +++++ SUMMARY.md | 3 + package.json | 14 ++ scripts/guard.sh | 14 ++ src/expand.js | 9 ++ src/intent.js | 67 +++++++++ src/keep.js | 237 +++++++++++++++++++++++++++++++ src/probe.js | 83 +++++++++++ src/spawn.js | 48 +++++++ src/units.js | 52 +++++++ src/watch.js | 115 +++++++++++++++ systemd/verae-keep-guard.service | 15 ++ test/intent.test.js | 33 +++++ test/keep.test.js | 111 +++++++++++++++ units.json | 7 + units.ns1.json | 150 +++++++++++++++++++ 17 files changed, 1000 insertions(+) create mode 100644 NATS.md create mode 100644 README.md create mode 100644 SUMMARY.md create mode 100644 package.json create mode 100755 scripts/guard.sh create mode 100644 src/expand.js create mode 100644 src/intent.js create mode 100644 src/keep.js create mode 100644 src/probe.js create mode 100644 src/spawn.js create mode 100644 src/units.js create mode 100644 src/watch.js create mode 100644 systemd/verae-keep-guard.service create mode 100644 test/intent.test.js create mode 100644 test/keep.test.js create mode 100644 units.json create mode 100644 units.ns1.json diff --git a/NATS.md b/NATS.md new file mode 100644 index 0000000..7f80ff9 --- /dev/null +++ b/NATS.md @@ -0,0 +1,3 @@ +# NATS — verae-keep + +Keep does **not** subscribe to NATS. It probes HTTP `/health` (and TCP `:4222` for NATS as observe-only). Pause/stop intent is files + fleet HTTP, not subjects. diff --git a/README.md b/README.md new file mode 100644 index 0000000..4deb268 --- /dev/null +++ b/README.md @@ -0,0 +1,39 @@ +# verae-keep + +Host keep-alive for Verae workers. Restarts a unit that crashed **unless** the operator console paused or stopped it. + +**Forgejo:** https://git.georgelambert.org/marchon/verae-keep + +Two processes: + +| Process | Port | Job | +|---------|------|-----| +| `node src/keep.js` | `:3860` | Probe units, spawn if intent is `run` | +| `node src/watch.js` | `:3861` | Restart keep if `/health` fails | +| `scripts/guard.sh` | — | Restart watch if watch exits | + +Intent is `run` | `pause` | `stop`. Sources, newest wins: + +1. `POST /intent/:id` (admin / API) +2. `~/verae-fleet-runtime/data//intent.json` (fleet writes this on start/pause/stop) +3. live `FLEET_URL/api/status` when set +4. unit `defaultIntent` (`run`) + +Pause: leave the process down or POST `/pause`. Stop: SIGTERM and do not spawn. Run: spawn if health fails. + +## NS1 (70.88.205.138) + +```bash +KEEP_UNITS=$HOME/verae-keep/units.ns1.json KEEP_STATE=$HOME/verae-keep/data \ + nohup bash scripts/guard.sh >/tmp/verae-keep-guard.out 2>&1 & +``` + +Or systemd user unit `systemd/verae-keep-guard.service` (`loginctl enable-linger`). + +Units: NATS (observe only), job-poller, webhook-deliver, archive-aggregator, archive-worm ×3, tree-node ×3. + +## API + +- `GET /health` `GET /status` +- `POST /intent/:id` `{"state":"pause|stop|run"}` +- `POST /reconcile` diff --git a/SUMMARY.md b/SUMMARY.md new file mode 100644 index 0000000..5985428 --- /dev/null +++ b/SUMMARY.md @@ -0,0 +1,3 @@ +# verae-keep + +Host supervisor: restart crashed units unless the operator console paused or stopped them. Watch (`:3861`) restarts keep (`:3860`). `guard.sh` restarts watch. diff --git a/package.json b/package.json new file mode 100644 index 0000000..868b8e3 --- /dev/null +++ b/package.json @@ -0,0 +1,14 @@ +{ + "name": "verae-keep", + "version": "0.1.0", + "private": true, + "type": "module", + "description": "Host service keep-alive: restart crashed units unless the admin console paused or stopped them", + "scripts": { + "start": "node src/keep.js", + "watch": "node src/watch.js", + "guard": "bash scripts/guard.sh", + "test": "node --test test/*.test.js" + }, + "engines": { "node": ">=20" } +} diff --git a/scripts/guard.sh b/scripts/guard.sh new file mode 100755 index 0000000..46d106d --- /dev/null +++ b/scripts/guard.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +# Last-resort backup: if watch (and therefore keep) dies, start it again. +set -euo pipefail +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +STATE="${KEEP_STATE:-$ROOT/data}" +mkdir -p "$STATE" +cd "$ROOT" +export KEEP_UNITS="${KEEP_UNITS:-$ROOT/units.json}" +export KEEP_STATE="$STATE" +while true; do + echo "{\"t\":\"$(date -u +%Y-%m-%dT%H:%M:%SZ)\",\"event\":\"guard-start-watch\"}" >>"$STATE/guard.log" + node src/watch.js >>"$STATE/guard.out" 2>&1 || true + sleep 2 +done diff --git a/src/expand.js b/src/expand.js new file mode 100644 index 0000000..27af414 --- /dev/null +++ b/src/expand.js @@ -0,0 +1,9 @@ +import os from 'node:os'; +import path from 'node:path'; + +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; +} diff --git a/src/intent.js b/src/intent.js new file mode 100644 index 0000000..c6d42e6 --- /dev/null +++ b/src/intent.js @@ -0,0 +1,67 @@ +import fs from 'node:fs'; +import path from 'node:path'; + +export const STATES = ['run', 'pause', 'stop']; + +export function emptyStore() { + return { units: {} }; +} + +export function loadStore(file) { + try { + return JSON.parse(fs.readFileSync(file, 'utf8')); + } catch { + return emptyStore(); + } +} + +export function saveStore(file, store) { + fs.mkdirSync(path.dirname(file), { recursive: true }); + const tmp = `${file}.tmp`; + fs.writeFileSync(tmp, JSON.stringify(store, null, 2)); + fs.renameSync(tmp, file); +} + +export function record(store, id, state, source = 'keep') { + if (!STATES.includes(state)) throw new Error(`invalid intent ${state}`); + store.units[id] = { state, source, updatedAt: new Date().toISOString() }; + return store.units[id]; +} + +export function readUnitFile(file) { + try { + const j = JSON.parse(fs.readFileSync(file, 'utf8')); + if (!STATES.includes(j.state)) return null; + return { + state: j.state, + source: j.source || 'file', + updatedAt: j.updatedAt || '', + }; + } catch { + return null; + } +} + +export function writeUnitFile(file, id, state, source = 'keep') { + fs.mkdirSync(path.dirname(file), { recursive: true }); + const body = { id, state, source, updatedAt: new Date().toISOString() }; + fs.writeFileSync(file, JSON.stringify(body, null, 2)); + return body; +} + +/** Newest updatedAt wins. Missing timestamps lose. */ +export function pickIntent(...candidates) { + const ok = candidates.filter((c) => c && STATES.includes(c.state)); + if (!ok.length) return null; + ok.sort((a, b) => String(b.updatedAt || '').localeCompare(String(a.updatedAt || ''))); + return ok[0]; +} + +export function fromFleetService(sv, instanceId) { + if (!sv) return null; + if (sv.enabled === false) return { state: 'stop', source: 'fleet', updatedAt: new Date().toISOString() }; + const inst = (sv.instances || []).find((i) => i.id === instanceId); + if (inst?.paused) return { state: 'pause', source: 'fleet', updatedAt: new Date().toISOString() }; + if (sv.enabled) return { state: 'run', source: 'fleet', updatedAt: new Date().toISOString() }; + return null; +} diff --git a/src/keep.js b/src/keep.js new file mode 100644 index 0000000..5a241a4 --- /dev/null +++ b/src/keep.js @@ -0,0 +1,237 @@ +#!/usr/bin/env node +/** + * Host keep-alive. Restarts units that should be running. + * Honors admin pause/stop (intent files + optional fleet /api/status). + */ +import fs from 'node:fs'; +import http from 'node:http'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { expandHome } from './expand.js'; +import { + loadStore, + saveStore, + record, + pickIntent, + writeUnitFile, + fromFleetService, +} from './intent.js'; +import { httpProbe, httpPost, tcpProbe } from './probe.js'; +import { spawnUnit, killPid } from './spawn.js'; +import { loadUnitsFile, scanRuntimeIntents } from './units.js'; + +const HERE = path.dirname(fileURLToPath(import.meta.url)); +const ROOT = path.join(HERE, '..'); +const UNITS_FILE = expandHome(process.env.KEEP_UNITS || path.join(ROOT, 'units.json')); +const STATE_DIR = expandHome(process.env.KEEP_STATE || path.join(ROOT, 'data')); +const STORE_FILE = path.join(STATE_DIR, 'intent.json'); +const LOG_DIR = path.join(STATE_DIR, 'logs'); +const FLEET_URL = (process.env.FLEET_URL || '').replace(/\/$/, ''); + +const cfg = loadUnitsFile(UNITS_FILE); +const store = loadStore(STORE_FILE); +const last = {}; +const restarts = {}; + +function log(event, extra = {}) { + const line = JSON.stringify({ t: new Date().toISOString(), event, ...extra }); + fs.mkdirSync(STATE_DIR, { recursive: true }); + fs.appendFileSync(path.join(STATE_DIR, 'keep.log'), `${line}\n`); + process.stdout.write(`${line}\n`); +} + +function unitById(id) { + return cfg.units.find((u) => u.id === id); +} + +function intentOf(unit, runtimeIntents, fleetMap) { + const fromStore = store.units[unit.id]; + const fromRuntime = runtimeIntents[unit.id]; + const fromFleet = fleetMap[unit.id]; + const picked = pickIntent(fromStore, fromRuntime, fromFleet); + if (picked) return picked; + return { state: unit.defaultIntent || 'run', source: 'default', updatedAt: '' }; +} + +async function probe(unit) { + if (unit.healthUrl) return httpProbe(unit.healthUrl); + if (unit.tcp) return tcpProbe(unit.tcp.host || '127.0.0.1', Number(unit.tcp.port)); + return { ok: false, paused: false, error: 'no probe' }; +} + +function healthBase(unit) { + if (!unit.healthUrl) return null; + try { + const u = new URL(unit.healthUrl); + return `${u.protocol}//${u.host}`; + } catch { + return null; + } +} + +function bumpRestart(id) { + const hour = Math.floor(Date.now() / 3600000); + const slot = restarts[id] || { hour, n: 0 }; + if (slot.hour !== hour) { + slot.hour = hour; + slot.n = 0; + } + slot.n += 1; + restarts[id] = slot; + return slot.n <= 40; +} + +async function apply(unit, intent) { + const p = await probe(unit); + last[unit.id] = { intent, probe: { ok: p.ok, paused: p.paused, error: p.error || null, pid: p.pid || null } }; + const base = healthBase(unit); + if (!unit.managed) { + last[unit.id].action = p.ok ? 'observe' : 'down-unmanaged'; + return last[unit.id]; + } + if (intent.state === 'stop') { + if (p.ok || p.paused || p.pid) { + killPid(p.pid || last[unit.id]?.spawnedPid); + last[unit.id].action = 'killed'; + log('stop', { id: unit.id }); + } else last[unit.id].action = 'already-stopped'; + return last[unit.id]; + } + if (intent.state === 'pause') { + if (p.ok && !p.paused && base) { + await httpPost(`${base}${unit.pausePath}`); + last[unit.id].action = 'paused'; + log('pause', { id: unit.id }); + } else if (!p.ok && !p.paused) { + last[unit.id].action = 'leave-down'; + } else last[unit.id].action = 'already-paused'; + return last[unit.id]; + } + // run + if (p.ok) { + last[unit.id].action = 'healthy'; + return last[unit.id]; + } + if (p.paused && base) { + await httpPost(`${base}${unit.resumePath}`); + last[unit.id].action = 'resumed'; + log('resume', { id: unit.id }); + return last[unit.id]; + } + if (!unit.command) { + last[unit.id].action = 'down-no-command'; + return last[unit.id]; + } + if (!bumpRestart(unit.id)) { + last[unit.id].action = 'restart-capped'; + log('restart-capped', { id: unit.id }); + return last[unit.id]; + } + const spawned = spawnUnit(unit, { logDir: LOG_DIR }); + last[unit.id].spawnedPid = spawned.pid; + last[unit.id].action = 'spawned'; + log('spawn', { id: unit.id, pid: spawned.pid }); + return last[unit.id]; +} + +async function fleetMap() { + const map = {}; + if (!FLEET_URL) return map; + try { + const r = await httpProbe(`${FLEET_URL}/api/status`, { timeoutMs: 2500 }); + const services = r.body?.services || {}; + for (const unit of cfg.units) { + const sid = unit.env?.FLEET_SERVICE || unit.id.replace(/-\d+$/, ''); + const sv = services[sid]; + const via = fromFleetService(sv, unit.id); + if (via) map[unit.id] = via; + } + } catch { + /* fleet optional */ + } + return map; +} + +async function tick() { + const runtimeIntents = cfg.runtimeDir ? scanRuntimeIntents(cfg.runtimeDir) : {}; + const fromFleet = await fleetMap(); + const actions = []; + for (const unit of cfg.units) { + const intent = intentOf(unit, runtimeIntents, fromFleet); + const row = await apply(unit, intent); + actions.push({ id: unit.id, intent: intent.state, source: intent.source, action: row.action }); + } + return { t: new Date().toISOString(), actions }; +} + +function setIntent(id, state, source = 'admin') { + const unit = unitById(id); + if (!unit) throw Object.assign(new Error(`unknown unit ${id}`), { status: 404 }); + const row = record(store, id, state, source); + saveStore(STORE_FILE, store); + if (cfg.runtimeDir) { + writeUnitFile(path.join(cfg.runtimeDir, 'data', id, 'intent.json'), id, state, source); + } + log('intent', { id, state, source }); + return row; +} + +const server = http.createServer(async (req, res) => { + const url = new URL(req.url || '/', `http://127.0.0.1:${cfg.port}`); + const json = (code, obj) => { + res.writeHead(code, { 'content-type': 'application/json', 'cache-control': 'no-store' }); + res.end(JSON.stringify(obj, null, 2)); + }; + try { + if (req.method === 'GET' && url.pathname === '/health') { + return json(200, { ok: true, role: 'verae-keep', units: cfg.units.length, pid: process.pid }); + } + if (req.method === 'GET' && url.pathname === '/status') { + return json(200, { + ok: true, + units: cfg.units.map((u) => ({ + id: u.id, + title: u.title, + intent: store.units[u.id] || { state: u.defaultIntent, source: 'default' }, + last: last[u.id] || null, + })), + }); + } + const m = url.pathname.match(/^\/intent\/([^/]+)$/); + if (req.method === 'POST' && m) { + const chunks = []; + for await (const c of req) chunks.push(c); + const body = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}'); + const row = setIntent(decodeURIComponent(m[1]), body.state, body.source || 'admin'); + return json(200, row); + } + if (req.method === 'POST' && url.pathname === '/reconcile') { + return json(200, await tick()); + } + json(404, { error: 'not found' }); + } catch (err) { + json(err.status || 400, { error: err.message }); + } +}); + +const bind = process.env.KEEP_BIND || cfg.bind || '127.0.0.1'; +const port = Number(process.env.KEEP_PORT || cfg.port || 3860); + +if (process.argv[1] === fileURLToPath(import.meta.url)) { + server.listen(port, bind, () => { + log('listen', { url: `http://${bind}:${port}/`, units: cfg.units.map((u) => u.id) }); + }); + const iv = setInterval(() => { + tick().catch((err) => log('tick-error', { error: err.message })); + }, cfg.intervalMs); + tick().catch((err) => log('tick-error', { error: err.message })); + const stop = () => { + clearInterval(iv); + server.close(); + process.exit(0); + }; + process.on('SIGTERM', stop); + process.on('SIGINT', stop); +} + +export { tick, setIntent, cfg, store }; diff --git a/src/probe.js b/src/probe.js new file mode 100644 index 0000000..ebdf35a --- /dev/null +++ b/src/probe.js @@ -0,0 +1,83 @@ +import http from 'node:http'; +import net from 'node:net'; + +export function httpProbe(url, { timeoutMs = 1500 } = {}) { + return new Promise((resolve) => { + const u = new URL(url); + const req = http.request( + { + hostname: u.hostname, + port: u.port, + path: `${u.pathname}${u.search}`, + method: 'GET', + timeout: timeoutMs, + }, + (res) => { + const chunks = []; + res.on('data', (c) => chunks.push(c)); + res.on('end', () => { + const raw = Buffer.concat(chunks).toString('utf8'); + let body = {}; + try { + body = JSON.parse(raw); + } catch { + body = { raw }; + } + resolve({ + ok: res.statusCode >= 200 && res.statusCode < 300 && body.ok !== false, + paused: Boolean(body.paused), + statusCode: res.statusCode, + pid: body.pid || null, + body, + }); + }); + }, + ); + req.on('error', (err) => resolve({ ok: false, paused: false, error: err.message })); + req.on('timeout', () => { + req.destroy(); + resolve({ ok: false, paused: false, error: 'timeout' }); + }); + req.end(); + }); +} + +export function httpPost(url, { timeoutMs = 1500 } = {}) { + return new Promise((resolve) => { + const u = new URL(url); + const req = http.request( + { + hostname: u.hostname, + port: u.port, + path: `${u.pathname}${u.search}`, + method: 'POST', + timeout: timeoutMs, + }, + (res) => { + const chunks = []; + res.on('data', (c) => chunks.push(c)); + res.on('end', () => resolve({ ok: res.statusCode < 400, statusCode: res.statusCode })); + }, + ); + req.on('error', (err) => resolve({ ok: false, error: err.message })); + req.on('timeout', () => { + req.destroy(); + resolve({ ok: false, error: 'timeout' }); + }); + req.end(); + }); +} + +export function tcpProbe(host, port, { timeoutMs = 800 } = {}) { + return new Promise((resolve) => { + const sock = net.connect({ host, port, timeout: timeoutMs }, () => { + sock.end(); + resolve({ ok: true, paused: false }); + }); + sock.on('error', (err) => resolve({ ok: false, paused: false, error: err.message })); + sock.on('timeout', () => { + sock.destroy(); + resolve({ ok: false, paused: false, error: 'timeout' }); + }); + }); +} diff --git a/src/spawn.js b/src/spawn.js new file mode 100644 index 0000000..013448f --- /dev/null +++ b/src/spawn.js @@ -0,0 +1,48 @@ +import { spawn } from 'node:child_process'; +import fs from 'node:fs'; +import path from 'node:path'; +import { expandHome } from './expand.js'; + +export function spawnUnit(unit, { logDir } = {}) { + const cwd = expandHome(unit.cwd || process.cwd()); + const cmd = unit.command?.[0]; + if (!cmd) throw new Error(`unit ${unit.id} has no command`); + const args = unit.command.slice(1); + fs.mkdirSync(cwd, { recursive: true }); + let stdout = 'ignore'; + let stderr = 'ignore'; + if (logDir) { + fs.mkdirSync(logDir, { recursive: true }); + const log = fs.openSync(path.join(logDir, `${unit.id}.log`), 'a'); + stdout = log; + stderr = log; + } + const child = spawn(cmd, args, { + cwd, + env: { ...process.env, ...(unit.env || {}) }, + detached: true, + stdio: ['ignore', stdout, stderr], + }); + child.unref(); + return { pid: child.pid, child }; +} + +export function killPid(pid, signal = 'SIGTERM') { + if (!pid) return false; + try { + process.kill(Number(pid), signal); + return true; + } catch { + return false; + } +} + +export function pidAlive(pid) { + if (!pid) return false; + try { + process.kill(Number(pid), 0); + return true; + } catch { + return false; + } +} diff --git a/src/units.js b/src/units.js new file mode 100644 index 0000000..c512694 --- /dev/null +++ b/src/units.js @@ -0,0 +1,52 @@ +import fs from 'node:fs'; +import path from 'node:path'; +import { expandHome } from './expand.js'; +import { readUnitFile } from './intent.js'; + +export function loadUnitsFile(file) { + const raw = JSON.parse(fs.readFileSync(file, 'utf8')); + const runtimeDir = expandHome(raw.runtimeDir || ''); + return { + bind: raw.bind || '127.0.0.1', + port: Number(raw.port || 3860), + intervalMs: Number(raw.intervalMs || 2000), + runtimeDir, + units: (raw.units || []).map((u) => normalizeUnit(u, runtimeDir)), + }; +} + +function normalizeUnit(u, runtimeDir) { + const healthUrl = typeof u.health === 'string' ? u.health : u.health?.url; + const tcp = u.health?.tcp || (u.tcp ? { host: u.tcp.host, port: u.tcp.port } : null); + return { + id: u.id, + title: u.title || u.id, + healthUrl: healthUrl || null, + tcp: tcp || null, + cwd: expandHome(u.cwd || runtimeDir || process.cwd()), + command: u.command || null, + env: u.env || {}, + defaultIntent: u.defaultIntent || 'run', + managed: u.managed !== false, + pausePath: u.pausePath || '/pause', + resumePath: u.resumePath || '/resume', + }; +} + +export function scanRuntimeIntents(runtimeDir) { + const root = expandHome(runtimeDir); + const data = path.join(root, 'data'); + const out = {}; + let names = []; + try { + names = fs.readdirSync(data); + } catch { + return out; + } + for (const name of names) { + const file = path.join(data, name, 'intent.json'); + const intent = readUnitFile(file); + if (intent) out[name] = intent; + } + return out; +} diff --git a/src/watch.js b/src/watch.js new file mode 100644 index 0000000..2b99f99 --- /dev/null +++ b/src/watch.js @@ -0,0 +1,115 @@ +#!/usr/bin/env node +/** + * Backup for verae-keep. If keep's /health fails, spawn keep again. + */ +import { spawn } from 'node:child_process'; +import fs from 'node:fs'; +import http from 'node:http'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { httpProbe } from './probe.js'; +import { killPid, pidAlive } from './spawn.js'; +import { expandHome } from './expand.js'; + +const HERE = path.dirname(fileURLToPath(import.meta.url)); +const ROOT = path.join(HERE, '..'); +const KEEP_URL = (process.env.KEEP_URL || 'http://127.0.0.1:3860').replace(/\/$/, ''); +const STATE_DIR = expandHome(process.env.KEEP_STATE || path.join(ROOT, 'data')); +const PID_FILE = path.join(STATE_DIR, 'keep.pid'); +const INTERVAL = Number(process.env.WATCH_INTERVAL_MS || 2000); +const BIND = process.env.WATCH_BIND || '127.0.0.1'; +const PORT = Number(process.env.WATCH_PORT || 3861); + +let child = null; +let restarts = 0; + +function log(event, extra = {}) { + const line = JSON.stringify({ t: new Date().toISOString(), event, ...extra }); + fs.mkdirSync(STATE_DIR, { recursive: true }); + fs.appendFileSync(path.join(STATE_DIR, 'watch.log'), `${line}\n`); + process.stdout.write(`${line}\n`); +} + +function readPid() { + try { + return Number(fs.readFileSync(PID_FILE, 'utf8').trim()); + } catch { + return null; + } +} + +function writePid(pid) { + fs.mkdirSync(STATE_DIR, { recursive: true }); + fs.writeFileSync(PID_FILE, String(pid)); +} + +function startKeep() { + const existing = readPid(); + if (existing && pidAlive(existing)) { + log('adopt', { pid: existing }); + return existing; + } + const env = { ...process.env }; + const proc = spawn(process.execPath, [path.join(HERE, 'keep.js')], { + cwd: ROOT, + env, + detached: true, + stdio: ['ignore', 'pipe', 'pipe'], + }); + proc.unref(); + child = proc; + writePid(proc.pid); + restarts += 1; + log('start-keep', { pid: proc.pid, restarts }); + proc.stdout?.on('data', (c) => process.stdout.write(c)); + proc.stderr?.on('data', (c) => process.stderr.write(c)); + proc.on('exit', (code, signal) => { + log('keep-exit', { code, signal, pid: proc.pid }); + if (readPid() === proc.pid) { + try { + fs.unlinkSync(PID_FILE); + } catch { + /* ignore */ + } + } + }); + return proc.pid; +} + +async function tick() { + const h = await httpProbe(`${KEEP_URL}/health`, { timeoutMs: 1500 }); + if (h.ok) return { ok: true, keep: h.body }; + log('keep-down', { error: h.error || h.statusCode }); + const old = readPid(); + if (old && pidAlive(old)) killPid(old, 'SIGTERM'); + startKeep(); + return { ok: false, restarted: true }; +} + +const server = http.createServer(async (req, res) => { + const url = new URL(req.url || '/', `http://127.0.0.1:${PORT}`); + const json = (code, obj) => { + res.writeHead(code, { 'content-type': 'application/json' }); + res.end(JSON.stringify(obj)); + }; + if (url.pathname === '/health') { + return json(200, { ok: true, role: 'verae-keep-watch', pid: process.pid, keepPid: readPid(), restarts }); + } + if (url.pathname === '/tick' && req.method === 'POST') { + return json(200, await tick()); + } + json(404, { error: 'not found' }); +}); + +server.listen(PORT, BIND, () => { + log('listen', { url: `http://${BIND}:${PORT}/`, keep: KEEP_URL }); + startKeep(); + setInterval(() => { + tick().catch((err) => log('tick-error', { error: err.message })); + }, INTERVAL); +}); + +process.on('SIGTERM', () => { + if (child?.pid) killPid(child.pid, 'SIGTERM'); + process.exit(0); +}); diff --git a/systemd/verae-keep-guard.service b/systemd/verae-keep-guard.service new file mode 100644 index 0000000..75d69a8 --- /dev/null +++ b/systemd/verae-keep-guard.service @@ -0,0 +1,15 @@ +[Unit] +Description=Verae keep-alive guard (restarts watch + keep) +After=network.target + +[Service] +Type=simple +WorkingDirectory=%h/verae-keep +Environment=KEEP_UNITS=%h/verae-keep/units.ns1.json +Environment=KEEP_STATE=%h/verae-keep/data +ExecStart=/usr/bin/bash %h/verae-keep/scripts/guard.sh +Restart=always +RestartSec=2 + +[Install] +WantedBy=default.target diff --git a/test/intent.test.js b/test/intent.test.js new file mode 100644 index 0000000..ad01bc5 --- /dev/null +++ b/test/intent.test.js @@ -0,0 +1,33 @@ +import { test } 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 { emptyStore, record, pickIntent, writeUnitFile, readUnitFile, fromFleetService } from '../src/intent.js'; + +test('record and pick newest intent', () => { + const s = emptyStore(); + record(s, 'a', 'run', 'keep'); + const older = { state: 'stop', source: 'file', updatedAt: '2020-01-01T00:00:00Z' }; + const picked = pickIntent(s.units.a, older); + assert.equal(picked.state, 'run'); + assert.equal(picked.source, 'keep'); +}); + +test('unit file round-trip', () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'keep-intent-')); + const file = path.join(dir, 'intent.json'); + writeUnitFile(file, 'x', 'pause', 'admin'); + const got = readUnitFile(file); + assert.equal(got.state, 'pause'); + assert.equal(got.source, 'admin'); +}); + +test('fleet enabled false is stop; paused instance is pause', () => { + assert.equal(fromFleetService({ enabled: false, instances: [] }, 't-0').state, 'stop'); + assert.equal( + fromFleetService({ enabled: true, instances: [{ id: 't-0', paused: true }] }, 't-0').state, + 'pause', + ); + assert.equal(fromFleetService({ enabled: true, instances: [{ id: 't-0', paused: false }] }, 't-0').state, 'run'); +}); diff --git a/test/keep.test.js b/test/keep.test.js new file mode 100644 index 0000000..de1084f --- /dev/null +++ b/test/keep.test.js @@ -0,0 +1,111 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import http from 'node:http'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { spawn } from 'node:child_process'; +import { fileURLToPath } from 'node:url'; + +const root = path.join(path.dirname(fileURLToPath(import.meta.url)), '..'); + +function listen(handler) { + return new Promise((resolve) => { + const s = http.createServer(handler); + s.listen(0, '127.0.0.1', () => resolve(s)); + }); +} + +test('keep restarts a crashed unit and leaves a paused unit down', async () => { + const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'keep-lab-')); + const dummy = path.join(tmp, 'dummy.js'); + fs.writeFileSync( + dummy, + `import http from 'node:http'; +const port = Number(process.env.PORT); +let paused = false; +http.createServer((req, res) => { + const u = new URL(req.url, 'http://127.0.0.1'); + const body = JSON.stringify({ ok: !paused, paused, pid: process.pid }); + if (u.pathname === '/pause' && req.method === 'POST') paused = true; + if (u.pathname === '/resume' && req.method === 'POST') paused = false; + res.writeHead(paused ? 503 : 200, { 'content-type': 'application/json' }); + res.end(body); +}).listen(port, '127.0.0.1'); +`, + ); + const port = await new Promise((resolve) => { + const s = http.createServer(); + s.listen(0, '127.0.0.1', () => { + const p = s.address().port; + s.close(() => resolve(p)); + }); + }); + const keepPort = port + 1; + const units = { + bind: '127.0.0.1', + port: keepPort, + intervalMs: 400, + runtimeDir: tmp, + units: [ + { + id: 'dummy-0', + health: `http://127.0.0.1:${port}/health`, + cwd: tmp, + command: [process.execPath, dummy], + env: { PORT: String(port) }, + }, + ], + }; + const unitsFile = path.join(tmp, 'units.json'); + fs.writeFileSync(unitsFile, JSON.stringify(units)); + const child = spawn(process.execPath, ['src/keep.js'], { + cwd: root, + env: { ...process.env, KEEP_UNITS: unitsFile, KEEP_STATE: path.join(tmp, 'state'), KEEP_PORT: String(keepPort) }, + stdio: ['ignore', 'pipe', 'pipe'], + }); + const waitHealth = async (url, wantOk) => { + for (let i = 0; i < 25; i += 1) { + try { + const r = await fetch(url); + const b = await r.json(); + if (Boolean(b.ok) === wantOk) return b; + } catch { + if (!wantOk) return { ok: false }; + } + await new Promise((r) => setTimeout(r, 200)); + } + throw new Error(`timeout ${url} wantOk=${wantOk}`); + }; + try { + await waitHealth(`http://127.0.0.1:${keepPort}/health`, true); + const up = await waitHealth(`http://127.0.0.1:${port}/health`, true); + assert.equal(up.ok, true); + const pid1 = up.pid; + process.kill(pid1, 'SIGKILL'); + const up2 = await waitHealth(`http://127.0.0.1:${port}/health`, true); + assert.notEqual(up2.pid, pid1); + + await fetch(`http://127.0.0.1:${keepPort}/intent/dummy-0`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ state: 'stop' }), + }); + await new Promise((r) => setTimeout(r, 800)); + try { + process.kill(up2.pid, 'SIGKILL'); + } catch { + /* already stopped */ + } + await new Promise((r) => setTimeout(r, 900)); + let down = false; + try { + await fetch(`http://127.0.0.1:${port}/health`, { signal: AbortSignal.timeout(400) }); + } catch { + down = true; + } + assert.equal(down, true); + } finally { + child.kill('SIGTERM'); + } +}); diff --git a/units.json b/units.json new file mode 100644 index 0000000..7f4300d --- /dev/null +++ b/units.json @@ -0,0 +1,7 @@ +{ + "bind": "127.0.0.1", + "port": 3860, + "intervalMs": 2000, + "runtimeDir": "", + "units": [] +} diff --git a/units.ns1.json b/units.ns1.json new file mode 100644 index 0000000..07d3b5b --- /dev/null +++ b/units.ns1.json @@ -0,0 +1,150 @@ +{ + "bind": "127.0.0.1", + "port": 3860, + "intervalMs": 2000, + "runtimeDir": "~/verae-fleet-runtime", + "units": [ + { + "id": "nats", + "title": "NATS JetStream (loopback)", + "tcp": { "host": "127.0.0.1", "port": 4222 }, + "managed": false, + "defaultIntent": "run" + }, + { + "id": "job-poller-0", + "health": "http://127.0.0.1:13200/health", + "cwd": "~/verae-fleet-runtime", + "command": ["node", "src/worker.js"], + "env": { + "FLEET_ROLE": "job-poller", + "FLEET_SERVICE": "job-poller", + "FLEET_INSTANCE": "job-poller-0", + "FLEET_MACHINE": "ns1", + "FLEET_HEALTH_PORT": "13200", + "FLEET_HEALTH_BIND": "127.0.0.1", + "FLEET_STATE_DIR": "data/job-poller-0" + } + }, + { + "id": "webhook-deliver-0", + "health": "http://127.0.0.1:13300/health", + "cwd": "~/verae-fleet-runtime", + "command": ["node", "src/worker.js"], + "env": { + "FLEET_ROLE": "webhook-deliver", + "FLEET_SERVICE": "webhook-deliver", + "FLEET_INSTANCE": "webhook-deliver-0", + "FLEET_MACHINE": "ns1", + "FLEET_HEALTH_PORT": "13300", + "FLEET_HEALTH_BIND": "127.0.0.1", + "FLEET_STATE_DIR": "data/webhook-deliver-0" + } + }, + { + "id": "archive-aggregator-0", + "health": "http://127.0.0.1:13400/health", + "cwd": "~/verae-fleet-runtime", + "command": ["node", "src/worker.js"], + "env": { + "FLEET_ROLE": "archive-aggregator", + "FLEET_SERVICE": "archive-aggregator", + "FLEET_INSTANCE": "archive-aggregator-0", + "FLEET_MACHINE": "ns1", + "FLEET_HEALTH_PORT": "13400", + "FLEET_HEALTH_BIND": "127.0.0.1", + "FLEET_STATE_DIR": "data/archive-aggregator-0" + } + }, + { + "id": "archive-worm-0", + "health": "http://127.0.0.1:13500/health", + "cwd": "~/verae-fleet-runtime", + "command": ["node", "src/worker.js"], + "env": { + "FLEET_ROLE": "archive-worm", + "FLEET_SERVICE": "archive-worm", + "FLEET_INSTANCE": "archive-worm-0", + "FLEET_MACHINE": "ns1", + "FLEET_HEALTH_PORT": "13500", + "FLEET_HEALTH_BIND": "127.0.0.1", + "FLEET_STATE_DIR": "data/archive-worm-0" + } + }, + { + "id": "archive-worm-1", + "health": "http://127.0.0.1:13501/health", + "cwd": "~/verae-fleet-runtime", + "command": ["node", "src/worker.js"], + "env": { + "FLEET_ROLE": "archive-worm", + "FLEET_SERVICE": "archive-worm", + "FLEET_INSTANCE": "archive-worm-1", + "FLEET_MACHINE": "ns1", + "FLEET_HEALTH_PORT": "13501", + "FLEET_HEALTH_BIND": "127.0.0.1", + "FLEET_STATE_DIR": "data/archive-worm-1" + } + }, + { + "id": "archive-worm-2", + "health": "http://127.0.0.1:13502/health", + "cwd": "~/verae-fleet-runtime", + "command": ["node", "src/worker.js"], + "env": { + "FLEET_ROLE": "archive-worm", + "FLEET_SERVICE": "archive-worm", + "FLEET_INSTANCE": "archive-worm-2", + "FLEET_MACHINE": "ns1", + "FLEET_HEALTH_PORT": "13502", + "FLEET_HEALTH_BIND": "127.0.0.1", + "FLEET_STATE_DIR": "data/archive-worm-2" + } + }, + { + "id": "tree-node-0", + "health": "http://127.0.0.1:13600/health", + "cwd": "~/verae-fleet-runtime", + "command": ["node", "src/worker.js"], + "env": { + "FLEET_ROLE": "tree-node", + "FLEET_SERVICE": "tree-node", + "FLEET_INSTANCE": "tree-node-0", + "FLEET_MACHINE": "ns1", + "FLEET_HEALTH_PORT": "13600", + "FLEET_HEALTH_BIND": "127.0.0.1", + "FLEET_STATE_DIR": "data/tree-node-0" + } + }, + { + "id": "tree-node-1", + "health": "http://127.0.0.1:13601/health", + "cwd": "~/verae-fleet-runtime", + "command": ["node", "src/worker.js"], + "env": { + "FLEET_ROLE": "tree-node", + "FLEET_SERVICE": "tree-node", + "FLEET_INSTANCE": "tree-node-1", + "FLEET_MACHINE": "ns1", + "FLEET_HEALTH_PORT": "13601", + "FLEET_HEALTH_BIND": "127.0.0.1", + "FLEET_STATE_DIR": "data/tree-node-1" + } + }, + { + "id": "tree-node-2", + "health": "http://127.0.0.1:13602/health", + "cwd": "~/verae-fleet-runtime", + "command": ["node", "src/worker.js"], + "env": { + "FLEET_ROLE": "tree-node", + "FLEET_SERVICE": "tree-node", + "FLEET_INSTANCE": "tree-node-2", + "FLEET_MACHINE": "ns1", + "FLEET_HEALTH_PORT": "13602", + "FLEET_HEALTH_BIND": "127.0.0.1", + "FLEET_STATE_DIR": "data/tree-node-2" + } + } + ] +}