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.
68 lines
3.8 KiB
JavaScript
68 lines
3.8 KiB
JavaScript
import test from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import { spawn } from 'node:child_process';
|
|
import { once } from 'node:events';
|
|
import { mkdtempSync, rmSync } from 'node:fs';
|
|
import { tmpdir } from 'node:os';
|
|
import { join } from 'node:path';
|
|
import { readControl, assertProcessing } from '../src/state.mjs';
|
|
import { registerRuntime, updateRuntime, processIdentity, unregisterRuntime } from '../src/runtimes.mjs';
|
|
const { createProcessing } = await import('../src/coordinator.mjs').catch(e => { if (e.code === 'ERR_MODULE_NOT_FOUND') return {}; throw e; });
|
|
const actor = { business: 'demo', human: 'jason' };
|
|
function fixture(t) {
|
|
assert.equal(typeof createProcessing, 'function');
|
|
const root = mkdtempSync(join(tmpdir(), 'processing-coordinator-'));
|
|
const c = createProcessing({ dataRoot: root, graceMs: 80, killWaitMs: 1000 });
|
|
t.after(async () => { await c.close(); rmSync(root, { recursive: true, force: true }); });
|
|
return { root, c };
|
|
}
|
|
|
|
test('halt publishes before awaiting stops, resume refuses pending, and only clear permits new work', async t => {
|
|
const { root, c } = fixture(t);
|
|
const p = spawn(process.execPath, ['-e', "process.on('SIGTERM',()=>{});console.log('ready');setInterval(()=>{},1000)"], { detached: true, stdio: ['ignore', 'pipe', 'ignore'] });
|
|
const exit = once(p, 'exit');
|
|
t.after(async () => { if (p.signalCode === null && p.exitCode === null) { try { process.kill(-p.pid, 'SIGKILL'); } catch {} await exit; } });
|
|
await once(p.stdout, 'data');
|
|
const r = registerRuntime(root, { business: 'demo', run: 'standin', kind: 'session' });
|
|
updateRuntime(root, r.id, { ...processIdentity(p.pid), group: true, phase: 'running' });
|
|
const halted = c.halt(actor);
|
|
assert.equal(readControl(root).halted, true, 'durable gate is set before any stop await');
|
|
await assert.rejects(c.resume({ ...actor, generation: readControl(root).generation }), { code: 'processing-stop-pending' });
|
|
const result = await halted;
|
|
assert.equal(result.quiescent, true);
|
|
assert.deepEqual(await exit, [null, 'SIGKILL']);
|
|
assert.throws(() => assertProcessing(root), { code: 'processing-halted' });
|
|
const resumed = await c.resume({ ...actor, generation: result.generation });
|
|
assert.equal(resumed.halted, false);
|
|
assert.doesNotThrow(() => assertProcessing(root));
|
|
});
|
|
|
|
test('a new coordinator replays a persisted halt and an unresolved start keeps clear refused', async t => {
|
|
const { root, c } = fixture(t);
|
|
const pending = registerRuntime(root, { business: 'demo', run: 'starting', kind: 'session' });
|
|
const result = await c.halt(actor);
|
|
assert.equal(result.quiescent, false);
|
|
await c.close();
|
|
const next = createProcessing({ dataRoot: root, graceMs: 10, killWaitMs: 10 });
|
|
try {
|
|
assert.equal(next.read().halted, true);
|
|
await assert.rejects(next.resume({ ...actor, generation: result.generation }), { code: 'processing-stop-pending' });
|
|
unregisterRuntime(root, pending.id);
|
|
const resumed = await next.resume({ ...actor, generation: result.generation });
|
|
assert.equal(resumed.halted, false);
|
|
} finally { await next.close(); }
|
|
});
|
|
|
|
test('a new halt generation never inherits a previous quiescent status before its sweep', async t => {
|
|
const { mkdtempSync, rmSync } = await import('node:fs');
|
|
const { tmpdir } = await import('node:os');
|
|
const { join } = await import('node:path');
|
|
const { transition } = await import('../src/state.mjs');
|
|
const root = mkdtempSync(join(tmpdir(), 'processing-counts-'));
|
|
const processing = createProcessing({ dataRoot: root });
|
|
t.after(async () => { await processing.close(); rmSync(root, { recursive: true, force: true }); });
|
|
transition(root, { business: 'demo', human: 'jason', halted: true });
|
|
assert.equal(processing.read().quiescent, false);
|
|
await processing.reconcile(); assert.equal(processing.read().quiescent, true);
|
|
});
|