stack/packages/webui/tests/bus-stop.test.mjs

// Stop bus (#1873, bus-stop): the Console's POST /api/orgs/ID/stop/bus stops one organization's bus
// host the way `mosaic bus stop` does, with SIGTERM and a clean close, so no host.lock or writer.lock
// is left and an offline schema migration can open the store straight after. A real scratch instance
// (bootstrap.sh), a real bus host and a real dev-owner Console on port 0.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { spawn, spawnSync } from 'node:child_process';
import { createHash } from 'node:crypto';
import { existsSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync, writeFileSync } from 'node:fs';
import { request } from 'node:http';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { humanEnv } from '../../cli/tests/helpers.mjs';
import { busStatus, busStop } from '../../cli/src/org-bus.mjs';
import { consoleStart, consoleStop } from '../../cli/src/org-console.mjs';
import { createOrg, templateRender } from '../../organization/src/org-service.mjs';

const REPO = new URL('../../../', import.meta.url).pathname;
const MOSAIC = join(REPO, 'scripts/mosaic');
const MIGRATE = join(REPO, 'packages/bus/src/migrate-schema.mjs');
const DEV = { provider: 'local-dev', iss: 'local-dev', sub: 'local-dev-owner' };
const FORBIDDEN = [7332, 7340, 7341, 33223, 17450, 17443, 44307, 7350];

function mosaic(env, args, timeout = 60000) {
  return new Promise((resolve, reject) => {
    const child = spawn(MOSAIC, args, { env, stdio: ['ignore', 'pipe', 'pipe'] });
    let stdout = '', stderr = '';
    child.stdout.setEncoding('utf8').on('data', c => { stdout += c; });
    child.stderr.setEncoding('utf8').on('data', c => { stderr += c; });
    const timer = setTimeout(() => child.kill('SIGKILL'), timeout);
    child.on('error', reject);
    child.on('close', status => { clearTimeout(timer); resolve({ status, stdout, stderr }); });
  });
}
// A request the way the page sends it to its own Console.
function post(port, path) {
  return new Promise((resolve, reject) => {
    const host = `127.0.0.1:${port}`;
    const req = request({ host: '127.0.0.1', port, path, method: 'POST', headers: { host, origin: `http://${host}`, 'sec-fetch-site': 'same-origin', 'content-type': 'application/json', 'content-length': 2 } }, res => {
      let text = ''; res.setEncoding('utf8'); res.on('data', c => { text += c; });
      res.on('end', () => { let body = null; try { body = JSON.parse(text); } catch { /* not JSON */ } resolve({ status: res.statusCode, body, text }); });
    });
    req.on('error', reject); req.end('{}');
  });
}
// Every file under the bus state directories, with its content hash.
function snapshot(root) {
  const out = {};
  for (const dir of ['bus-host', 'bus']) {
    const base = join(root, dir);
    if (!existsSync(base)) continue;
    for (const name of readdirSync(base).sort()) {
      const p = join(base, name), st = statSync(p);
      out[`${dir}/${name}`] = st.isFile() ? createHash('sha256').update(readFileSync(p)).digest('hex') : st.isDirectory() ? 'dir' : 'other';
    }
  }
  return out;
}

