rocko
4bdc7e0e86
Unify halt and fire cancellation by verified run identity with durable first-owner receipts and shared deadlines. Integrate the instance control with the current host, library and tester-socket paths. Focused nine-file evidence is 62 passing tests; broad suites, mutants and review remain pending.
184 lines
12 KiB
JavaScript
184 lines
12 KiB
JavaScript
import test from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import { spawn, spawnSync } from 'node:child_process';
|
|
import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
|
|
import { tmpdir } from 'node:os';
|
|
import { join } from 'node:path';
|
|
import { once } from 'node:events';
|
|
import { initializeControl, transition } from '../src/state.mjs';
|
|
const api = await import('../src/runtimes.mjs').catch(e => { if (e.code === 'ERR_MODULE_NOT_FOUND') return {}; throw e; });
|
|
const actor = { business: 'demo', human: 'jason', halted: true };
|
|
const delay = ms => new Promise(resolve => setTimeout(resolve, ms));
|
|
function fixture(t) {
|
|
const root = mkdtempSync(join(tmpdir(), 'processing-runtime-'));
|
|
initializeControl(root); t.after(() => rmSync(root, { recursive: true, force: true }));
|
|
return root;
|
|
}
|
|
async function child(t, stubborn = false) {
|
|
const p = spawn(process.execPath, ['-e', `process.on('SIGTERM',()=>{${stubborn ? '' : 'process.exit(0)'}});console.log('ready');setInterval(()=>{},1000)`], { detached: true, stdio: ['ignore', 'pipe', 'pipe'] });
|
|
const ended = once(p, 'exit');
|
|
t.after(async () => { if (p.exitCode === null && p.signalCode === null) { try { process.kill(-p.pid, 'SIGKILL'); } catch {} await ended; } });
|
|
await once(p.stdout, 'data');
|
|
return { p, ended };
|
|
}
|
|
function receipt(root, p) {
|
|
const r = api.registerRuntime(root, { kind: 'session', business: 'demo', run: 'run' });
|
|
return api.updateRuntime(root, r.id, { ...api.processIdentity(p.pid), phase: 'running', group: true });
|
|
}
|
|
|
|
// Without TERM, a cooperative child would survive until unnecessary escalation.
|
|
test('halt terminates a tracked stand-in and leaves honest write-once stop evidence', async t => {
|
|
assert.equal(typeof api.registerRuntime, 'function');
|
|
const root = fixture(t), { p, ended } = await child(t);
|
|
const r = receipt(root, p);
|
|
const evidence = join(root, 'evidence.json'); writeFileSync(evidence, 'immutable existing result');
|
|
const state = transition(root, actor);
|
|
const result = await api.sweepRuntimes(root, state.generation, { graceMs: 150, killWaitMs: 1000 });
|
|
assert.equal(result.quiescent, true);
|
|
assert.equal(result.remaining, 0);
|
|
assert.deepEqual(await ended, [0, null]);
|
|
const stops = api.readStops(root, state.generation);
|
|
assert.equal(stops.length, 1);
|
|
assert.equal(stops[0].runtime, r.id);
|
|
assert.equal(stops[0].outcome, 'terminated');
|
|
assert.deepEqual(stops[0].signals, ['SIGTERM']);
|
|
assert.equal(readFileSync(evidence, 'utf8'), 'immutable existing result');
|
|
const bytes = JSON.stringify(stops);
|
|
await api.sweepRuntimes(root, state.generation, { graceMs: 10, killWaitMs: 1000 });
|
|
assert.equal(JSON.stringify(api.readStops(root, state.generation)), bytes);
|
|
});
|
|
|
|
// Dropping the KILL escalation leaves this actual child alive and fails quiescence.
|
|
test('a TERM-resistant workload is killed after the bounded wait', async t => {
|
|
assert.equal(typeof api.registerRuntime, 'function');
|
|
const root = fixture(t), { p, ended } = await child(t, true);
|
|
receipt(root, p);
|
|
const state = transition(root, actor), begin = Date.now();
|
|
const result = await api.sweepRuntimes(root, state.generation, { graceMs: 80, killWaitMs: 1000 });
|
|
assert.equal(result.quiescent, true);
|
|
assert.deepEqual(await ended, [null, 'SIGKILL']);
|
|
assert.ok(Date.now() - begin >= 80);
|
|
assert.deepEqual(api.readStops(root, state.generation)[0].signals, ['SIGTERM', 'SIGKILL']);
|
|
});
|
|
|
|
// An unchecked stale start identity would terminate a different process.
|
|
test('wrong start ticks or boot identity cannot signal a live replacement', async t => {
|
|
assert.equal(typeof api.registerRuntime, 'function');
|
|
const root = fixture(t), { p } = await child(t);
|
|
const r = receipt(root, p);
|
|
api.updateRuntime(root, r.id, { startTime: '0' });
|
|
const state = transition(root, actor);
|
|
await api.sweepRuntimes(root, state.generation, { graceMs: 30, killWaitMs: 30 });
|
|
await delay(30);
|
|
assert.equal(p.exitCode, null); assert.equal(p.signalCode, null);
|
|
assert.equal(api.readStops(root, state.generation)[0].outcome, 'unknown');
|
|
});
|
|
|
|
test('a pending start cannot be reported quiescent and a second root is never signalled', async t => {
|
|
assert.equal(typeof api.registerRuntime, 'function');
|
|
const a = fixture(t), b = fixture(t), { p } = await child(t);
|
|
api.registerRuntime(a, { kind: 'session', business: 'demo', run: 'pending' });
|
|
receipt(b, p);
|
|
const state = transition(a, actor);
|
|
const result = await api.sweepRuntimes(a, state.generation, { graceMs: 30, killWaitMs: 30 });
|
|
assert.equal(result.quiescent, false);
|
|
assert.equal(result.remaining, 1);
|
|
assert.equal(p.exitCode, null); assert.equal(p.signalCode, null);
|
|
});
|
|
|
|
// A group leader can exit on TERM while its child ignores it. Quiescence is the whole group.
|
|
test('escalation still kills the tracked group after its leader exits', async t => {
|
|
const root = fixture(t);
|
|
const script = `const {spawn}=require('node:child_process');
|
|
const c=spawn(process.execPath,['-e',"process.on('SIGTERM',()=>{});console.log('ready');setInterval(()=>{},1000)"],{stdio:['ignore','pipe','inherit']});
|
|
c.stdout.once('data',()=>console.log(c.pid));process.on('SIGTERM',()=>process.exit(0));setInterval(()=>{},1000);`;
|
|
const p = spawn(process.execPath, ['-e', script], { detached: true, stdio: ['ignore', 'pipe', 'pipe'] });
|
|
t.after(() => { try { process.kill(-p.pid, 'SIGKILL'); } catch {} });
|
|
const [line] = await once(p.stdout, 'data');
|
|
const descendant = Number(String(line).trim());
|
|
const r = receipt(root, p), state = transition(root, actor);
|
|
const outcome = await api.sweepRuntimes(root, state.generation, { graceMs: 80, killWaitMs: 1000 });
|
|
assert.equal(outcome.quiescent, true);
|
|
assert.equal(api.processIdentity(descendant), null, 'stubborn descendant must not survive the group leader');
|
|
assert.equal(api.readStops(root, state.generation).find(s => s.runtime === r.id).outcome, 'killed');
|
|
});
|
|
|
|
// Session TERM goes to the runner, leaving the outer namespace for forced escalation.
|
|
test('session TERM targets its verified runner before forcing the outer namespace', async t => {
|
|
const root = fixture(t);
|
|
const script = `const {spawn}=require('node:child_process');
|
|
process.on('SIGTERM',()=>require('node:fs').writeFileSync(${JSON.stringify(join(root, 'outer-term'))},'unexpected'));
|
|
const c=spawn(process.execPath,['-e',"process.on('SIGTERM',()=>process.exit(0));console.log('ready');setInterval(()=>{},1000)"],{stdio:['ignore','pipe','inherit']});
|
|
c.stdout.once('data',()=>console.log(c.pid));c.once('exit',()=>process.exit(0));setInterval(()=>{},1000);`;
|
|
const p = spawn(process.execPath, ['-e', script], { detached: true, stdio: ['ignore', 'pipe', 'pipe'] });
|
|
t.after(() => { try { process.kill(-p.pid, 'SIGKILL'); } catch {} });
|
|
const [line] = await once(p.stdout, 'data'), runner = api.processIdentity(Number(String(line).trim()));
|
|
const r = receipt(root, p);
|
|
api.updateRuntime(root, r.id, { runnerPid: runner.pid, runnerStartTime: runner.startTime });
|
|
const state = transition(root, actor);
|
|
const result = await api.sweepRuntimes(root, state.generation, { graceMs: 300, killWaitMs: 1000 });
|
|
assert.equal(result.quiescent, true);
|
|
assert.deepEqual(api.readStops(root, state.generation)[0].signals, ['SIGTERM']);
|
|
assert.equal(api.runtime(root, r.id).haltGeneration, state.generation);
|
|
assert.equal(existsSync(join(root, 'outer-term')), false);
|
|
});
|
|
|
|
// Publishing a receipt is not durable if its newly-created containing directory can vanish.
|
|
test('new runtime and stop directories sync their containing directories', t => {
|
|
const root = fixture(t);
|
|
const script = `import fs from 'node:fs';import {syncBuiltinESMExports} from 'node:module';
|
|
const real=fs.fsyncSync,seen=[];fs.fsyncSync=fd=>{seen.push(fs.readlinkSync('/proc/self/fd/'+fd));return real(fd)};
|
|
syncBuiltinESMExports();const api=await import(process.argv[1]);
|
|
const {transition}=await import(process.argv[2]);
|
|
const r=api.registerRuntime(process.argv[3],{kind:'task',business:'demo',run:'gone'});
|
|
api.updateRuntime(process.argv[3],r.id,{phase:'ended'});
|
|
const s=transition(process.argv[3],{business:'demo',human:'jason',halted:true});
|
|
await api.sweepRuntimes(process.argv[3],s.generation);
|
|
console.log(JSON.stringify({seen,generation:s.generation}));`;
|
|
const child = spawnSync(process.execPath, ['--input-type=module', '-e', script,
|
|
new URL('../src/runtimes.mjs', import.meta.url).href, new URL('../src/state.mjs', import.meta.url).href, root], { encoding: 'utf8' });
|
|
assert.equal(child.status, 0, child.stderr);
|
|
const { seen, generation } = JSON.parse(child.stdout);
|
|
assert.ok(seen.filter(p => p === join(root, 'control')).length >= 3, 'runtime mkdir, halt rename and stops mkdir sync control');
|
|
assert.ok(seen.includes(join(root, 'control', 'stops')), 'generation directory creation syncs stops');
|
|
assert.ok(seen.includes(join(root, 'control', 'stops', generation)), 'stop receipt syncs generation directory');
|
|
});
|
|
|
|
test('a vanished group leader with unverified surviving members stays unknown, never quiescent', async t => {
|
|
const root = fixture(t);
|
|
const script = `const {spawn}=require('node:child_process');const c=spawn(process.execPath,['-e',"console.log('ready');setInterval(()=>{},1000)"],{stdio:['ignore','pipe','inherit']});c.stdout.once('data',()=>console.log(c.pid));process.on('SIGTERM',()=>process.exit(0));setInterval(()=>{},1000)`;
|
|
const p = spawn(process.execPath, ['-e', script], { detached: true, stdio: ['ignore', 'pipe', 'pipe'] });
|
|
t.after(() => { try { process.kill(-p.pid, 'SIGKILL'); } catch {} });
|
|
const [line] = await once(p.stdout, 'data'); const descendant = Number(String(line).trim());
|
|
receipt(root, p); const exited = once(p, 'exit'); p.kill('SIGTERM'); await exited;
|
|
const state = transition(root, actor);
|
|
const result = await api.sweepRuntimes(root, state.generation, { graceMs: 20, killWaitMs: 30 });
|
|
assert.equal(result.quiescent, false); assert.equal(result.remaining, 1);
|
|
assert.notEqual(api.processIdentity(descendant), null, 'never signal a group with no verified surviving anchor');
|
|
});
|
|
test('persisted member identity anchors a group after the leader exited before the sweep', async t => {
|
|
const root = fixture(t);
|
|
const script = `const {spawn}=require('node:child_process');const c=spawn(process.execPath,['-e',"process.on('SIGTERM',()=>{});console.log('ready');setInterval(()=>{},1000)"],{stdio:['ignore','pipe','inherit']});c.stdout.once('data',()=>console.log(c.pid));process.on('SIGTERM',()=>process.exit(0));setInterval(()=>{},1000)`;
|
|
const p = spawn(process.execPath, ['-e', script], { detached: true, stdio: ['ignore', 'pipe', 'pipe'] });
|
|
t.after(() => { try { process.kill(-p.pid, 'SIGKILL'); } catch {} });
|
|
const [line] = await once(p.stdout, 'data'), descendant = Number(String(line).trim()), r = receipt(root, p);
|
|
api.updateRuntime(root, r.id, { members: [api.processIdentity(descendant)] });
|
|
const exited = once(p, 'exit'); p.kill('SIGTERM'); await exited;
|
|
const state = transition(root, actor);
|
|
const result = await api.sweepRuntimes(root, state.generation, { graceMs: 30, killWaitMs: 1000 });
|
|
assert.equal(result.quiescent, true); assert.equal(api.processIdentity(descendant), null);
|
|
assert.deepEqual(api.readStops(root, state.generation)[0].signals, ['SIGTERM', 'SIGKILL']);
|
|
});
|
|
|
|
test('an unconfirmed secondary brake does not leave a verified workload running or claim success', async t => {
|
|
const root = fixture(t), { p, ended } = await child(t);
|
|
receipt(root, p); const state = transition(root, actor);
|
|
const result = await api.sweepRuntimes(root, state.generation, { graceMs: 60, killWaitMs: 1000,
|
|
beforeStop: () => { throw Error('fixture journal unavailable'); } });
|
|
await ended;
|
|
assert.equal(api.processIdentity(p.pid), null);
|
|
assert.equal(result.quiescent, false); assert.equal(result.remaining, 1);
|
|
const stops = api.readStops(root, state.generation); assert.equal(stops[0].outcome, 'unknown');
|
|
assert.equal(stops[0].problem, 'brake-unconfirmed');
|
|
});
|