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.
75 lines
3.4 KiB
JavaScript
75 lines
3.4 KiB
JavaScript
import { initializeControl, processingStatus, transition, ProcessingError } from './state.mjs';
|
|
import { readStops, sweepRuntimes } from './runtimes.mjs';
|
|
import { withProcessing, currentProcessing } from './effects.mjs';
|
|
|
|
// The control bus stays reachable. Only its processing, supervised workloads and background effects stop.
|
|
export function createProcessing({ dataRoot, graceMs = 5000, killWaitMs = 2000, containerStop = null,
|
|
onTransition = () => {}, onStop = () => {}, beforeStop = () => {}, onResume = () => {}, log = () => {} }) {
|
|
try { initializeControl(dataRoot); } catch (e) {
|
|
if (!e.code?.startsWith('processing-')) throw e;
|
|
}
|
|
let active = null, timer = null, closed = false;
|
|
let last = { quiescent: true, remaining: 0, stopped: 0 };
|
|
let settledGeneration = null;
|
|
const read = () => {
|
|
const state = processingStatus(dataRoot);
|
|
const counts = state.invalid || (state.halted && state.generation !== settledGeneration)
|
|
? { quiescent: false, remaining: null, stopped: 0 } : last;
|
|
return { ...state, ...counts, stopping: active !== null };
|
|
};
|
|
function reconcile() {
|
|
if (active) return active;
|
|
const state = processingStatus(dataRoot);
|
|
if (state.invalid) return Promise.resolve({ quiescent: false, remaining: null, stopped: 0 });
|
|
active = Promise.resolve().then(async () => {
|
|
await onTransition(state);
|
|
if (state.halted) {
|
|
const result = await sweepRuntimes(dataRoot, state.generation, { graceMs, killWaitMs, containerStop, beforeStop });
|
|
for (const stop of readStops(dataRoot, state.generation)) await onStop(stop);
|
|
const { results, ...summary } = result;
|
|
last = summary;
|
|
settledGeneration = state.generation;
|
|
} else last = { quiescent: true, remaining: 0, stopped: 0 };
|
|
return last;
|
|
}).finally(() => { active = null; });
|
|
return active;
|
|
}
|
|
return {
|
|
read,
|
|
work: fn => withProcessing(dataRoot, fn),
|
|
guard: () => currentProcessing()?.guard(),
|
|
signal: () => currentProcessing()?.signal,
|
|
stopReceipts: generation => readStops(dataRoot, generation),
|
|
// Publication is synchronous: no timer/model/HTTP wait can delay the admission fence.
|
|
halt(actor) {
|
|
const published = transition(dataRoot, { ...actor, halted: true });
|
|
return reconcile().then(summary => ({ ...published, ...summary }));
|
|
},
|
|
async resume(actor) {
|
|
if (active) throw new ProcessingError('processing-stop-pending');
|
|
const state = processingStatus(dataRoot);
|
|
if (state.invalid) throw new ProcessingError(state.code);
|
|
if (state.generation !== actor.generation) throw new ProcessingError('processing-generation-changed');
|
|
const summary = await reconcile();
|
|
if (!summary.quiescent) throw new ProcessingError('processing-stop-pending');
|
|
await onResume(state.generation);
|
|
const published = transition(dataRoot, { ...actor, halted: false });
|
|
await onTransition(published);
|
|
last = { quiescent: true, remaining: 0, stopped: 0 };
|
|
return { ...published, ...last };
|
|
},
|
|
reconcile,
|
|
start() {
|
|
if (timer || closed) return;
|
|
const tick = () => {
|
|
if (!active && processingStatus(dataRoot).halted) reconcile().catch(() => log('processing-reconcile-failed'));
|
|
};
|
|
tick(); timer = setInterval(tick, 250); timer.unref();
|
|
},
|
|
async close() {
|
|
closed = true; clearInterval(timer); timer = null;
|
|
await active;
|
|
},
|
|
};
|
|
}
|