feat(discord): systemd user service with a supervised run; brakes exit 3 and are never retried (#1509)
QUEUE row 17, MVP iteration 2. scripts/discord-service.sh renders and installs mosaic-discord@<binding> from packages/discord/systemd/. The unit's main process is `run --supervised`, which applies the new recover policy first: a lock whose owner is gone is cleared and only the STOP written for that is removed; an operator STOP or a held binding refuses with exit 3, which RestartPreventExitStatus never retries. `recover` is also a CLI verb. First cut used ExecStartPre and looped live, since systemd honours the never-retry status only from the main process; replaced and re-verified before any message traffic. Suite 40/40, 95 node tests. Co-Authored-By: Claude Fable 5.1 <[email protected]>
This commit is contained in:
Claude Fable 5.1
@@ -1,9 +1,10 @@
|
||||
#!/usr/bin/env node
|
||||
// Usage:
|
||||
// mosaic-discord check <binding> [--config PATH] [--repo PATH]
|
||||
// mosaic-discord run <binding> [--config PATH] [--repo PATH]
|
||||
// mosaic-discord run <binding> [--config PATH] [--repo PATH] [--supervised]
|
||||
// mosaic-discord stop <binding> [--config PATH]
|
||||
// mosaic-discord unlock <binding> [--config PATH]
|
||||
// mosaic-discord recover <binding> [--config PATH]
|
||||
//
|
||||
// <binding> names <dataRoot>/discord/<binding>.json. The repository wrapper
|
||||
// is scripts/discord.sh.
|
||||
@@ -23,8 +24,18 @@
|
||||
// with unverifiable identity, or recorded in a file it cannot read. `run` never reclaims on its own, and a
|
||||
// claim that finds STOP after publishing releases itself, so unlock cannot
|
||||
// race a start. Remove STOP to run again.
|
||||
// recover: the supervised pre-start. Refuses while STOP is present or the
|
||||
// binding is held by a live or unverifiable process; clears a lock whose
|
||||
// owner is gone the way unlock does, then removes the STOP it wrote for
|
||||
// that, so the run that follows can claim. Never removes a STOP an operator
|
||||
// wrote. `run --supervised` does the same first thing itself; the service
|
||||
// unit (scripts/discord-service.sh) uses that form, because systemd only
|
||||
// honours a never-retry exit status from the main process, not from a
|
||||
// pre-start command.
|
||||
//
|
||||
// Exit codes: 0 ok; 1 operation failed; 2 invalid data or configuration; 4 usage.
|
||||
// Exit codes: 0 ok; 1 operation failed; 2 invalid data or configuration;
|
||||
// 3 refused by a brake (STOP present or the binding held; a supervisor must
|
||||
// not retry); 4 usage.
|
||||
|
||||
import { existsSync, mkdirSync, mkdtempSync, writeFileSync, statSync, readdirSync } from "node:fs";
|
||||
import { join, resolve } from "node:path";
|
||||
@@ -36,17 +47,18 @@ import { createGateway, CONNECTOR_INTENTS } from "./gateway.mjs";
|
||||
import { createEngine, buildPiArgs } from "./engine-pi.mjs";
|
||||
import { assembleContext } from "./context.mjs";
|
||||
import { createConnector } from "./connector.mjs";
|
||||
import { ensureJournal, requestStop, stopRequested, readPid, stopTarget, writePid, clearPid, unlock } from "./journal.mjs";
|
||||
import { ensureJournal, requestStop, stopRequested, readPid, stopTarget, writePid, clearPid, unlock, recover, BRAKE_EXIT } from "./journal.mjs";
|
||||
|
||||
const USAGE = [
|
||||
"usage: mosaic-discord check <binding> [--config PATH] [--repo PATH]",
|
||||
" mosaic-discord run <binding> [--config PATH] [--repo PATH]",
|
||||
" mosaic-discord run <binding> [--config PATH] [--repo PATH] [--supervised]",
|
||||
" mosaic-discord stop <binding> [--config PATH]",
|
||||
" mosaic-discord unlock <binding> [--config PATH]",
|
||||
" mosaic-discord recover <binding> [--config PATH]",
|
||||
].join("\n");
|
||||
|
||||
function parse(argv) {
|
||||
const opts = { command: null, binding: null, config: defaultConfigPath(), repo: process.cwd() };
|
||||
const opts = { command: null, binding: null, config: defaultConfigPath(), repo: process.cwd(), supervised: false };
|
||||
for (let i = 0; i < argv.length; i++) {
|
||||
const a = argv[i];
|
||||
if (a === "--config" || a === "--repo") {
|
||||
@@ -54,14 +66,17 @@ function parse(argv) {
|
||||
opts[a.slice(2)] = resolve(argv[++i]);
|
||||
} else if (a === "--help" || a === "-h") {
|
||||
opts.command = "help";
|
||||
} else if (a === "--supervised") {
|
||||
opts.supervised = true;
|
||||
} else if (a.startsWith("--")) throw new DiscordError(`unknown argument: ${a}\n${USAGE}`, 4);
|
||||
else if (opts.command === null) opts.command = a;
|
||||
else if (opts.binding === null) opts.binding = a;
|
||||
else throw new DiscordError(`unexpected argument: ${a}\n${USAGE}`, 4);
|
||||
}
|
||||
if (opts.command === "help") return opts;
|
||||
if (!["check", "run", "stop", "unlock"].includes(opts.command)) throw new DiscordError(USAGE, 4);
|
||||
if (!["check", "run", "stop", "unlock", "recover"].includes(opts.command)) throw new DiscordError(USAGE, 4);
|
||||
if (opts.binding === null) throw new DiscordError(`${opts.command} needs a binding name\n${USAGE}`, 4);
|
||||
if (opts.supervised && opts.command !== "run") throw new DiscordError(`--supervised applies to run only\n${USAGE}`, 4);
|
||||
return opts;
|
||||
}
|
||||
|
||||
@@ -136,7 +151,11 @@ async function run(opts) {
|
||||
const { binding, contextFiles, pi, journalDir, sessionDir } = prepare(opts);
|
||||
const token = readToken(binding);
|
||||
ensureJournal(journalDir);
|
||||
if (stopRequested(journalDir)) throw new DiscordError(`STOP is present in ${journalDir}; remove it to run`, 1);
|
||||
if (opts.supervised) {
|
||||
const outcome = recover(journalDir);
|
||||
if (outcome === "cleared") warn("supervised start: run.lock left by a process that is gone was removed");
|
||||
}
|
||||
if (stopRequested(journalDir)) throw new DiscordError(`STOP is present in ${journalDir}; remove it to run`, BRAKE_EXIT);
|
||||
writePid(journalDir, process.pid);
|
||||
const cleanupPid = () => clearPid(journalDir, process.pid);
|
||||
try {
|
||||
@@ -242,6 +261,16 @@ function unlockCommand(opts) {
|
||||
say("remove STOP to run again");
|
||||
}
|
||||
|
||||
function recoverCommand(opts) {
|
||||
const dataRoot = loadDataRoot(opts.config);
|
||||
const binding = loadBinding(bindingPath(dataRoot, opts.binding));
|
||||
const journalDir = bindingDataDir(dataRoot, binding.name);
|
||||
ensureJournal(journalDir);
|
||||
const outcome = recover(journalDir);
|
||||
if (outcome === "cleared") say("run.lock left by a process that is gone was removed; STOP is absent; ready to run");
|
||||
else say("no lock and no STOP; ready to run");
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const opts = parse(process.argv.slice(2));
|
||||
if (opts.command === "help") {
|
||||
@@ -251,6 +280,7 @@ async function main() {
|
||||
if (opts.command === "check") await check(opts);
|
||||
else if (opts.command === "run") await run(opts);
|
||||
else if (opts.command === "unlock") unlockCommand(opts);
|
||||
else if (opts.command === "recover") recoverCommand(opts);
|
||||
else stop(opts);
|
||||
return opts.command === "run" ? null : 0;
|
||||
}
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
// One error class for the package. exitCode follows docs/TOOLS.md: 1 operation
|
||||
// failed, 2 invalid data or configuration, 4 usage.
|
||||
// failed, 2 invalid data or configuration, 3 refused by a brake (STOP or a
|
||||
// held binding; a supervisor must not retry), 4 usage.
|
||||
export class DiscordError extends Error {
|
||||
constructor(message, exitCode = 2, details = undefined) {
|
||||
super(message);
|
||||
|
||||
@@ -6,12 +6,15 @@
|
||||
// drops.jsonl one counter line per dropped or refused inbound message
|
||||
// admissions.jsonl one line per turn admitted, before the engine is asked
|
||||
// turns/<id>.json one write-once record per turn
|
||||
// STOP presence refuses new turns
|
||||
// STOP presence refuses new turns; one JSON line per writer
|
||||
// ({at, reason}), appended, so the last line names who
|
||||
// braked; `recover` removes only a STOP it wrote itself
|
||||
// notices.jsonl once-per-day fixed lines already attempted (ceiling)
|
||||
// run.lock/ ownership directory (mkdir is atomic) holding owner.json
|
||||
// {pid, start, boot}; `stop` signals only a live pid whose
|
||||
// start time and boot id match; a stale lock refuses `run`
|
||||
// until `unlock`, which is gated by STOP
|
||||
// until `unlock`, which is gated by STOP; `recover` is the
|
||||
// supervised form for a service unit's pre-start
|
||||
// Directories are 0700, files 0600. Lines are appended, never rewritten.
|
||||
|
||||
import {
|
||||
@@ -185,6 +188,21 @@ export function clearStop(dir) {
|
||||
if (existsSync(path)) unlinkSync(path);
|
||||
}
|
||||
|
||||
// The STOP lines as written, oldest first. A line that does not parse is
|
||||
// kept as {reason: null}: it was not written by this code, so it is never
|
||||
// treated as ours.
|
||||
export function readStop(dir) {
|
||||
if (!stopRequested(dir)) return [];
|
||||
return readFileSync(stopPath(dir), "utf8").split("\n").filter((l) => l.trim() !== "").map((l) => {
|
||||
try {
|
||||
const v = JSON.parse(l);
|
||||
return v && typeof v === "object" && typeof v.reason === "string" ? v : { reason: null };
|
||||
} catch {
|
||||
return { reason: null };
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// --- run lock ---
|
||||
// One directory, <dir>/run.lock, is the ownership primitive: mkdir is atomic,
|
||||
// so two starts cannot both create it. The owner record is published inside
|
||||
@@ -308,17 +326,23 @@ export function ownerAlive(rec, opts) {
|
||||
return ownerState(rec, opts) === "live";
|
||||
}
|
||||
|
||||
// Exit code for a refusal that a supervisor must not retry: STOP is present
|
||||
// or the binding is held. Distinct from 1 (operation failed) so a service
|
||||
// unit can restart after a crash and stay down after a brake.
|
||||
export const BRAKE_EXIT = 3;
|
||||
export const RECOVER_REASON = "recover";
|
||||
|
||||
export const UNLOCK_HINT = "if no connector is running for this binding, run `scripts/discord.sh unlock <binding>`";
|
||||
|
||||
// Explains why an existing lock refuses a new claim. Always a DiscordError.
|
||||
function lockRefusal(dir, opts) {
|
||||
const existing = readPid(dir);
|
||||
const state = ownerState(existing, opts);
|
||||
if (state === "absent") return new DiscordError(`run.lock exists without an owner record: a start is in progress or was interrupted; ${UNLOCK_HINT}`, 1);
|
||||
if (state === "invalid") return new DiscordError(`run.lock has an owner record that cannot be read; refusing. Inspect ${ownerPath(dir)} by hand`, 1);
|
||||
if (state === "live") return new DiscordError(`another connector is running for this binding (pid ${existing.pid})`, 1);
|
||||
if (state === "unknown") return new DiscordError(`run.lock belongs to pid ${existing.pid}, which is alive but whose identity cannot be verified; refusing`, 1);
|
||||
return new DiscordError(`run.lock belongs to pid ${existing.pid}, which is gone or is a different process now; ${UNLOCK_HINT}`, 1);
|
||||
if (state === "absent") return new DiscordError(`run.lock exists without an owner record: a start is in progress or was interrupted; ${UNLOCK_HINT}`, BRAKE_EXIT);
|
||||
if (state === "invalid") return new DiscordError(`run.lock has an owner record that cannot be read; refusing. Inspect ${ownerPath(dir)} by hand`, BRAKE_EXIT);
|
||||
if (state === "live") return new DiscordError(`another connector is running for this binding (pid ${existing.pid})`, BRAKE_EXIT);
|
||||
if (state === "unknown") return new DiscordError(`run.lock belongs to pid ${existing.pid}, which is alive but whose identity cannot be verified; refusing`, BRAKE_EXIT);
|
||||
return new DiscordError(`run.lock belongs to pid ${existing.pid}, which is gone or is a different process now; ${UNLOCK_HINT}`, BRAKE_EXIT);
|
||||
}
|
||||
|
||||
export function writePid(dir, pid, { now = Date.now(), identity = identityOf } = {}) {
|
||||
@@ -338,7 +362,7 @@ export function writePid(dir, pid, { now = Date.now(), identity = identityOf } =
|
||||
// this claim must not stand, however it interleaved with an unlock.
|
||||
if (stopRequested(dir)) {
|
||||
clearPid(dir, pid);
|
||||
throw new DiscordError(`STOP is present in ${dir}; remove it to run`, 1);
|
||||
throw new DiscordError(`STOP is present in ${dir}; remove it to run`, BRAKE_EXIT);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -348,9 +372,10 @@ export function writePid(dir, pid, { now = Date.now(), identity = identityOf } =
|
||||
// Refuses an owner record it cannot read. Otherwise removes the lock.
|
||||
// Returns the record that was cleared (null for a lock without one), or
|
||||
// false when there was no lock. STOP stays in place;
|
||||
// remove it to run again. `beforeRemove` and `identity` are test seams.
|
||||
export function unlock(dir, { beforeRemove = null, identity = identityOf } = {}) {
|
||||
requestStop(dir, "unlock");
|
||||
// remove it to run again. `beforeRemove` and `identity` are test seams;
|
||||
// `reason` is what the STOP line says (`recover` uses its own).
|
||||
export function unlock(dir, { beforeRemove = null, identity = identityOf, reason = "unlock" } = {}) {
|
||||
requestStop(dir, reason);
|
||||
const lock = lockPath(dir);
|
||||
if (!existsSync(lock)) return false;
|
||||
const rec = readPid(dir);
|
||||
@@ -363,6 +388,41 @@ export function unlock(dir, { beforeRemove = null, identity = identityOf } = {})
|
||||
return rec;
|
||||
}
|
||||
|
||||
// Supervised pre-start, for the service unit's ExecStartPre. Lets a run
|
||||
// that crashed (or a reboot) start again without a hand, while every
|
||||
// operator brake still holds. In order:
|
||||
// STOP present refuse with BRAKE_EXIT and touch nothing. `stop`
|
||||
// and `unlock` wrote it; nothing automatic removes it.
|
||||
// The one exception is a STOP made of exactly one
|
||||
// line that an earlier `recover` wrote and then died
|
||||
// before removing: that one is ours and goes.
|
||||
// no run.lock nothing to do.
|
||||
// owner live refuse: another connector holds the binding.
|
||||
// owner unknown, or record invalid refuse, as `unlock` would.
|
||||
// owner dead, or no record `unlock` (STOP written first, so a
|
||||
// claim that publishes meanwhile releases itself),
|
||||
// then that STOP is removed again, but only if it
|
||||
// still consists of the single line `unlock` wrote
|
||||
// for us. Any other line means an operator braked in
|
||||
// the meantime; STOP stays and the start is refused.
|
||||
// Returns "clean" or "cleared". Every refusal is a DiscordError with
|
||||
// BRAKE_EXIT so a supervisor does not retry it. `beforeRemove` and
|
||||
// `identity` are test seams.
|
||||
export function recover(dir, { identity = identityOf, beforeRemove = null } = {}) {
|
||||
const ours = (lines) => lines.length === 1 && lines[0].reason === RECOVER_REASON;
|
||||
if (stopRequested(dir)) {
|
||||
if (!ours(readStop(dir))) throw new DiscordError(`STOP is present in ${dir}; the brake is on. Remove it to run`, BRAKE_EXIT);
|
||||
clearStop(dir);
|
||||
}
|
||||
if (!existsSync(lockPath(dir))) return "clean";
|
||||
const state = ownerState(readPid(dir), { identity });
|
||||
if (state === "live" || state === "unknown" || state === "invalid") throw lockRefusal(dir, { identity });
|
||||
unlock(dir, { identity, reason: RECOVER_REASON, beforeRemove });
|
||||
if (!ours(readStop(dir))) throw new DiscordError(`STOP was written by an operator while recovering ${dir}; the brake stays. Remove it to run`, BRAKE_EXIT);
|
||||
clearStop(dir);
|
||||
return "cleared";
|
||||
}
|
||||
|
||||
// The verified live owner to signal, or null. Never returns a pid whose
|
||||
// identity cannot be proven.
|
||||
export function stopTarget(dir, opts) {
|
||||
|
||||
Reference in New Issue
Block a user