Compare commits

..
Author SHA1 Message Date
jason.woltje 900daf5213 fix(org): D2 security review: live identity refuses, clear only inspected entries, no revive from an agent run
- busRecover: a live pid with the recorded start time is the recorded process, so it refuses
  ("bus host identity is uncertain") whatever its argv. It used to read a mismatching argv as a dead
  host. The pinned recovery test now covers a reused pid (live, other start time) as the wrong
  process, and a new test covers the refusal.
- busRecover clears through setAside: each inspected entry is renamed aside and checked by device
  and inode; a swapped entry puts everything back and refuses before any receipt or deletion, and
  the deletion names only the aside entries.
- mosaic bus recover refuses inside an agent run (refuseInsideAgent), like mosaic org reload. The
  Console-start revive does nothing when the Console's own environment carries an agent marker.
- Tests: org-recovery (identity, agent refusal), org-revive (setAside swap), bus-revive (markers);
  the other reviveBuses tests pass an explicit env.
2026-11-11 14:03:18 -06:00
jason.woltje f9dae1ebad fix(org): D2 build review: pin pm:false, wait for the start-up revive, say when a clear was followed by a failed start
- The Console-start revive asks busRevive for pm:false; reviveStartOptions holds the rule (never
  wider than the dead host's record) and is tested on its own. A dropped --pm is logged with the
  way back.
- A Restart bus press waits for that organization's start-up revive (revivePending) before it reads
  the bus, so it no longer races it for the lifecycle lock; a press that still meets org-locked gets
  409 bus-busy, nothing changed.
- busRevive marks an error from busStart after a successful clear (error.recovered); the trail, the
  Console log and the press response then say "failed", not "refused". A non-CliError logs its code.
- Tests: bus-revive.test.mjs (new), org-admin and org-revive cases for the above.
2026-11-11 14:03:18 -06:00
jason.woltje f063158875 fix(org): Console start reattaches to org buses and revives a provably dead host (P0-3)
A Console unit stop (KillMode=mixed) SIGKILLs the bus hosts the Console
started, leaving host.json, host.lock and the socket behind, and the
dev Console's Restart bus refused 409 on that stale state.

- busRevive (org-bus): busRecover, then busStart detached on the
  options the dead host recorded. busRecover still refuses a live host
  pid, a lock that is not the dead host's, a live writer pid, or a
  socket that answers or is uncertain.
- reviveBuses (webui): when a dev-local-owner Console starts, a bus that
  answers is left (reattached); a bus with stale state goes to busRevive
  with pm:false, trailed as bus.revive before anything changes.
- Restart bus (org-admin): for the exact dev owner only, a stale-state
  refusal goes to busRevive; a signed-in owner keeps the refusal.
- The receipt reads "Dead bus host cleared; bus started".
2026-11-11 14:03:18 -06:00
10 changed files with 585 additions and 18 deletions

No files matched your search

+2 -1
View File
@@ -703,7 +703,8 @@ async function bus(rest, io, deps) {
if (["stop", "restart", "status", "recover"].includes(sub)) {
const parsed = parse(args, { flags: sub === "status" ? ["--json"] : [] });
if (parsed.positional.length !== 1) throw usage(`bus ${sub} takes one organization`);
if (sub === "recover") devOwnerInstance(io.env);
// D2 security review: recovery clears host state, so an agent run is refused before any read.
if (sub === "recover") { refuseInsideAgent(io.env); devOwnerInstance(io.env); }
const ctx = organizationContext(io, deps), id = parsed.positional[0];
const result = sub === "stop" ? await busStop(ctx, id) : sub === "restart" ? await busRestart(ctx, id) : sub === "recover" ? await busRecover(ctx, id) : await busStatus(ctx, id);
io.stdout.write(`${JSON.stringify(result, null, parsed.flags.has("--json") ? 2 : 0)}\n`);
+62 -10
View File
@@ -2,7 +2,7 @@
import { randomBytes } from 'node:crypto';
import { processState } from '../../seat/src/process-state.mjs';
import { spawn } from 'node:child_process';
import { constants, closeSync, lstatSync, mkdirSync, openSync, readFileSync, fstatSync, unlinkSync, writeFileSync } from 'node:fs';
import { constants, closeSync, linkSync, lstatSync, mkdirSync, openSync, readFileSync, fstatSync, renameSync, unlinkSync, writeFileSync } from 'node:fs';
import { connect } from 'node:net';
import { join } from 'node:path';
import { fileURLToPath } from 'node:url';
@@ -151,6 +151,27 @@ export async function busRestart(ctx, id) {
return busStart(ctx, id, { detach: true, ...options });
}
// D2 security review: clear only the entries busRecover inspected. Each is renamed aside (atomic)
// and the renamed entry is checked by device and inode; an entry swapped after the inspection is
// put back with everything moved so far, and the recovery refuses before any receipt or deletion.
// The deletion then names only the aside entries, never the live paths.
export function setAside(entries) {
const tag = randomBytes(4).toString('hex'), moved = [];
const restore = () => {
// link() refuses an existing path, so a put-back never replaces whatever now holds the name; an
// entry that cannot go back stays under its .recovering name for a person to inspect.
for (const m of moved.slice().reverse()) { try { linkSync(m.aside, m.path); unlinkSync(m.aside); } catch { /* left aside */ } }
};
for (const entry of entries) {
const aside = `${entry.path}.recovering-${tag}`;
let current;
try { renameSync(entry.path, aside); moved.push({ path: entry.path, aside }); current = lstatSync(aside); }
catch { restore(); fail('bus recovery state changed'); }
if (current.dev !== entry.st.dev || current.ino !== entry.st.ino) { restore(); fail('bus recovery state changed'); }
}
return { restore, remove: () => { for (const m of moved.slice().reverse()) unlinkSync(m.aside); } };
}
// Recover only while the org lifecycle fence excludes new starts and deletion.
export async function busRecover(ctx, id) {
return withOrgLifecycle(ctx, id, async () => {
@@ -177,10 +198,12 @@ export async function busRecover(ctx, id) {
const h = state.value;
if (!Number.isSafeInteger(h?.pid) || h.pid <= 0 || typeof h.startTime !== 'string' || !/^\d+$/.test(h.startTime) || typeof h.startedAt !== 'string' || !Number.isFinite(Date.parse(h.startedAt)) || h.business !== id) fail('invalid or mismatched bus host state');
let p; try { p = processState(h.pid); } catch { fail('bus host identity is uncertain'); }
// A live pid with the recorded start time is the recorded process: never proof of death. Its argv
// only picks the message (D2 security review: a mismatching argv used to read as a dead host).
if (p.alive && p.startTime === h.startTime) {
const argv = cmdlineOf(h.pid); if (!argv || !argv.length) fail('bus host identity is uncertain');
const i = argv.indexOf(CLI);
const argv = cmdlineOf(h.pid), i = argv?.indexOf(CLI) ?? -1;
if (i >= 0 && argv[i + 1] === 'bus' && argv[i + 2] === 'start' && argv[i + 3] === id) fail(`bus host pid ${h.pid} is alive`);
fail('bus host identity is uncertain');
}
if (lock && (lock.value?.pid !== h.pid || lock.value?.startTime !== h.startTime || lock.value?.business !== id || lock.value?.startedAt !== h.startedAt)) fail('bus host lock does not match host state');
if (writer) {
@@ -198,14 +221,43 @@ export async function busRecover(ctx, id) {
if (occupied) fail('bus socket answers or is uncertain');
}
const entries = [state, lock, writer, socket].filter(Boolean);
for (const entry of entries) {
const current = lstatSync(entry.path);
if (current.dev !== entry.st.dev || current.ino !== entry.st.ino) fail('bus recovery state changed');
}
const aside = setAside(entries);
const at = new Date().toISOString();
const receipt = join(org.paths.hostDir, `recovery-${at.replace(/[:.]/g, '-')}-${randomBytes(4).toString('hex')}.json`);
writeFileSync(receipt, JSON.stringify({ action: 'bus.recover', at, org: id, stopped: true, host: h, lock: lock?.value ?? null, writer: writer?.value ?? null, socket: Boolean(socket) }, null, 2) + '\n', { mode: 0o600, flag: 'wx' });
for (const entry of entries.slice().reverse()) unlinkSync(entry.path);
return { org: id, recovered: true, pid: h.pid, receipt };
try { writeFileSync(receipt, JSON.stringify({ action: 'bus.recover', at, org: id, stopped: true, host: h, lock: lock?.value ?? null, writer: writer?.value ?? null, socket: Boolean(socket) }, null, 2) + '\n', { mode: 0o600, flag: 'wx' }); }
catch (e) { aside.restore(); throw e; }
aside.remove();
const result = { org: id, recovered: true, pid: h.pid, receipt };
// The dead host's start options, for busRevive. Not part of the printed receipt.
const o = h.options && typeof h.options === 'object' ? h.options : {};
Object.defineProperty(result, 'options', { value: Object.freeze({ devLocalOwner: o.devLocalOwner === true, pm: o.pm === true, noLaunch: o.noLaunch === true }) });
return result;
});
}
// D2 (P0-3): bring back a bus whose host was killed, as a Console unit stop does to the hosts the
// Console started. busRecover is the proof: it refuses a live or unprovable host, a lock that does
// not match the dead host, a live writer and an answering socket. The host restarts on the options
// the dead host recorded, never wider; `pm: false` also keeps the chief from launching.
export function reviveStartOptions(recorded, { pm = true } = {}) {
const o = recorded && typeof recorded === 'object' ? recorded : {};
return { detach: true, devLocalOwner: o.devLocalOwner === true, pm: pm === true && o.pm === true, noLaunch: o.noLaunch === true };
}
export async function busRevive(ctx, id, { pm = true } = {}) {
await target(ctx, id);
const recovered = await busRecover(ctx, id);
const options = reviveStartOptions(recovered.options, { pm });
const cleared = { pid: recovered.pid, receipt: recovered.receipt };
let started;
try { started = await busStart(ctx, id, options); }
catch (e) {
// The dead host's state is gone by now: the caller must not report "nothing changed".
if (e && typeof e === 'object') { try { Object.defineProperty(e, 'recovered', { value: cleared }); } catch { /* frozen: the caller sees a plain failure */ } }
throw e;
}
const receipt = { ...started, recovered: cleared };
Object.defineProperty(receipt, 'done', { value: started.done });
// The chief was not relaunched although the dead host was started with --pm.
Object.defineProperty(receipt, 'pmDropped', { value: recovered.options.pm === true && !options.pm });
return receipt;
}
+25 -3
View File
@@ -21,8 +21,9 @@ async function fixture() {
const env = humanEnv({ HOME: root, MOSAIC_CONFIG: join(configDir, 'config.json') });
const moduleUrl = new URL('../src/cli.mjs', import.meta.url).href;
const code = `import { run } from ${JSON.stringify(moduleUrl)}; process.exitCode = await run(JSON.parse(process.argv[1]), undefined, {system: {dataRoot: process.argv[2]}});`;
const call = async (...args) => { const x = await exec(process.execPath, ['--input-type=module', '-e', code, JSON.stringify(args), dataRoot], { env }); return { out: { text: x.stdout }, err: { text: x.stderr } }; };
return { root, ctx, org, call };
const callWith = async (extra, ...args) => { const x = await exec(process.execPath, ['--input-type=module', '-e', code, JSON.stringify(args), dataRoot], { env: { ...env, ...extra } }); return { out: { text: x.stdout }, err: { text: x.stderr } }; };
const call = (...args) => callWith({}, ...args);
return { root, ctx, org, call, callWith };
}
const dead = 2147483647;
const json = (path, value) => writeFileSync(path, JSON.stringify(value), { mode: 0o600 });
@@ -121,8 +122,9 @@ test('org lock show and clear require a provably dead holder', async () => {
test('bus recover preserves evidence for dead and wrong-process hosts without signals', async () => {
const f = await fixture(), dir = join(f.org.root, 'bus-host'); mkdirSync(dir, { mode: 0o700 });
// A wrong process is a live pid whose start time is not the recorded one (the pid was reused).
for (const pid of [dead, process.pid]) {
const state = { pid, startTime: startTimeOf(pid) ?? '1', business: 'alpha', startedAt: '2026-01-01T00:00:00Z' };
const state = { pid, startTime: '1', business: 'alpha', startedAt: '2026-01-01T00:00:00Z' };
json(join(dir, 'host.json'), state); json(join(dir, 'host.lock'), state);
const x = await f.call('bus', 'recover', 'alpha');
const result = JSON.parse(x.out.text);
@@ -138,6 +140,26 @@ test('bus recover preserves evidence for dead and wrong-process hosts without si
assert.equal(process.pid > 0, true);
});
test('D2: bus recover refuses a live pid with the recorded start time, whatever its argv', async () => {
const f = await fixture(), dir = join(f.org.root, 'bus-host'); mkdirSync(dir, { mode: 0o700 });
const state = { pid: process.pid, startTime: startTimeOf(process.pid), business: 'alpha', startedAt: '2026-01-01T00:00:00Z' };
assert.match(state.startTime, /^\d+$/);
json(join(dir, 'host.json'), state); json(join(dir, 'host.lock'), state);
await assert.rejects(f.call('bus', 'recover', 'alpha'), /bus host identity is uncertain/);
assert.deepEqual(JSON.parse(readFileSync(join(dir, 'host.json'), 'utf8')), state);
assert.equal(existsSync(join(dir, 'host.lock')), true);
assert.ok(!readdirSync(dir).some(x => x.startsWith('recovery-')));
});
test('D2: bus recover refuses inside an agent run before it reads anything', async () => {
const f = await fixture(), dir = join(f.org.root, 'bus-host'); mkdirSync(dir, { mode: 0o700 });
const state = { pid: dead, startTime: '1', business: 'alpha', startedAt: '2026-01-01T00:00:00Z' };
json(join(dir, 'host.json'), state); json(join(dir, 'host.lock'), state);
await assert.rejects(f.callWith({ MOSAIC_AGENT_RUN: '1' }, 'bus', 'recover', 'alpha'), e => e.code === 3 && /outside any agent run/.test(e.stderr));
assert.equal(existsSync(join(dir, 'host.json')), true);
assert.ok(!readdirSync(dir).some(x => x.startsWith('recovery-')));
});
test('lock recovery refuses malformed or unreadable holders', async () => {
const f = await fixture(), path = join(f.ctx.configDir, 'org-lifecycle-alpha.lock');
for (const value of [{ pid: 0, at: '2026-01-01T00:00:00Z' }, { pid: dead }, null]) {
+162
View File
@@ -0,0 +1,162 @@
// D2 (P0-3): a Console unit stop (KillMode=mixed) SIGKILLs the bus hosts the Console started. These
// tests start real bus hosts, kill them the same way, and check that busRevive and the Console-start
// revive bring them back, and that a live host or a live writer lock is never taken over.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { spawn, spawnSync } from 'node:child_process';
import { existsSync, lstatSync, mkdtempSync, readFileSync, readdirSync, renameSync, rmSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { tmpdir } from 'node:os';
import { reserveOrg, activateOrg, listOrgs } from '../../organization/src/org-registry.mjs';
import { businessDoc, rolesCopy, writeJson } from '../../organization/tests/helpers.mjs';
import { humanEnv, notifyConfig } from './helpers.mjs';
import { busRevive, busStart, busStatus, busStop, setAside } from '../src/org-bus.mjs';
import { reviveBuses } from '../../webui/src/bus-revive.mjs';
const REPO = new URL('../../../', import.meta.url).pathname;
const IDS = ['alpha', 'beta'];
async function instance(t) {
const home = mkdtempSync(join(tmpdir(), 'rv-'));
const env = humanEnv({ HOME: home, MOSAIC_CONFIG: join(home, '.config/mosaic-rvtest/config.json') });
const setup = spawnSync(join(REPO, 'scripts/bootstrap.sh'), ['--instance', 'rvtest'], { env, encoding: 'utf8' });
assert.equal(setup.status, 0, setup.stderr);
const ctx = { configDir: join(home, '.config/mosaic-rvtest'), dataRoot: join(home, '.mosaic-rvtest'), instance: 'rvtest', env };
env.MOSAIC_ROLES_DIR = rolesCopy(home);
const roots = {};
for (const id of IDS) {
const org = await reserveOrg(ctx, id);
await activateOrg(ctx, id);
writeJson(join(ctx.configDir, 'businesses', `${id}.json`), businessDoc(home, id));
notifyConfig(org.root, id, null);
roots[id] = org.root;
}
const extra = [];
t.after(async () => {
for (const child of extra) { try { child.kill('SIGKILL'); } catch {} }
for (const id of IDS) { try { await busStop(ctx, id); } catch {} }
rmSync(home, { recursive: true, force: true });
});
return { ctx, roots, extra };
}
// What a Console unit stop does to a host it started: SIGKILL to the host and every process under it.
// The host is a session leader (detached), so its process group is the test's own; `done` resolves
// once the host is reaped, and the loop waits for the rest of the group to go.
async function kill(receipt) {
process.kill(-receipt.pid, 'SIGKILL');
await receipt.done;
for (let i = 0; i < 100; i++) {
try { process.kill(-receipt.pid, 0); } catch (e) { if (e.code === 'ESRCH') return; throw e; }
await new Promise(r => setTimeout(r, 50));
}
throw new Error(`process group ${receipt.pid} did not go`);
}
const host = root => JSON.parse(readFileSync(join(root, 'bus-host/host.json'), 'utf8'));
const recoveries = root => readdirSync(join(root, 'bus-host')).filter(name => /^recovery-.*\.json$/.test(name));
test('D2: a SIGKILLed bus host comes back from one revive, on the options it was started with', { timeout: 120000 }, async t => {
const { ctx, roots } = await instance(t);
const first = await busStart(ctx, 'alpha', { detach: true, devLocalOwner: true, noLaunch: true });
await kill(first);
// What a Console unit stop leaves behind: the dead host's state, lock and socket.
assert.equal(existsSync(join(roots.alpha, 'bus-host/host.json')), true);
await assert.rejects(busStatus(ctx, 'alpha'), e => e.name === 'CliError' && /stale/.test(e.message));
const revived = await busRevive(ctx, 'alpha');
assert.notEqual(revived.pid, first.pid);
assert.equal(revived.recovered.pid, first.pid);
assert.equal((await busStatus(ctx, 'alpha')).healthy, true);
assert.deepEqual(host(roots.alpha).options, { devLocalOwner: true, pm: false, noLaunch: true });
assert.equal(recoveries(roots.alpha).length, 1);
const receipt = JSON.parse(readFileSync(revived.recovered.receipt, 'utf8'));
assert.equal(receipt.host.pid, first.pid);
});
test('D2: a live bus host is never taken over, and its state is left as it was', { timeout: 120000 }, async t => {
const { ctx, roots } = await instance(t);
const live = await busStart(ctx, 'alpha', { detach: true, noLaunch: true });
const before = readFileSync(join(roots.alpha, 'bus-host/host.json'), 'utf8');
await assert.rejects(busRevive(ctx, 'alpha'), e => e.name === 'CliError' && e.message === `bus host pid ${live.pid} is alive`);
assert.equal(readFileSync(join(roots.alpha, 'bus-host/host.json'), 'utf8'), before);
assert.equal(recoveries(roots.alpha).length, 0);
const status = await busStatus(ctx, 'alpha');
assert.equal(status.healthy, true);
assert.equal(status.pid, live.pid);
});
test('D2: a writer lock held by a live process refuses the revive, even with the host dead', { timeout: 120000 }, async t => {
const { ctx, roots, extra } = await instance(t);
const first = await busStart(ctx, 'alpha', { detach: true, noLaunch: true });
await kill(first);
// A live process the test started holds the writer lock.
const holder = spawn('sleep', ['60'], { stdio: 'ignore' }); extra.push(holder);
writeFileSync(join(roots.alpha, 'bus/writer.lock'), JSON.stringify({ pid: holder.pid, at: new Date().toISOString() }), { mode: 0o600 });
const lockBefore = readFileSync(join(roots.alpha, 'bus/writer.lock'), 'utf8');
await assert.rejects(busRevive(ctx, 'alpha'), e => e.name === 'CliError' && e.message === `bus writer pid ${holder.pid} is alive`);
assert.equal(readFileSync(join(roots.alpha, 'bus/writer.lock'), 'utf8'), lockBefore);
assert.equal(existsSync(join(roots.alpha, 'bus-host/host.json')), true);
assert.equal(recoveries(roots.alpha).length, 0);
// The Console-start revive leaves it too, and logs the refusal.
const lines = [], trail = [];
const results = await reviveBuses({ ctx, listOrgs, busStatus, busRevive, env: {}, trail: e => trail.push(e), log: line => lines.push(line) });
assert.deepEqual(results.find(r => r.id === 'alpha'), { id: 'alpha', outcome: 'refused' });
assert.ok(lines.some(line => line.includes(`alpha: refused (bus writer pid ${holder.pid} is alive)`)), lines.join('\n'));
assert.deepEqual(trail.filter(e => e.org === 'alpha').map(e => e.outcome), ['requested', 'refused']);
assert.equal(existsSync(join(roots.alpha, 'bus-host/host.json')), true);
});
test('D2: Console start brings both killed buses back and reattaches to one that still runs', { timeout: 180000 }, async t => {
const { ctx, roots } = await instance(t);
const a = await busStart(ctx, 'alpha', { detach: true, devLocalOwner: true, noLaunch: true });
const b = await busStart(ctx, 'beta', { detach: true, devLocalOwner: true, noLaunch: true });
await kill(a); await kill(b);
const trail = [], lines = [];
const results = await reviveBuses({ ctx, listOrgs, busStatus, busRevive, env: {}, trail: e => trail.push(e), log: line => lines.push(line) });
assert.deepEqual(results.map(r => [r.id, r.outcome]), [['alpha', 'revived'], ['beta', 'revived']]);
for (const id of IDS) {
const status = await busStatus(ctx, id);
assert.equal(status.healthy, true, id);
assert.equal(host(roots[id]).options.devLocalOwner, true);
assert.deepEqual(trail.filter(e => e.org === id).map(e => [e.action, e.outcome, e.by.sub]), [['bus.revive', 'requested', 'local-dev-owner'], ['bus.revive', 'done', 'local-dev-owner']]);
}
// A second Console start finds them running and leaves them.
const pids = Object.fromEntries(await Promise.all(IDS.map(async id => [id, (await busStatus(ctx, id)).pid])));
const again = await reviveBuses({ ctx, listOrgs, busStatus, busRevive, env: {}, trail: e => trail.push(e), log: () => {} });
assert.deepEqual(again.map(r => [r.id, r.outcome]), [['alpha', 'running'], ['beta', 'running']]);
for (const id of IDS) assert.equal((await busStatus(ctx, id)).pid, pids[id]);
});
test('D2: a start that fails after the dead state was cleared says so (error.recovered)', { timeout: 120000 }, async t => {
const { ctx, roots } = await instance(t);
const first = await busStart(ctx, 'alpha', { detach: true, noLaunch: true });
await kill(first);
// A record busStart refuses (--pm with --no-launch), rewritten in place: the clear goes through and
// the start that follows fails, the way a host that cannot come up would.
const file = join(roots.alpha, 'bus-host/host.json');
writeFileSync(file, JSON.stringify({ ...host(roots.alpha), options: { devLocalOwner: false, pm: true, noLaunch: true } }));
await assert.rejects(busRevive(ctx, 'alpha'), e => e.name === 'CliError' && /--pm or --no-launch/.test(e.message) && e.recovered?.pid === first.pid);
assert.equal(existsSync(file), false);
assert.equal(recoveries(roots.alpha).length, 1);
});
test('D2: an entry swapped after inspection is put back and nothing is removed', t => {
const dir = mkdtempSync(join(tmpdir(), 'aside-'));
t.after(() => rmSync(dir, { recursive: true, force: true }));
const a = join(dir, 'host.json'), b = join(dir, 'writer.lock');
writeFileSync(a, 'dead host', { mode: 0o600 }); writeFileSync(b, 'dead writer', { mode: 0o600 });
const entries = [{ path: a, st: lstatSync(a) }, { path: b, st: lstatSync(b) }];
// The swap: the inspected writer.lock is moved off (its inode stays in use) and a new one takes the name.
renameSync(b, join(dir, 'kept')); writeFileSync(b, 'live writer', { mode: 0o600 });
assert.throws(() => setAside(entries), e => e.name === 'CliError' && e.message === 'bus recovery state changed');
assert.equal(readFileSync(a, 'utf8'), 'dead host');
assert.equal(readFileSync(b, 'utf8'), 'live writer');
assert.deepEqual(readdirSync(dir).sort(), ['host.json', 'kept', 'writer.lock']);
// Unswapped, the entries go aside and remove() deletes only those.
renameSync(join(dir, 'kept'), b);
const fresh = [{ path: a, st: lstatSync(a) }, { path: b, st: lstatSync(b) }];
const aside = setAside(fresh);
assert.equal(existsSync(a) || existsSync(b), false);
writeFileSync(a, 'a new host', { mode: 0o600 });
aside.remove();
assert.deepEqual(readdirSync(dir), ['host.json']);
assert.equal(readFileSync(a, 'utf8'), 'a new host');
});
+82
View File
@@ -0,0 +1,82 @@
// D2 (P0-3): when a dev-local-owner Console starts, it reattaches to the organization buses that
// still run and brings back the ones a Console unit stop killed. The unit's KillMode=mixed sends
// SIGKILL to the bus hosts the Console started; a host killed that way leaves its host.json behind,
// which a host stopped on purpose (mosaic bus stop, Stop in the page) removes as it closes.
//
// For each active organization:
// - a bus whose state checks out (running, or stopped cleanly) is left as it is;
// - a bus whose state does not check out goes to busRevive, whose busRecover clears the state only
// when the host is proven dead (no live pid with that start time, the lock is the dead host's, no
// live writer, no socket that answers). Anything else refuses and the state stays for a person.
// The chief is not relaunched (pm: false): no agent starts without a person's press.
// Every attempt is a trail line before anything changes; a trail that cannot be written skips it.
// A Console whose own environment carries an agent run's markers revives nothing (the human-only
// rule `mosaic bus recover` and Restart bus's `mosaic org reload` follow).
// While an organization's revive runs, revivePending(id) holds it, so a Restart bus press waits for
// it instead of racing it for the lifecycle lock.
import { DEV_OWNER } from '../../cli/src/dev-local-owner.mjs';
import { AGENT_MARKERS } from '../../cli/src/transport.mjs';
const isCliError = e => e?.name === 'CliError' && Number.isInteger(e.exitCode);
const BY = Object.freeze({ ...DEV_OWNER, name: 'Console start' });
const pending = new Map();
export const revivePending = id => pending.get(id) ?? null;
export async function reviveBuses({ ctx, listOrgs, busStatus, busRevive, trail, log = () => {}, env = process.env }) {
const results = [];
if (AGENT_MARKERS.some(k => env?.[k])) { log('bus revive: this Console runs inside an agent run; no bus was checked'); return results; }
let orgs;
try { orgs = await listOrgs(ctx); }
catch { log('bus revive: the organization list could not be read; no bus was checked'); return results; }
const active = (orgs ?? []).filter(org => org?.state === 'active').map(org => org.id);
const settle = new Map();
for (const id of active) {
let done;
const wait = new Promise(resolve => { done = resolve; });
pending.set(id, wait);
settle.set(id, () => { if (pending.get(id) === wait) pending.delete(id); done(); });
}
try {
for (const id of active) {
try { results.push(await reviveOne({ ctx, id, busStatus, busRevive, trail, log })); }
finally { settle.get(id)(); }
}
} finally { for (const release of settle.values()) release(); }
return results;
}
async function reviveOne({ ctx, id, busStatus, busRevive, trail, log }) {
let status = null;
try { status = await busStatus(ctx, id); }
catch (e) {
if (!isCliError(e)) { log(`bus revive: ${id}: bus state could not be read; left as it is`); return { id, outcome: 'uncertain' }; }
}
if (status) {
const outcome = status.healthy ? 'running' : Number.isSafeInteger(status.pid) ? 'not-answering' : 'stopped';
if (outcome === 'running') log(`bus revive: ${id}: bus answers; reattached`);
return { id, outcome };
}
try { trail({ by: BY, action: 'bus.revive', org: id, outcome: 'requested' }); }
catch { log(`bus revive: ${id}: the action trail could not be written; nothing was recovered`); return { id, outcome: 'trail-unavailable' }; }
const note = outcome => { try { trail({ by: BY, action: 'bus.revive', org: id, outcome, ...(outcome === 'failed' ? { recovered: true } : {}) }); } catch { log(`bus revive: ${id}: trail write failed after ${outcome}`); } };
try {
const receipt = await busRevive(ctx, id, { pm: false });
note('done');
log(`bus revive: ${id}: dead bus host cleared; bus started (pid ${receipt.pid})`);
if (receipt.pmDropped) log(`bus revive: ${id}: the dead host was started with --pm; the chief was not relaunched. mosaic bus stop ${id}; mosaic bus start ${id} --pm brings it back`);
return { id, outcome: 'revived', pid: receipt.pid };
} catch (e) {
// The reason names the organization id and pids only, never a host path (busRecover's and
// busStart's messages); an error that is not a CliError shows its code, not its text.
const reason = isCliError(e) ? e.message : `unexpected error${typeof e?.code === 'string' ? ` ${e.code}` : ''}`;
if (e?.recovered) {
note('failed');
log(`bus revive: ${id}: dead bus host cleared, but the bus did not start (${reason}); see mosaic bus status ${id}`);
return { id, outcome: 'failed' };
}
note('refused');
log(`bus revive: ${id}: refused (${reason}); left for mosaic bus status ${id}`);
return { id, outcome: 'refused' };
}
}
+5 -1
View File
@@ -14,7 +14,7 @@ import { startTesterSocket } from './tester-socket.mjs';
import { TESTER_CLI } from '../../cli/src/transport.mjs';
import { views } from '../../bus/src/views.mjs';
import { loadSystem, socketPath, orgPaths, REPO, rolesDir } from '../../cli/src/config.mjs';
import { busStatus as orgBusStatus } from '../../cli/src/org-bus.mjs';
import { busStatus as orgBusStatus, busRevive as orgBusRevive } from '../../cli/src/org-bus.mjs';
import { listOrgs, resolveOrg } from '../../organization/src/org-registry.mjs';
import { showOrg } from '../../organization/src/org-service.mjs';
import { configDir } from '../../organization/src/organization-config.mjs';
@@ -28,6 +28,7 @@ import { readFileSync, statSync } from 'node:fs';
import { fileURLToPath } from 'node:url';
import { isAbsolute, join } from 'node:path';
import { actionTrail } from './action-trail.mjs';
import { reviveBuses } from './bus-revive.mjs';
import { agentLevelsWriteRoutes } from './agent-levels-write.mjs';
import { RESTART_EXIT } from '../../cli/src/org-console.mjs';
import { ID_PATTERN } from '../../organization/src/vocabulary.mjs';
@@ -235,6 +236,9 @@ try {
console.log(`Listening: ${lan.bind}:${port}, allowed Host names ${lan.allowHosts.map(name => `${name}:${port}`).join(', ')}`);
}
console.log(`Mosaic Console: http://${urlHost}:${port}/\nBoard: ${options.board}\n${busLine}${queueLine}${previewLine}${ssoLines}${proofLine}${testerServer ? `\nTester Console: unix:${testerServer.address()} (binds the tester)` : ''}\nCtrl-C stops this server.`);
// D2 (P0-3): the dev Console reattaches to the buses that still run and brings back the ones its
// unit stop killed. It runs after the Console answers, so a slow bus never delays the page.
if (organizations && trail) reviveBuses({ ctx: { configDir: configDir(), dataRoot: system.dataRoot, instance }, listOrgs, busStatus: orgBusStatus, busRevive: orgBusRevive, trail, log: line => console.log(line) }).catch(() => console.log('bus revive: stopped on an unexpected error'));
for (const signal of ['SIGINT', 'SIGTERM']) process.once(signal, () => { sso?.close(); server.close(); testerServer?.close(); });
} catch (err) {
console.error(`refused: ${err.message}`);
+40 -2
View File
@@ -16,7 +16,8 @@
// POST /api/orgs/ID/setup/queue C2 queueInit mosaic org queue init
// POST /api/orgs/ID/setup/project/P C2 projectRepoInit mosaic org project repo init
// POST /api/orgs/ID/setup/bus C3 busStart mosaic bus start --detach
// POST /api/orgs/ID/restart/bus C3 busRestart mosaic bus restart (F3; busStart when stopped)
// POST /api/orgs/ID/restart/bus C3 busRestart mosaic bus restart (F3; busStart when stopped;
// busRevive for a host that died, dev-local owner only, D2)
//
// GET /api/orgs (the list) and other /api/orgs/ID/... routes are S5's; this
// handler returns false for them.
@@ -27,6 +28,8 @@ import { checkProject, initProject, projectRepo, recordContext } from '../../pro
import { refuseInsideAgent } from '../../cli/src/transport.mjs';
import { join } from 'node:path';
import { actor } from './action-trail.mjs';
import { DEV_OWNER } from '../../cli/src/dev-local-owner.mjs';
import { revivePending } from './bus-revive.mjs';
const ID = /^[a-z0-9][a-z0-9._-]{0,63}$/;
const KEY = /^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$/;
@@ -51,6 +54,7 @@ export function verbFor(step, id, opts = {}) {
if (step === 'records') return `mosaic project init ${id} ${opts.project}`;
if (step === 'bus') return `mosaic bus start ${id} --dev-local-owner --detach`;
if (step === 'bus-restart') return `mosaic bus restart ${id}`;
if (step === 'bus-revive') return `mosaic bus recover ${id}; mosaic bus start ${id} --dev-local-owner --detach`;
throw new Error(`unknown step ${step}`);
}
@@ -563,7 +567,8 @@ async function route(req, res, { path, search, org, admin, principal, write, log
// F3 (REQ-WEB-3): one control restarts an organization's bus and answers with a done receipt. A
// running host restarts on the options it was started with (C3 busRestart keeps --dev-local-owner);
// a stopped one starts as setup starts it. The trail line is written before anything changes, and
// a trail that cannot be written refuses the restart. Stale host state is never cleared here.
// a trail that cannot be written refuses the restart. Stale host state is cleared only for the
// dev-local owner, and only through busRevive (D2), whose busRecover refuses a host it cannot prove dead.
async function restartBus(req, res, { sub, write, id, ctx, service, admin, log, trail, by }) {
const step = 'bus-restart';
if (sub !== 'bus' || !write) return json(res, 404, { error: 'not found' });
@@ -571,11 +576,14 @@ async function restartBus(req, res, { sub, write, id, ctx, service, admin, log,
try { await service.showOrg(ctx, id); } catch (e) { return refusal(res, e, { step, id }); }
const bus = await need(admin, 'bus', res, step); if (!bus) return;
if (!trail) return json(res, 503, { error: 'trail-unavailable', step, message: 'this Console has no action trail, so it does not restart buses' });
// D2: the dev Console's start-up revive may be bringing this bus back; the press acts on the result.
await revivePending(id);
let status;
try { status = await bus.busStatus(ctx, id); }
catch (e) {
if (['org-unknown', 'org-mismatch', 'org-unavailable', 'org-archived'].includes(e?.code)) return refusal(res, e, { step, id });
if (isCliError(e)) {
if (isDevOwner(by) && typeof bus.busRevive === 'function') return reviveBus(res, { id, ctx, bus, log, trail, by, step });
// Refused before anything changed, but the owner asked: the trail records the attempt.
try { trail({ by, action: 'bus.restart', org: id, outcome: 'refused', running: null }); } catch { log(`org-admin: ${id} bus restart: trail write failed after refused`); }
return json(res, 409, { error: 'bus-refused', step, message: `${plainMessage(ctx, e.message)}. Inspect it with mosaic bus status ${id}; mosaic bus recover ${id} clears a host that died.` });
@@ -600,3 +608,33 @@ async function restartBus(req, res, { sub, write, id, ctx, service, admin, log,
try { healthy = (await bus.busStatus(ctx, id)).healthy; } catch { /* the receipt still stands */ }
return json(res, 200, { outcome: 'done', step, restarted: running, healthy, receipt: forReceipt(ctx, receipt), verb: running ? verbFor('bus-restart', id) : verbFor('bus', id) });
}
// The dev-local owner exactly (packages/cli/src/dev-local-owner.mjs). A signed-in session carries
// its provider's issuer, so no SSO user matches.
const isDevOwner = by => by?.provider === DEV_OWNER.provider && by.iss === DEV_OWNER.iss && by.sub === DEV_OWNER.sub;
// D2 (P0-3): the dev Console's Restart bus on a bus whose state did not check out. A Console unit
// stop kills the hosts the Console started and leaves their state behind; busRevive clears that
// state only when busRecover proves the host dead, then starts the bus on its recorded options.
// A live or unprovable host, or a lock that is not the dead host's, refuses and nothing changes.
async function reviveBus(res, { id, ctx, bus, log, trail, by, step }) {
try { trail({ by, action: 'bus.restart', org: id, outcome: 'requested', running: false, recover: true }); }
catch { return json(res, 500, { error: 'trail-unavailable', step, message: 'the action trail could not be written, so nothing was recovered' }); }
const note = outcome => { try { trail({ by, action: 'bus.restart', org: id, outcome, recover: true }); } catch { log(`org-admin: ${id} bus recover: trail write failed after ${outcome}`); } };
let receipt;
try { receipt = await bus.busRevive(ctx, id); }
catch (e) {
// busRevive marks a failure after busRecover cleared the dead host's state (e.recovered).
const cleared = Boolean(e?.recovered);
note(cleared ? 'failed' : 'refused');
if (!cleared && e?.code === 'org-locked') return json(res, 409, { error: 'bus-busy', step, message: `another bus or lifecycle action on ${id} is running, and nothing was changed. Press Restart bus again when it ends.` });
if (cleared) return json(res, 409, { error: 'bus-failed', step, message: isCliError(e) ? plainMessage(ctx, `The dead host's state was cleared, but ${e.message}`) : `The dead host's state was cleared, but the bus did not start. Check it with mosaic bus status ${id}.` });
if (!isCliError(e)) return refusal(res, e, { step, id, unknownOk: true });
return json(res, 409, { error: 'bus-refused', step, message: `${plainMessage(ctx, e.message)}. Inspect it with mosaic bus status ${id}.` });
}
note('done');
log(`org-admin: ${id} bus recovered and started`);
let healthy = null;
try { healthy = (await bus.busStatus(ctx, id)).healthy; } catch { /* the receipt still stands */ }
return json(res, 200, { outcome: 'done', step, restarted: false, recovered: true, healthy, receipt: forReceipt(ctx, receipt), verb: verbFor('bus-revive', id) });
}
+3 -1
View File
@@ -146,7 +146,9 @@
if (response.type === 'opaqueredirect') throw new Error('Sign in again, then retry.');
const body = await response.json();
if (!response.ok || body.outcome !== 'done' || body.step !== 'bus-restart') throw new Error(body.message || 'The restart was refused. Nothing was retried.');
busReceipt = `${body.restarted ? 'Bus restarted' : 'Bus started'} at ${clock()}. ${body.healthy ? 'It answers.' : `It has not answered yet. Check it with ${body.verb.replace('restart', 'status')}.`}`;
// D2: a recovered bus's verb is two commands; its check is the status verb.
const check = body.recovered ? `mosaic bus status ${id}` : body.verb.replace('restart', 'status');
busReceipt = `${body.recovered ? 'Dead bus host cleared; bus started' : body.restarted ? 'Bus restarted' : 'Bus started'} at ${clock()}. ${body.healthy ? 'It answers.' : `It has not answered yet. Check it with ${check}.`}`;
// The page's send flags were read while the bus was in its old state; read them again. This
// read goes through the page's fetch, which display.js scopes to the selected organization.
// The organization list goes first, so the view the config event redraws shows the bus as it is now.
+95
View File
@@ -0,0 +1,95 @@
// D2 (P0-3): the dev Console's start-up revive (bus-revive.mjs), on fakes. The real-host paths
// (a SIGKILLed host, a live host, a live writer lock) are in packages/cli/tests/org-revive.test.mjs.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { reviveBuses, revivePending } from '../src/bus-revive.mjs';
import { reviveStartOptions } from '../../cli/src/org-bus.mjs';
const cliError = (message, extra = {}) => Object.assign(new Error(message), { name: 'CliError', exitCode: 3 }, extra);
const orgs = ids => async () => ids.map(id => ({ id, state: 'active' }));
function run({ ids = ['alpha'], status, revive, trail, env = {} }) {
const calls = [], lines = [], trailed = [];
const done = reviveBuses({
ctx: {}, listOrgs: orgs(ids),
busStatus: async (ctx, id) => status(id),
busRevive: async (ctx, id, opts) => { calls.push([id, opts]); return revive(id); },
trail: trail ?? (e => trailed.push(e)),
log: line => lines.push(line), env,
});
return { done, calls, lines, trailed };
}
test('the start-up revive asks busRevive for pm:false, so no chief launches without a press', async () => {
const r = run({ status: () => { throw cliError('stale bus host lock'); }, revive: () => ({ pid: 77 }) });
assert.deepEqual(await r.done, [{ id: 'alpha', outcome: 'revived', pid: 77 }]);
assert.deepEqual(r.calls, [['alpha', { pm: false }]]);
});
test('reviveStartOptions keeps the recorded options, never wider, and pm only when both allow it', () => {
assert.deepEqual(reviveStartOptions({ devLocalOwner: true, pm: true, noLaunch: false }, { pm: false }), { detach: true, devLocalOwner: true, pm: false, noLaunch: false });
assert.deepEqual(reviveStartOptions({ devLocalOwner: true, pm: true, noLaunch: false }), { detach: true, devLocalOwner: true, pm: true, noLaunch: false });
assert.deepEqual(reviveStartOptions({ pm: false, noLaunch: true }, { pm: true }), { detach: true, devLocalOwner: false, pm: false, noLaunch: true });
// Anything but a literal true is false: a malformed record widens nothing.
assert.deepEqual(reviveStartOptions({ devLocalOwner: 'yes', pm: 1 }), { detach: true, devLocalOwner: false, pm: false, noLaunch: false });
assert.deepEqual(reviveStartOptions(null), { detach: true, devLocalOwner: false, pm: false, noLaunch: false });
});
test('a dropped --pm is logged with the way back', async () => {
const r = run({ status: () => { throw cliError('stale bus host lock'); }, revive: () => Object.defineProperty({ pid: 9 }, 'pmDropped', { value: true }) });
await r.done;
assert.ok(r.lines.some(l => l === 'bus revive: alpha: the dead host was started with --pm; the chief was not relaunched. mosaic bus stop alpha; mosaic bus start alpha --pm brings it back'), r.lines.join('\n'));
});
test('running, not-answering, stopped and unreadable buses are left alone and never trailed', async () => {
const states = { a: { healthy: true, pid: 1 }, b: { healthy: false, pid: 2 }, c: { healthy: false, pid: null } };
const r = run({ ids: ['a', 'b', 'c', 'd'], status: id => { if (id === 'd') throw new Error('EACCES /secret/path'); return states[id]; }, revive: () => assert.fail('revived') });
assert.deepEqual(await r.done, [{ id: 'a', outcome: 'running' }, { id: 'b', outcome: 'not-answering' }, { id: 'c', outcome: 'stopped' }, { id: 'd', outcome: 'uncertain' }]);
assert.deepEqual(r.trailed, []);
assert.ok(!r.lines.join('\n').includes('/secret/path'));
});
test('a trail that cannot be written skips the revive', async () => {
const r = run({ status: () => { throw cliError('stale bus host lock'); }, revive: () => assert.fail('revived without a trail line'), trail: () => { throw new Error('disk full'); } });
assert.deepEqual(await r.done, [{ id: 'alpha', outcome: 'trail-unavailable' }]);
assert.deepEqual(r.calls, []);
});
test('a start that fails after the dead state was cleared is "failed", not "refused"', async () => {
const r = run({ status: () => { throw cliError('stale bus host lock'); }, revive: () => { throw cliError('bus failed to start; see bus-host/host.log in the runtime directory of organization alpha', { recovered: { pid: 4 } }); } });
assert.deepEqual(await r.done, [{ id: 'alpha', outcome: 'failed' }]);
assert.deepEqual(r.trailed.map(e => [e.outcome, e.recovered]), [['requested', undefined], ['failed', true]]);
assert.match(r.lines.at(-1), /^bus revive: alpha: dead bus host cleared, but the bus did not start \(bus failed to start;/);
// A non-CliError shows its code, never its text.
const q = run({ status: () => { throw cliError('stale bus host lock'); }, revive: () => { throw Object.assign(new Error('EIO /home/someone/x'), { code: 'org-locked' }); } });
assert.deepEqual(await q.done, [{ id: 'alpha', outcome: 'refused' }]);
assert.equal(q.lines.at(-1), 'bus revive: alpha: refused (unexpected error org-locked); left for mosaic bus status alpha');
});
test('revivePending holds each organization until its own revive ends', async () => {
let release;
const gate = new Promise(resolve => { release = resolve; });
const r = run({ ids: ['alpha', 'beta'], status: id => { if (id === 'alpha') throw cliError('stale bus host lock'); return { healthy: true, pid: 3 }; }, revive: async () => { await gate; return { pid: 5 }; } });
await new Promise(resolve => setImmediate(resolve));
const alpha = revivePending('alpha'), beta = revivePending('beta');
assert.ok(alpha instanceof Promise); assert.ok(beta instanceof Promise);
assert.equal(revivePending('gamma'), null);
let settled = false; alpha.then(() => { settled = true; });
await new Promise(resolve => setImmediate(resolve));
assert.equal(settled, false);
release();
await r.done;
assert.equal(settled, true);
assert.equal(revivePending('alpha'), null); assert.equal(revivePending('beta'), null);
});
test('a Console that runs inside an agent run revives nothing and reads no organization', async () => {
for (const marker of ['MOSAIC_AGENT_RUN', 'CLAUDECODE', 'CODEX_THREAD_ID']) {
let listed = false;
const lines = [];
const results = await reviveBuses({ ctx: {}, listOrgs: async () => { listed = true; return []; }, busStatus: () => assert.fail('read'), busRevive: () => assert.fail('revived'), trail: () => assert.fail('trailed'), log: line => lines.push(line), env: { [marker]: '1' } });
assert.deepEqual(results, []);
assert.equal(listed, false);
assert.deepEqual(lines, ['bus revive: this Console runs inside an agent run; no bus was checked']);
}
});
+109
View File
@@ -565,3 +565,112 @@ test('a bus that refuses to restart is a visible failure and the trail says refu
assert.deepEqual(tr.lines().slice(2), [{ by: DEV, action: 'bus.restart', org: 'acme', outcome: 'refused', running: null }]);
assert.equal(c.calls.slice(before).filter(y => ['busStart', 'busRestart'].includes(y[0])).length, 0);
});
// D2 (P0-3): on the dev Console, Restart bus on a host that died (its state left behind by a
// Console unit stop) clears that state through busRevive and starts the bus. busRevive's
// busRecover is the proof; a live or unprovable host refuses and nothing starts.
const cliError = message => Object.assign(new Error(message), { name: 'CliError', exitCode: 3 });
const withRevive = (c, revive) => c.admin.bus().then(bus => {
bus.busStatus = async () => { throw cliError('stale bus host lock'); };
bus.busRevive = async (ctx, id, opts) => { c.calls.push(['busRevive', id, opts]); return revive(bus, id); };
return bus;
});
test('D2: the dev owner\'s Restart bus brings back a host that died, and the trail records the recovery', async t => {
const tr = trailIn(t);
const c = await console_(t, { trail: tr.trail });
await c.call('POST', '/api/orgs', { id: 'acme', doc: browserDoc('acme') });
await withRevive(c, (bus, id) => {
bus.busStatus = async () => ({ org: id, healthy: true, pid: 5151 });
return { org: id, pid: 5151, socket: '/data/org-runtime/acme/bus/broker.sock', root: '/data/org-runtime/acme', recovered: { pid: 4242, receipt: '/data/org-runtime/acme/bus-host/recovery-1.json' } };
});
const r = await c.call('POST', '/api/orgs/acme/restart/bus', {});
assert.equal(r.status, 200, r.raw);
assert.deepEqual(r.body, { outcome: 'done', step: 'bus-restart', restarted: false, recovered: true, healthy: true,
receipt: { org: 'acme', pid: 5151, socket: '<data root>/org-runtime/acme/bus/broker.sock', root: '<data root>/org-runtime/acme', recovered: { pid: 4242, receipt: '<data root>/org-runtime/acme/bus-host/recovery-1.json' } },
verb: 'mosaic bus recover acme; mosaic bus start acme --dev-local-owner --detach' });
assert.deepEqual(c.calls.filter(x => ['busRevive', 'busStart', 'busRestart'].includes(x[0])), [['busRevive', 'acme', undefined]]);
assert.deepEqual(tr.lines(), [
{ by: DEV, action: 'bus.restart', org: 'acme', outcome: 'requested', running: false, recover: true },
{ by: DEV, action: 'bus.restart', org: 'acme', outcome: 'done', recover: true },
]);
});
test('D2: a host that is alive or unprovable still refuses on the dev Console, and nothing starts', async t => {
const tr = trailIn(t);
const c = await console_(t, { trail: tr.trail });
await c.call('POST', '/api/orgs', { id: 'acme', doc: browserDoc('acme') });
for (const reason of ['bus host pid 4242 is alive', 'bus host lock does not match host state', 'bus socket answers or is uncertain', 'bus host identity is uncertain']) {
await withRevive(c, () => { throw cliError(reason); });
const r = await c.call('POST', '/api/orgs/acme/restart/bus', {});
assert.equal(r.status, 409, reason); assert.equal(r.body.error, 'bus-refused');
assert.equal(r.body.message, `${reason}. Inspect it with mosaic bus status acme.`);
}
assert.equal(c.calls.filter(x => ['busStart', 'busRestart'].includes(x[0])).length, 0);
assert.deepEqual(tr.lines().map(e => e.outcome), ['requested', 'refused', 'requested', 'refused', 'requested', 'refused', 'requested', 'refused']);
// A host that was cleared but did not come back says so, without host paths.
await withRevive(c, () => { throw Object.assign(cliError('bus failed to start; see bus-host/host.log in the runtime directory of organization acme'), { recovered: { pid: 4242 } }); });
const f = await c.call('POST', '/api/orgs/acme/restart/bus', {});
assert.equal(f.status, 409); assert.equal(f.body.error, 'bus-failed');
assert.equal(f.body.message, "The dead host's state was cleared, but bus failed to start; see bus-host/host.log in the runtime directory of organization acme");
assert.deepEqual(tr.lines().slice(-1), [{ by: DEV, action: 'bus.restart', org: 'acme', outcome: 'failed', recover: true }]);
// Another lifecycle action holds the organization's lock: nothing changed, and the page says to press again.
await withRevive(c, () => { throw Object.assign(new Error('organization lifecycle lock is held; inspect the lock and resume'), { code: 'org-locked' }); });
const l = await c.call('POST', '/api/orgs/acme/restart/bus', {});
assert.equal(l.status, 409); assert.equal(l.body.error, 'bus-busy');
assert.equal(l.body.message, 'another bus or lifecycle action on acme is running, and nothing was changed. Press Restart bus again when it ends.');
assert.deepEqual(tr.lines().slice(-1).map(e => e.outcome), ['refused']);
// No trail, no recovery.
const broken = await console_(t, { trail: () => { throw new Error('disk full'); } });
await broken.call('POST', '/api/orgs', { id: 'acme', doc: browserDoc('acme') });
await withRevive(broken, () => assert.fail('busRevive ran without a trail line'));
const b = await broken.call('POST', '/api/orgs/acme/restart/bus', {});
assert.equal(b.status, 500); assert.equal(b.body.error, 'trail-unavailable');
});
test('D2: a signed-in (non-dev) owner keeps the stale-state refusal and never reaches busRevive', async t => {
const { Readable } = await import('node:stream');
const tr = trailIn(t);
const f = fakes();
await f.service.createOrg({}, 'acme', template('acme'));
const bus = await f.admin.bus();
bus.busStatus = async () => { throw cliError('stale bus host lock'); };
bus.busRevive = async () => assert.fail('busRevive reached from a signed-in session');
const call = async session => {
const req = Object.assign(Readable.from([Buffer.from('{}')]), { method: 'POST', headersDistinct: { origin: ['http://console.example'], 'sec-fetch-site': ['same-origin'], 'content-type': ['application/json'] } });
const res = { writeHead(status) { this.status = status; }, end(body) { this.body = JSON.parse(body); } };
await orgAdminHTTP(req, res, { path: '/api/orgs/acme/restart/bus', admin: f.admin, session, owner: true, send: true, origin: 'http://console.example', trail: tr.trail });
return res;
};
// An SSO owner, and a session that names the local-dev provider under another issuer.
for (const session of [{ provider: 'fixture', iss: 'https://issuer.example', sub: 'owner', name: 'jason', rootOwner: true }, { provider: 'local-dev', iss: 'https://issuer.example', sub: 'local-dev-owner', rootOwner: true }]) {
const r = await call(session);
assert.equal(r.status, 409); assert.equal(r.body.error, 'bus-refused');
assert.equal(r.body.message, 'stale bus host lock. Inspect it with mosaic bus status acme; mosaic bus recover acme clears a host that died.');
}
assert.deepEqual(tr.lines().map(e => [e.outcome, e.running, e.recover]), [['refused', null, undefined], ['refused', null, undefined]]);
});
test('D2: a press during the Console start-up revive waits for it and acts on the bus it brought back', async t => {
const { reviveBuses } = await import('../src/bus-revive.mjs');
const tr = trailIn(t);
const c = await console_(t, { trail: tr.trail });
await c.call('POST', '/api/orgs', { id: 'acme', doc: browserDoc('acme') });
const bus = await c.admin.bus();
let up = false, release;
const gate = new Promise(resolve => { release = resolve; });
bus.busStatus = async () => { if (!up) throw cliError('stale bus host lock'); return { org: 'acme', healthy: true, pid: 6161 }; };
bus.busRevive = async () => assert.fail('the press revived a bus the Console was already bringing back');
bus.busRestart = async (ctx, id) => { c.calls.push(['busRestart', id]); return { org: id, pid: 6262 }; };
const startup = reviveBuses({ ctx: {}, env: {}, listOrgs: async () => [{ id: 'acme', state: 'active' }], busStatus: bus.busStatus,
busRevive: async () => { await gate; up = true; return { pid: 6161 }; }, trail: () => {}, log: () => {} });
await new Promise(resolve => setImmediate(resolve));
const press = c.call('POST', '/api/orgs/acme/restart/bus', {});
await new Promise(resolve => setTimeout(resolve, 50));
assert.equal(c.calls.filter(x => x[0] === 'busRestart').length, 0);
release();
await startup;
const r = await press;
assert.equal(r.status, 200, r.raw); assert.equal(r.body.restarted, true);
assert.deepEqual(c.calls.filter(x => x[0] === 'busRestart'), [['busRestart', 'acme']]);
});