stack/packages/cli/tests/org-revive.test.mjs

// 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');
});