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