jason.woltje
d070ed6d4a
A stale-record init now renames the record aside, checks it is still the one it observed, and puts it back otherwise. Two inits racing over the same stale record can no longer delete the record of a live endpoint the other one just started. Fixes the P2 from the test-health review.
249 lines
16 KiB
JavaScript
249 lines
16 KiB
JavaScript
// `mosaic dev fixture init|status|stop`: the deterministic model fixture for a
|
|
// throwaway named instance (r4 §7). init starts the fixture provider's loopback
|
|
// endpoint (packages/plugins/model-provider/fixture/endpoint.mjs) and records a
|
|
// non-secret state file under the instance data root. It never writes the
|
|
// system config. A bus started with --dev-local-owner on that instance hands the
|
|
// endpoint to the launcher as its session proxy context (fixtureLauncherOptions),
|
|
// so a launch runs the real launcher, sandbox, runner and pi against it.
|
|
// Refuses: the default dev instance, an instance with a proxy configured, a
|
|
// provider credential in the environment, a non-loopback endpoint and a second
|
|
// fixture. A record whose endpoint is gone (its lifetime ended, or it crashed)
|
|
// is stale: init replaces it. Exit codes as errors.mjs: 2 invalid, 3 refused, 4 usage.
|
|
import { spawn } from 'node:child_process';
|
|
import { closeSync, constants, fstatSync, linkSync, lstatSync, mkdirSync, openSync, readFileSync, renameSync, rmSync, writeFileSync } from 'node:fs';
|
|
import { get } from 'node:http';
|
|
import { dirname, join, resolve } from 'node:path';
|
|
import { fileURLToPath } from 'node:url';
|
|
import { CliError } from './errors.mjs';
|
|
import { DEV_OWNER, devOwnerInstance } from './dev-local-owner.mjs';
|
|
import { loadSystem } from './config.mjs';
|
|
import { parseArgs } from './thread.mjs';
|
|
import { isProviderVariable } from '../../seat/src/session.mjs';
|
|
import { FIXTURE_ENTRY, FIXTURE_KEY, FIXTURE_KEY_NAME, FIXTURE_MODEL } from '../../plugins/model-provider/fixture/index.mjs';
|
|
import { TTL_DEFAULT_S, TTL_MAX_S, TTL_MIN_S } from '../../plugins/model-provider/fixture/endpoint.mjs';
|
|
|
|
export const ENDPOINT = resolve(dirname(fileURLToPath(import.meta.url)), '../../plugins/model-provider/fixture/endpoint.mjs');
|
|
export const DEV_USAGE = `mosaic dev fixture init --model deterministic [--instance NAME] [--bind 127.0.0.1] [--ttl SECONDS] [--json]
|
|
mosaic dev fixture status [--json]
|
|
mosaic dev fixture stop [--json]`;
|
|
export const fixtureDir = dataRoot => join(dataRoot, 'dev-fixture');
|
|
export const fixtureFile = dataRoot => join(fixtureDir(dataRoot), 'fixture.json');
|
|
const LOOPBACK = /^http:\/\/127\.0\.0\.1:([0-9]{1,5})$/;
|
|
const PRINCIPAL = `${DEV_OWNER.provider}:${DEV_OWNER.sub}`;
|
|
|
|
const refuse = (code, message) => Object.assign(new CliError(`${code}: ${message}`, 3), { code });
|
|
const invalid = message => new CliError(`${message}\nusage: ${DEV_USAGE}`, 4);
|
|
|
|
// Names, never values, of provider credential variables the environment holds.
|
|
export function credentialVariables(env) {
|
|
return Object.keys(env).filter(k => isProviderVariable(k) && Boolean(env[k])).sort();
|
|
}
|
|
function noCredentials(env) {
|
|
const held = credentialVariables(env);
|
|
if (held.length) throw refuse('fixture-credential-present', `the fixture runs without real credentials; unset ${held.join(', ')}`);
|
|
}
|
|
|
|
function ownedPrivate(path, directory) {
|
|
const st = lstatSync(path);
|
|
if (st.isSymbolicLink() || (directory ? !st.isDirectory() : !st.isFile()) || st.uid !== process.getuid() || (st.mode & 0o077))
|
|
throw refuse('fixture-state-unsafe', `${path} is not a private ${directory ? 'directory' : 'file'} of this user`);
|
|
}
|
|
|
|
// The recorded fixture, or null when there is none. Anything malformed refuses.
|
|
export function readFixture(dataRoot) {
|
|
const file = fixtureFile(dataRoot);
|
|
try { ownedPrivate(fixtureDir(dataRoot), true); }
|
|
catch (e) { if (e.code === 'ENOENT') return null; throw e; }
|
|
let fd;
|
|
try { fd = openSync(file, constants.O_RDONLY | constants.O_NOFOLLOW); }
|
|
catch (e) { if (e.code === 'ENOENT') return null; throw refuse('fixture-state-unsafe', `${file} can't be opened (${e.code})`); }
|
|
let doc;
|
|
try {
|
|
const st = fstatSync(fd);
|
|
if (!st.isFile() || st.uid !== process.getuid() || (st.mode & 0o777) !== 0o600) throw refuse('fixture-state-unsafe', `${file} is not a private file of this user`);
|
|
doc = JSON.parse(readFileSync(fd, 'utf8'));
|
|
} catch (e) { if (e instanceof CliError) throw e; throw refuse('fixture-state-invalid', `${file} is not JSON`); }
|
|
finally { closeSync(fd); }
|
|
const m = LOOPBACK.exec(doc?.endpoint ?? '');
|
|
if (!m) throw refuse('fixture-endpoint-not-loopback', `${file} names an endpoint that is not http://127.0.0.1:<port>`);
|
|
if (doc.fixtureVersion !== 1 || doc.model !== 'deterministic' || Number(m[1]) !== doc.port || !Number.isSafeInteger(doc.pid) || doc.pid < 2
|
|
|| typeof doc.instance !== 'string' || JSON.stringify(doc.entry) !== JSON.stringify(FIXTURE_ENTRY))
|
|
throw refuse('fixture-state-invalid', `${file} is not a fixture record`);
|
|
return doc;
|
|
}
|
|
|
|
const cmdline = pid => { try { return readFileSync(`/proc/${pid}/cmdline`, 'utf8').split('\0'); } catch { return null; } };
|
|
// Signal only the endpoint this command started: its argv[1] is ENDPOINT.
|
|
export const isEndpoint = pid => (cmdline(pid) ?? [])[1] === ENDPOINT;
|
|
|
|
function health(endpoint, timeoutMs = 2000) {
|
|
return new Promise(resolve => {
|
|
const req = get(`${endpoint}/health`, { timeout: timeoutMs }, res => {
|
|
let body = ''; res.setEncoding('utf8'); res.on('data', c => { if (body.length < 4096) body += c; });
|
|
res.on('end', () => { try { resolve(res.statusCode === 200 ? JSON.parse(body) : null); } catch { resolve(null); } });
|
|
});
|
|
req.on('timeout', () => req.destroy()); req.on('error', () => resolve(null));
|
|
});
|
|
}
|
|
|
|
function instanceOf(io, parsed) {
|
|
let instance;
|
|
try { instance = devOwnerInstance(io.env); }
|
|
catch (e) { throw refuse('fixture-instance-refused', `${e.message}; set MOSAIC_CONFIG to the named instance's config`); }
|
|
const named = parsed.values['--instance'];
|
|
if (named !== undefined && named !== instance) throw refuse('fixture-instance-mismatch', `--instance ${named} is not the instance MOSAIC_CONFIG names (${instance})`);
|
|
return instance;
|
|
}
|
|
|
|
// --ttl: the endpoint's lifetime in whole seconds, TTL_DEFAULT_S when absent. 0 is no expiry,
|
|
// for a long-running dev instance; it is safe only because this command runs on a named dev
|
|
// instance and nothing else starts the endpoint with 0.
|
|
function ttlOf(raw) {
|
|
if (raw === undefined) return TTL_DEFAULT_S;
|
|
const n = /^(0|[1-9][0-9]{0,6})$/.test(raw) ? Number(raw) : NaN;
|
|
if (n === 0 || (n >= TTL_MIN_S && n <= TTL_MAX_S)) return n;
|
|
throw invalid(`--ttl takes 0 (no expiry) or whole seconds from ${TTL_MIN_S} to ${TTL_MAX_S}`);
|
|
}
|
|
|
|
function startEndpoint(logFile, ttlSeconds, { timeoutMs = 10000 } = {}) {
|
|
const log = openSync(logFile, 'a', 0o600);
|
|
let child;
|
|
try {
|
|
child = spawn(process.execPath, [ENDPOINT], { detached: true, stdio: ['ignore', 'pipe', log], env: { PATH: process.env.PATH ?? '/usr/bin:/bin', MOSAIC_FIXTURE_TTL_S: String(ttlSeconds) } });
|
|
} finally { closeSync(log); }
|
|
return new Promise((resolve, reject) => {
|
|
let out = '';
|
|
const fail = why => { clearTimeout(clock); try { child.kill('SIGKILL'); } catch {} reject(refuse('fixture-start-failed', why)); };
|
|
const clock = setTimeout(() => fail('the endpoint did not report ready'), timeoutMs);
|
|
child.on('error', e => fail(e.message));
|
|
child.on('exit', code => fail(`the endpoint exited ${code}`));
|
|
child.stdout.setEncoding('utf8').on('data', s => {
|
|
out += s; const nl = out.indexOf('\n'); if (nl < 0) return;
|
|
clearTimeout(clock); child.removeAllListeners('exit'); child.stdout.destroy(); child.unref();
|
|
try { const ready = JSON.parse(out.slice(0, nl)); if (ready.ready !== true || !Number.isSafeInteger(ready.port)) throw Error(); resolve({ child, port: ready.port }); }
|
|
catch { fail('the endpoint printed something that is not a ready line'); }
|
|
});
|
|
});
|
|
}
|
|
|
|
// Remove the stale record this init inspected, and only that one. The record is taken by an
|
|
// atomic rename, then checked: if a concurrent init already replaced it with a live endpoint's
|
|
// record, that record goes back (link refuses to overwrite) and this init refuses fixture-exists.
|
|
// An unlink by path after the check would delete the other init's new record instead.
|
|
export function takeStale(dataRoot, existing, instance) {
|
|
const file = fixtureFile(dataRoot), aside = `${file}.stale-${process.pid}`;
|
|
try { renameSync(file, aside); } catch (e) { if (e.code === 'ENOENT') return; throw e; }
|
|
let taken = null;
|
|
try { taken = JSON.parse(readFileSync(aside, 'utf8')); } catch {}
|
|
if (taken?.pid === existing.pid && taken?.startedAt === existing.startedAt) { rmSync(aside, { force: true }); return; }
|
|
// EEXIST means a third init published meanwhile; that record stays and this init still refuses.
|
|
try { linkSync(aside, file); } catch {} finally { rmSync(aside, { force: true }); }
|
|
throw refuse('fixture-exists', `instance ${instance} already has a fixture; mosaic dev fixture stop removes it`);
|
|
}
|
|
|
|
const print = (io, json, value, line) => io.stdout.write(json ? `${JSON.stringify(value, null, 2)}\n` : `${line}\n`);
|
|
|
|
async function init(args, io, deps) {
|
|
const parsed = parseArgs(args, `usage: ${DEV_USAGE}`, { flags: ['--json'], values: ['--model', '--instance', '--bind', '--ttl'] });
|
|
if (parsed.positional.length) throw invalid('fixture init takes no positional arguments');
|
|
if (parsed.values['--model'] !== 'deterministic') throw invalid('fixture init needs --model deterministic');
|
|
const ttlSeconds = ttlOf(parsed.values['--ttl']);
|
|
const bind = parsed.values['--bind'] ?? '127.0.0.1';
|
|
if (bind !== '127.0.0.1') throw refuse('fixture-endpoint-not-loopback', `the fixture binds 127.0.0.1 only, not ${bind}`);
|
|
const instance = instanceOf(io, parsed);
|
|
noCredentials(io.env);
|
|
const system = deps.system ?? loadSystem({ env: io.env, instance });
|
|
if (system.proxy) throw refuse('fixture-proxy-configured', `instance ${instance} has a model proxy configured; the fixture is for an instance without one`);
|
|
const existing = readFixture(system.dataRoot);
|
|
if (existing && isEndpoint(existing.pid)) throw refuse('fixture-exists', `instance ${instance} already has a fixture; mosaic dev fixture stop removes it`);
|
|
// P19 #882: an endpoint whose lifetime ended left its record, and init refused fixture-exists
|
|
// until `dev fixture stop`. A record whose pid is no longer the endpoint is stale; it goes.
|
|
// A second init racing this one still loses at the exclusive write of the new record.
|
|
if (existing) takeStale(system.dataRoot, existing, instance);
|
|
const replaced = existing ? { pid: existing.pid, startedAt: existing.startedAt ?? null } : null;
|
|
try { mkdirSync(fixtureDir(system.dataRoot), { mode: 0o700 }); } catch (e) { if (e.code !== 'EEXIST') throw e; }
|
|
ownedPrivate(fixtureDir(system.dataRoot), true);
|
|
const { child, port } = await startEndpoint(join(fixtureDir(system.dataRoot), 'endpoint.log'), ttlSeconds);
|
|
const endpoint = `http://127.0.0.1:${port}`;
|
|
const started = new Date();
|
|
const record = { fixtureVersion: 1, fixture: 'fixture', model: 'deterministic', modelKey: FIXTURE_KEY_NAME, entry: FIXTURE_ENTRY,
|
|
instance, endpoint, port, pid: child.pid, principal: PRINCIPAL, upstream: 'none', startedAt: started.toISOString(),
|
|
ttlSeconds, expiresAt: ttlSeconds ? new Date(started.getTime() + ttlSeconds * 1000).toISOString() : null };
|
|
try {
|
|
if (!(await health(endpoint))?.ok) throw refuse('fixture-start-failed', 'the endpoint did not answer its health check');
|
|
writeFileSync(fixtureFile(system.dataRoot), `${JSON.stringify(record, null, 2)}\n`, { mode: 0o600, flag: 'wx' });
|
|
} catch (e) { try { process.kill(child.pid, 'SIGTERM'); } catch {} throw e; }
|
|
const receipt = { receipt: 'dev-fixture-init', ...record, replaced, loopback: true, external: false, state: fixtureFile(system.dataRoot) };
|
|
print(io, parsed.flags.has('--json'), receipt,
|
|
`fixture ready: instance ${instance}, model ${FIXTURE_MODEL} (launch.models key ${FIXTURE_KEY_NAME}), endpoint ${endpoint} (loopback, no upstream), principal ${PRINCIPAL}, pid ${child.pid}, ${ttlSeconds ? `expires ${record.expiresAt}` : 'no expiry'}${replaced ? `; replaced the stale record of pid ${replaced.pid}` : ''}`);
|
|
}
|
|
|
|
async function status(args, io, deps) {
|
|
const parsed = parseArgs(args, `usage: ${DEV_USAGE}`, { flags: ['--json'], values: ['--instance'] });
|
|
if (parsed.positional.length) throw invalid('fixture status takes no positional arguments');
|
|
const instance = instanceOf(io, parsed);
|
|
const system = deps.system ?? loadSystem({ env: io.env, instance });
|
|
const record = readFixture(system.dataRoot);
|
|
if (!record) throw new CliError(`instance ${instance} has no fixture; run mosaic dev fixture init --model deterministic`, 2);
|
|
const alive = isEndpoint(record.pid);
|
|
const answer = alive ? await health(record.endpoint) : null;
|
|
const report = { instance, ready: Boolean(alive && answer?.ok && !system.proxy), endpoint: record.endpoint, loopback: true, upstream: 'none', external: false,
|
|
model: record.model, modelKey: record.modelKey, entry: record.entry, principal: record.principal, pid: record.pid, alive,
|
|
proxyConfigured: Boolean(system.proxy), health: answer, startedAt: record.startedAt, ttlSeconds: record.ttlSeconds ?? null, expiresAt: record.expiresAt ?? null };
|
|
print(io, parsed.flags.has('--json'), report, `fixture ${report.ready ? 'ready' : 'not ready'}: ${record.endpoint}, pid ${record.pid}${alive ? '' : ' (not running; init replaces the record)'}`);
|
|
}
|
|
|
|
async function stop(args, io, deps) {
|
|
const parsed = parseArgs(args, `usage: ${DEV_USAGE}`, { flags: ['--json'], values: ['--instance'] });
|
|
if (parsed.positional.length) throw invalid('fixture stop takes no positional arguments');
|
|
const instance = instanceOf(io, parsed);
|
|
const system = deps.system ?? loadSystem({ env: io.env, instance });
|
|
const record = readFixture(system.dataRoot);
|
|
if (!record) throw new CliError(`instance ${instance} has no fixture`, 2);
|
|
let signaled = false;
|
|
if (isEndpoint(record.pid)) {
|
|
process.kill(record.pid, 'SIGTERM'); signaled = true;
|
|
for (let i = 0; i < 50 && isEndpoint(record.pid); i++) await new Promise(r => setTimeout(r, 100));
|
|
if (isEndpoint(record.pid)) throw new CliError(`fixture endpoint pid ${record.pid} did not stop`, 1);
|
|
}
|
|
rmSync(fixtureFile(system.dataRoot));
|
|
print(io, parsed.flags.has('--json'), { receipt: 'dev-fixture-stop', instance, pid: record.pid, signaled, endpoint: record.endpoint },
|
|
`fixture stopped: instance ${instance}, pid ${record.pid}${signaled ? '' : ' (was not running)'}`);
|
|
}
|
|
|
|
export async function devCommand(rest, io, deps = {}) {
|
|
const [what, sub, ...args] = rest;
|
|
if (what !== 'fixture') throw invalid(`unknown dev command ${what ?? ''}`.trim());
|
|
if (sub === 'init') return init(args, io, deps);
|
|
if (sub === 'status') return status(args, io, deps);
|
|
if (sub === 'stop') return stop(args, io, deps);
|
|
throw invalid(`unknown dev fixture command ${sub ?? ''}`.trim());
|
|
}
|
|
|
|
// For `mosaic bus start --dev-local-owner` only: the fixture as the launcher's
|
|
// session proxy context, or nothing when the instance has no fixture. A fixture
|
|
// next to a configured proxy, or a credential in the bus environment, refuses.
|
|
// True when the organization file routes a launch to the fixture provider.
|
|
export function orgUsesFixture(doc) {
|
|
const providers = doc?.plugins?.['model-provider'];
|
|
if (Array.isArray(providers) && providers.includes('fixture')) return true;
|
|
return Object.values(doc?.launch?.models ?? {}).some(m => m?.module === 'fixture');
|
|
}
|
|
// A fixture organization's launches get no provider credential, so one instance can hold
|
|
// fixture organizations and real-model organizations side by side. Other organizations keep env.
|
|
export function fixtureLaunchEnv(doc, env) {
|
|
if (!orgUsesFixture(doc)) return env;
|
|
return Object.fromEntries(Object.entries(env).filter(([k]) => !credentialVariables({ [k]: env[k] }).length));
|
|
}
|
|
// `doc`, when given, is the organization file: an organization that does not use the fixture
|
|
// launches as configured, without the fixture proxy.
|
|
export function fixtureLauncherOptions({ system, env, instance, doc }) {
|
|
const record = readFixture(system.dataRoot);
|
|
if (!record) return {};
|
|
if (doc !== undefined && !orgUsesFixture(doc)) return {};
|
|
if (record.instance !== instance) throw refuse('fixture-instance-mismatch', `the fixture was started for instance ${record.instance}, not ${instance}`);
|
|
if (system.proxy) throw refuse('fixture-proxy-configured', 'the instance has both a model proxy and a fixture');
|
|
noCredentials(env);
|
|
return { testProxy: { baseUrl: record.endpoint, key: FIXTURE_KEY } };
|
|
}
|