Files
stack/packages/processing-control/src/state.mjs
T
rocko 4bdc7e0e86 processing: inspectable per-run stop seam for lifecycle fire
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.
2026-10-21 22:42:12 -05:00

130 lines
5.9 KiB
JavaScript

// Host-owned, per-data-root admission state. No model, git or credential imports.
import { randomUUID } from 'node:crypto';
import { constants, closeSync, fstatSync, fsyncSync, lstatSync, mkdirSync, openSync, readSync, realpathSync,
renameSync, unlinkSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
export class ProcessingError extends Error {
constructor(code) { super(code); this.name = 'ProcessingError'; this.code = code; }
}
const fail = code => { throw new ProcessingError(code); };
const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/;
const ID = /^[a-zA-Z0-9][a-zA-Z0-9_.:@/-]{0,159}$/;
const identity = value => typeof value === 'string' && ID.test(value);
const FIELDS = ['version', 'generation', 'operation', 'halted', 'at', 'business', 'human', 'recovered'];
const MAX_BYTES = 8192;
export function controlDirectory(dataRoot) {
try {
const root = realpathSync(dataRoot), st = lstatSync(root);
if (!st.isDirectory() || st.uid !== process.getuid()) fail('processing-control-unsafe');
return join(root, 'control');
} catch { fail('processing-control-unsafe'); }
}
function parent(dataRoot) {
const dir = controlDirectory(dataRoot);
let st;
try { st = lstatSync(dir); }
catch (e) {
if (e.code === 'ENOENT') fail('processing-uninitialized');
fail('processing-control-unsafe');
}
if (!st.isDirectory() || st.isSymbolicLink() || st.uid !== process.getuid() || (st.mode & 0o7777) !== 0o700)
fail('processing-control-unsafe');
return dir;
}
function valid(doc) {
if (!doc || typeof doc !== 'object' || Array.isArray(doc) ||
Object.keys(doc).length !== FIELDS.length || Object.keys(doc).some(k => !FIELDS.includes(k)) ||
doc.version !== 1 || !UUID.test(doc.generation) || typeof doc.halted !== 'boolean' ||
typeof doc.recovered !== 'boolean' || typeof doc.at !== 'string' ||
!Number.isFinite(Date.parse(doc.at)) || new Date(doc.at).toISOString() !== doc.at)
fail('processing-state-invalid');
if (doc.operation === null) {
if (doc.halted || doc.business !== null || doc.human !== null || doc.recovered) fail('processing-state-invalid');
} else if (!UUID.test(doc.operation) || !identity(doc.business) || !identity(doc.human)) fail('processing-state-invalid');
return Object.freeze(doc);
}
export function readControl(dataRoot) {
const file = join(parent(dataRoot), 'processing.json');
let fd;
try {
fd = openSync(file, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
const st = fstatSync(fd);
if (!st.isFile() || st.uid !== process.getuid() || (st.mode & 0o7777) !== 0o600 || st.size > MAX_BYTES)
fail('processing-state-invalid');
const buf = Buffer.alloc(MAX_BYTES + 1);
let size = 0, n;
while ((n = readSync(fd, buf, size, buf.length - size, null)) > 0) {
size += n;
if (size > MAX_BYTES) fail('processing-state-invalid');
}
return valid(JSON.parse(new TextDecoder('utf-8', { fatal: true }).decode(buf.subarray(0, size))));
} catch { fail('processing-state-invalid'); }
finally { if (fd !== undefined) closeSync(fd); }
}
function publish(dataRoot, doc) {
const dir = parent(dataRoot), file = join(dir, 'processing.json'), temp = join(dir, `.state-${randomUUID()}`);
let fd, dirfd;
try {
fd = openSync(temp, constants.O_WRONLY | constants.O_CREAT | constants.O_EXCL | constants.O_NOFOLLOW, 0o600);
writeFileSync(fd, JSON.stringify(valid(doc)) + '\n');
fsyncSync(fd); closeSync(fd); fd = undefined;
renameSync(temp, file);
dirfd = openSync(dir, constants.O_RDONLY | constants.O_DIRECTORY | constants.O_NOFOLLOW);
fsyncSync(dirfd);
} catch { fail('processing-state-invalid'); }
finally {
if (fd !== undefined) closeSync(fd);
if (dirfd !== undefined) closeSync(dirfd);
try { unlinkSync(temp); } catch (e) { if (e.code !== 'ENOENT') fail('processing-state-invalid'); }
}
}
// First-use host bootstrap only. A present directory with no file is not a new installation.
export function initializeControl(dataRoot) {
const dir = controlDirectory(dataRoot);
try { mkdirSync(dir, { mode: 0o700 }); }
catch (e) {
if (e.code === 'EEXIST') return readControl(dataRoot);
fail('processing-control-unsafe');
}
const rootfd = openSync(realpathSync(dataRoot), constants.O_RDONLY | constants.O_DIRECTORY | constants.O_NOFOLLOW);
try { fsyncSync(rootfd); } finally { closeSync(rootfd); }
publish(dataRoot, { version: 1, generation: randomUUID(), operation: null, halted: false,
at: new Date().toISOString(), business: null, human: null, recovered: false });
return readControl(dataRoot);
}
export function assertProcessing(dataRoot, generation = null) {
const state = readControl(dataRoot);
if (state.halted) fail('processing-halted');
if (generation !== null && state.generation !== generation) fail('processing-generation-changed');
return state;
}
// Called only by the coordinator after the broker has proved a human.
export function transition(dataRoot, { halted, business, human, generation = null }) {
if (typeof halted !== 'boolean' || !identity(business) || !identity(human)) fail('invalid-request');
parent(dataRoot);
let current, recovered = false;
try { current = readControl(dataRoot); }
catch (e) {
if (!halted || e.code !== 'processing-state-invalid') throw e;
recovered = true;
}
if (!halted && generation !== current.generation) fail('processing-generation-changed');
if (current?.halted === halted) return { ...current, changed: false };
const operation = randomUUID();
const state = { version: 1, generation: operation, operation, halted,
at: new Date().toISOString(), business, human, recovered };
publish(dataRoot, state);
return { ...state, changed: true };
}
export function processingStatus(dataRoot) {
try { return readControl(dataRoot); }
catch (e) {
if (!(e instanceof ProcessingError)) throw e;
return { halted: e.code === 'processing-uninitialized' ? null : true, invalid: true, code: e.code };
}
}