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.
This commit was merged in pull request #1938.
This commit is contained in:
1 parent
f9dae1ebad
commit
900daf5213
7 files changed
+104
-22
No files matched your search
@@ -703,7 +703,8 @@ async function bus(rest, io, deps) {
|
|||||||
if (["stop", "restart", "status", "recover"].includes(sub)) {
|
if (["stop", "restart", "status", "recover"].includes(sub)) {
|
||||||
const parsed = parse(args, { flags: sub === "status" ? ["--json"] : [] });
|
const parsed = parse(args, { flags: sub === "status" ? ["--json"] : [] });
|
||||||
if (parsed.positional.length !== 1) throw usage(`bus ${sub} takes one organization`);
|
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 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);
|
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`);
|
io.stdout.write(`${JSON.stringify(result, null, parsed.flags.has("--json") ? 2 : 0)}\n`);
|
||||||
|
|||||||
@@ -2,7 +2,7 @@
|
|||||||
import { randomBytes } from 'node:crypto';
|
import { randomBytes } from 'node:crypto';
|
||||||
import { processState } from '../../seat/src/process-state.mjs';
|
import { processState } from '../../seat/src/process-state.mjs';
|
||||||
import { spawn } from 'node:child_process';
|
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 { connect } from 'node:net';
|
||||||
import { join } from 'node:path';
|
import { join } from 'node:path';
|
||||||
import { fileURLToPath } from 'node:url';
|
import { fileURLToPath } from 'node:url';
|
||||||
@@ -151,6 +151,27 @@ export async function busRestart(ctx, id) {
|
|||||||
return busStart(ctx, id, { detach: true, ...options });
|
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.
|
// Recover only while the org lifecycle fence excludes new starts and deletion.
|
||||||
export async function busRecover(ctx, id) {
|
export async function busRecover(ctx, id) {
|
||||||
return withOrgLifecycle(ctx, id, async () => {
|
return withOrgLifecycle(ctx, id, async () => {
|
||||||
@@ -177,10 +198,12 @@ export async function busRecover(ctx, id) {
|
|||||||
const h = state.value;
|
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');
|
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'); }
|
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) {
|
if (p.alive && p.startTime === h.startTime) {
|
||||||
const argv = cmdlineOf(h.pid); if (!argv || !argv.length) fail('bus host identity is uncertain');
|
const argv = cmdlineOf(h.pid), i = argv?.indexOf(CLI) ?? -1;
|
||||||
const i = argv.indexOf(CLI);
|
|
||||||
if (i >= 0 && argv[i + 1] === 'bus' && argv[i + 2] === 'start' && argv[i + 3] === id) fail(`bus host pid ${h.pid} is alive`);
|
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 (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) {
|
if (writer) {
|
||||||
@@ -198,14 +221,12 @@ export async function busRecover(ctx, id) {
|
|||||||
if (occupied) fail('bus socket answers or is uncertain');
|
if (occupied) fail('bus socket answers or is uncertain');
|
||||||
}
|
}
|
||||||
const entries = [state, lock, writer, socket].filter(Boolean);
|
const entries = [state, lock, writer, socket].filter(Boolean);
|
||||||
for (const entry of entries) {
|
const aside = setAside(entries);
|
||||||
const current = lstatSync(entry.path);
|
|
||||||
if (current.dev !== entry.st.dev || current.ino !== entry.st.ino) fail('bus recovery state changed');
|
|
||||||
}
|
|
||||||
const at = new Date().toISOString();
|
const at = new Date().toISOString();
|
||||||
const receipt = join(org.paths.hostDir, `recovery-${at.replace(/[:.]/g, '-')}-${randomBytes(4).toString('hex')}.json`);
|
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' });
|
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' }); }
|
||||||
for (const entry of entries.slice().reverse()) unlinkSync(entry.path);
|
catch (e) { aside.restore(); throw e; }
|
||||||
|
aside.remove();
|
||||||
const result = { org: id, recovered: true, pid: h.pid, receipt };
|
const result = { org: id, recovered: true, pid: h.pid, receipt };
|
||||||
// The dead host's start options, for busRevive. Not part of the printed 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 : {};
|
const o = h.options && typeof h.options === 'object' ? h.options : {};
|
||||||
|
|||||||
@@ -21,8 +21,9 @@ async function fixture() {
|
|||||||
const env = humanEnv({ HOME: root, MOSAIC_CONFIG: join(configDir, 'config.json') });
|
const env = humanEnv({ HOME: root, MOSAIC_CONFIG: join(configDir, 'config.json') });
|
||||||
const moduleUrl = new URL('../src/cli.mjs', import.meta.url).href;
|
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 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 } }; };
|
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 } }; };
|
||||||
return { root, ctx, org, call };
|
const call = (...args) => callWith({}, ...args);
|
||||||
|
return { root, ctx, org, call, callWith };
|
||||||
}
|
}
|
||||||
const dead = 2147483647;
|
const dead = 2147483647;
|
||||||
const json = (path, value) => writeFileSync(path, JSON.stringify(value), { mode: 0o600 });
|
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 () => {
|
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 });
|
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]) {
|
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);
|
json(join(dir, 'host.json'), state); json(join(dir, 'host.lock'), state);
|
||||||
const x = await f.call('bus', 'recover', 'alpha');
|
const x = await f.call('bus', 'recover', 'alpha');
|
||||||
const result = JSON.parse(x.out.text);
|
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);
|
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 () => {
|
test('lock recovery refuses malformed or unreadable holders', async () => {
|
||||||
const f = await fixture(), path = join(f.ctx.configDir, 'org-lifecycle-alpha.lock');
|
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]) {
|
for (const value of [{ pid: 0, at: '2026-01-01T00:00:00Z' }, { pid: dead }, null]) {
|
||||||
|
|||||||
@@ -4,13 +4,13 @@
|
|||||||
import { test } from 'node:test';
|
import { test } from 'node:test';
|
||||||
import assert from 'node:assert/strict';
|
import assert from 'node:assert/strict';
|
||||||
import { spawn, spawnSync } from 'node:child_process';
|
import { spawn, spawnSync } from 'node:child_process';
|
||||||
import { existsSync, mkdtempSync, readFileSync, readdirSync, rmSync, writeFileSync } from 'node:fs';
|
import { existsSync, lstatSync, mkdtempSync, readFileSync, readdirSync, renameSync, rmSync, writeFileSync } from 'node:fs';
|
||||||
import { join } from 'node:path';
|
import { join } from 'node:path';
|
||||||
import { tmpdir } from 'node:os';
|
import { tmpdir } from 'node:os';
|
||||||
import { reserveOrg, activateOrg, listOrgs } from '../../organization/src/org-registry.mjs';
|
import { reserveOrg, activateOrg, listOrgs } from '../../organization/src/org-registry.mjs';
|
||||||
import { businessDoc, rolesCopy, writeJson } from '../../organization/tests/helpers.mjs';
|
import { businessDoc, rolesCopy, writeJson } from '../../organization/tests/helpers.mjs';
|
||||||
import { humanEnv, notifyConfig } from './helpers.mjs';
|
import { humanEnv, notifyConfig } from './helpers.mjs';
|
||||||
import { busRevive, busStart, busStatus, busStop } from '../src/org-bus.mjs';
|
import { busRevive, busStart, busStatus, busStop, setAside } from '../src/org-bus.mjs';
|
||||||
import { reviveBuses } from '../../webui/src/bus-revive.mjs';
|
import { reviveBuses } from '../../webui/src/bus-revive.mjs';
|
||||||
|
|
||||||
const REPO = new URL('../../../', import.meta.url).pathname;
|
const REPO = new URL('../../../', import.meta.url).pathname;
|
||||||
@@ -97,7 +97,7 @@ test('D2: a writer lock held by a live process refuses the revive, even with the
|
|||||||
assert.equal(recoveries(roots.alpha).length, 0);
|
assert.equal(recoveries(roots.alpha).length, 0);
|
||||||
// The Console-start revive leaves it too, and logs the refusal.
|
// The Console-start revive leaves it too, and logs the refusal.
|
||||||
const lines = [], trail = [];
|
const lines = [], trail = [];
|
||||||
const results = await reviveBuses({ ctx, listOrgs, busStatus, busRevive, trail: e => trail.push(e), log: line => lines.push(line) });
|
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.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.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.deepEqual(trail.filter(e => e.org === 'alpha').map(e => e.outcome), ['requested', 'refused']);
|
||||||
@@ -110,7 +110,7 @@ test('D2: Console start brings both killed buses back and reattaches to one that
|
|||||||
const b = await busStart(ctx, 'beta', { detach: true, devLocalOwner: true, noLaunch: true });
|
const b = await busStart(ctx, 'beta', { detach: true, devLocalOwner: true, noLaunch: true });
|
||||||
await kill(a); await kill(b);
|
await kill(a); await kill(b);
|
||||||
const trail = [], lines = [];
|
const trail = [], lines = [];
|
||||||
const results = await reviveBuses({ ctx, listOrgs, busStatus, busRevive, trail: e => trail.push(e), log: line => lines.push(line) });
|
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']]);
|
assert.deepEqual(results.map(r => [r.id, r.outcome]), [['alpha', 'revived'], ['beta', 'revived']]);
|
||||||
for (const id of IDS) {
|
for (const id of IDS) {
|
||||||
const status = await busStatus(ctx, id);
|
const status = await busStatus(ctx, id);
|
||||||
@@ -120,7 +120,7 @@ test('D2: Console start brings both killed buses back and reattaches to one that
|
|||||||
}
|
}
|
||||||
// A second Console start finds them running and leaves them.
|
// 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 pids = Object.fromEntries(await Promise.all(IDS.map(async id => [id, (await busStatus(ctx, id)).pid])));
|
||||||
const again = await reviveBuses({ ctx, listOrgs, busStatus, busRevive, trail: e => trail.push(e), log: () => {} });
|
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']]);
|
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]);
|
for (const id of IDS) assert.equal((await busStatus(ctx, id)).pid, pids[id]);
|
||||||
});
|
});
|
||||||
@@ -137,3 +137,26 @@ test('D2: a start that fails after the dead state was cleared says so (error.rec
|
|||||||
assert.equal(existsSync(file), false);
|
assert.equal(existsSync(file), false);
|
||||||
assert.equal(recoveries(roots.alpha).length, 1);
|
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');
|
||||||
|
});
|
||||||
@@ -10,9 +10,12 @@
|
|||||||
// live writer, no socket that answers). Anything else refuses and the state stays for a person.
|
// 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.
|
// 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.
|
// 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
|
// 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.
|
// it instead of racing it for the lifecycle lock.
|
||||||
import { DEV_OWNER } from '../../cli/src/dev-local-owner.mjs';
|
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 isCliError = e => e?.name === 'CliError' && Number.isInteger(e.exitCode);
|
||||||
const BY = Object.freeze({ ...DEV_OWNER, name: 'Console start' });
|
const BY = Object.freeze({ ...DEV_OWNER, name: 'Console start' });
|
||||||
@@ -20,8 +23,9 @@ const pending = new Map();
|
|||||||
|
|
||||||
export const revivePending = id => pending.get(id) ?? null;
|
export const revivePending = id => pending.get(id) ?? null;
|
||||||
|
|
||||||
export async function reviveBuses({ ctx, listOrgs, busStatus, busRevive, trail, log = () => {} }) {
|
export async function reviveBuses({ ctx, listOrgs, busStatus, busRevive, trail, log = () => {}, env = process.env }) {
|
||||||
const results = [];
|
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;
|
let orgs;
|
||||||
try { orgs = await listOrgs(ctx); }
|
try { orgs = await listOrgs(ctx); }
|
||||||
catch { log('bus revive: the organization list could not be read; no bus was checked'); return results; }
|
catch { log('bus revive: the organization list could not be read; no bus was checked'); return results; }
|
||||||
|
|||||||
@@ -8,14 +8,14 @@ import { reviveStartOptions } from '../../cli/src/org-bus.mjs';
|
|||||||
const cliError = (message, extra = {}) => Object.assign(new Error(message), { name: 'CliError', exitCode: 3 }, extra);
|
const cliError = (message, extra = {}) => Object.assign(new Error(message), { name: 'CliError', exitCode: 3 }, extra);
|
||||||
const orgs = ids => async () => ids.map(id => ({ id, state: 'active' }));
|
const orgs = ids => async () => ids.map(id => ({ id, state: 'active' }));
|
||||||
|
|
||||||
function run({ ids = ['alpha'], status, revive, trail }) {
|
function run({ ids = ['alpha'], status, revive, trail, env = {} }) {
|
||||||
const calls = [], lines = [], trailed = [];
|
const calls = [], lines = [], trailed = [];
|
||||||
const done = reviveBuses({
|
const done = reviveBuses({
|
||||||
ctx: {}, listOrgs: orgs(ids),
|
ctx: {}, listOrgs: orgs(ids),
|
||||||
busStatus: async (ctx, id) => status(id),
|
busStatus: async (ctx, id) => status(id),
|
||||||
busRevive: async (ctx, id, opts) => { calls.push([id, opts]); return revive(id); },
|
busRevive: async (ctx, id, opts) => { calls.push([id, opts]); return revive(id); },
|
||||||
trail: trail ?? (e => trailed.push(e)),
|
trail: trail ?? (e => trailed.push(e)),
|
||||||
log: line => lines.push(line),
|
log: line => lines.push(line), env,
|
||||||
});
|
});
|
||||||
return { done, calls, lines, trailed };
|
return { done, calls, lines, trailed };
|
||||||
}
|
}
|
||||||
@@ -82,3 +82,14 @@ test('revivePending holds each organization until its own revive ends', async ()
|
|||||||
assert.equal(settled, true);
|
assert.equal(settled, true);
|
||||||
assert.equal(revivePending('alpha'), null); assert.equal(revivePending('beta'), null);
|
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']);
|
||||||
|
}
|
||||||
|
});
|
||||||
@@ -662,7 +662,7 @@ test('D2: a press during the Console start-up revive waits for it and acts on th
|
|||||||
bus.busStatus = async () => { if (!up) throw cliError('stale bus host lock'); return { org: 'acme', healthy: true, pid: 6161 }; };
|
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.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 }; };
|
bus.busRestart = async (ctx, id) => { c.calls.push(['busRestart', id]); return { org: id, pid: 6262 }; };
|
||||||
const startup = reviveBuses({ ctx: {}, listOrgs: async () => [{ id: 'acme', state: 'active' }], busStatus: bus.busStatus,
|
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: () => {} });
|
busRevive: async () => { await gate; up = true; return { pid: 6161 }; }, trail: () => {}, log: () => {} });
|
||||||
await new Promise(resolve => setImmediate(resolve));
|
await new Promise(resolve => setImmediate(resolve));
|
||||||
const press = c.call('POST', '/api/orgs/acme/restart/bus', {});
|
const press = c.call('POST', '/api/orgs/acme/restart/bus', {});
|
||||||
|
|||||||
Reference in new issue
Block a user