test('Stop bus stops a live bus cleanly: no locks, an offline migration runs, a second press answers stopped, setup starts it again', { timeout: 180000 }, async t => {
  const home = mkdtempSync(join(tmpdir(), 'bs-'));
  const env = humanEnv({ HOME: home, MOSAIC_CONFIG: join(home, '.config/mosaic-bstop/config.json') });
  const setup = spawnSync(join(REPO, 'scripts/bootstrap.sh'), ['--instance', 'bstop'], { env, encoding: 'utf8' });
  assert.equal(setup.status, 0, setup.stderr);
  const ctx = { configDir: join(home, '.config/mosaic-bstop'), dataRoot: join(home, '.mosaic-bstop'), instance: 'bstop', env };
  t.after(async () => {
    try { await busStop(ctx, 'alpha'); } catch {}
    try { await consoleStop(ctx, { warn: () => {} }); } catch {}
    rmSync(home, { recursive: true, force: true });
  });
  await createOrg({ ...ctx, principal: DEV }, 'alpha', await templateRender('alpha', { fixture: 'local' }));
  const started = await mosaic(env, ['bus', 'start', 'alpha', '--no-launch', '--dev-local-owner', '--detach']);
  assert.equal(started.status, 0, started.stderr);
  const first = await busStatus(ctx, 'alpha');
  assert.equal(first.healthy, true);
  const root = first.root;
  const hostLock = join(root, 'bus-host', 'host.lock'), writerLock = join(root, 'bus', 'writer.lock');
  // A live bus holds both locks, so their absence afterwards is the stop's doing.
  assert.ok(existsSync(hostLock), 'a live bus holds host.lock'); assert.ok(existsSync(writerLock), 'a live bus holds writer.lock');
  const run = await consoleStart(ctx, { devLocalOwner: true, detach: true, port: 'auto', warn: () => {} });
  const port = Number(new URL(run.url).port);
  assert.ok(port > 0 && !FORBIDDEN.includes(port), String(port));

  const stop = await post(port, '/api/orgs/alpha/stop/bus');
  assert.equal(stop.status, 200, stop.text);
  assert.equal(stop.body.step, 'bus-stop'); assert.equal(stop.body.state, 'stopped'); assert.equal(stop.body.stopped, true);
  assert.equal(stop.body.receipt.pid, first.pid);
  assert.ok(!stop.text.includes(home), 'the receipt names no host path');
  assert.ok(!existsSync(`/proc/${first.pid}`) || !readFileSync(`/proc/${first.pid}/cmdline`, 'utf8').includes('--internal-host'), 'the host process is gone');
  assert.ok(!existsSync(hostLock), 'no host.lock left'); assert.ok(!existsSync(writerLock), 'no writer.lock left');
  const status = await busStatus(ctx, 'alpha');
  assert.equal(status.healthy, false); assert.equal(status.pid ?? null, null);

  // The offline schema migration opens the store straight after (it refuses writer-locked otherwise).
  const migrate = spawnSync(process.execPath, [MIGRATE, root], { env: { PATH: process.env.PATH }, encoding: 'utf8' });
  assert.equal(migrate.status, 0, migrate.stderr);
  assert.match(migrate.stdout, /bus schema \S+ ready/);
  assert.ok(!existsSync(writerLock), 'the migration released the store');

  // A second press finds it stopped and changes nothing.
  const quiet = snapshot(root);
  const again = await post(port, '/api/orgs/alpha/stop/bus');
  assert.equal(again.status, 200, again.text);
  assert.deepEqual([again.body.step, again.body.state, again.body.stopped], ['bus-stop', 'stopped', false]);
  assert.deepEqual(snapshot(root), quiet);

  // A host lock this Console cannot prove (a live pid that is not the recorded process) refuses,
  // and every bus file stays as it was.
  writeFileSync(hostLock, JSON.stringify({ pid: process.pid, startTime: '1', business: 'alpha', startedAt: new Date().toISOString() }));
  const unproven = snapshot(root);
  const refused = await post(port, '/api/orgs/alpha/stop/bus');
  assert.equal(refused.status, 409, refused.text); assert.equal(refused.body.error, 'bus-refused');
  assert.match(refused.body.message, /^stale bus host lock\. Inspect it with mosaic bus status alpha/);
  assert.deepEqual(snapshot(root), unproven);
  rmSync(hostLock);

  // The setup step starts it again.
  const up = await post(port, '/api/orgs/alpha/setup/bus');
  assert.equal(up.status, 201, up.text); assert.equal(up.body.step, 'bus');
  const back = await busStatus(ctx, 'alpha');
  assert.equal(back.healthy, true); assert.notEqual(back.pid, first.pid);
  assert.ok(existsSync(hostLock)); assert.ok(existsSync(writerLock));
  // And stops again from the page.
  const last = await post(port, '/api/orgs/alpha/stop/bus');
  assert.equal(last.status, 200, last.text); assert.equal(last.body.stopped, true);
  assert.ok(!existsSync(hostLock)); assert.ok(!existsSync(writerLock));
});