stack/packages/processing-control/src/containers.mjs

import { randomUUID } from 'node:crypto';
import { spawnSync, execFile } from 'node:child_process';
import { promisify } from 'node:util';
import { initializeControl, assertProcessing } from './state.mjs';
import { registerRuntime, updateRuntime, runtime, rootTag } from './runtimes.mjs';
const execute = promisify(execFile);
const pause = ms => new Promise(resolve => setTimeout(resolve, ms));
export function prepareContainer(dataRoot, record) {
  initializeControl(dataRoot);
  const state = assertProcessing(dataRoot);
  const containerName = `mosaic-${rootTag(dataRoot).slice(0, 12)}-${randomUUID()}`;
  const managed = registerRuntime(dataRoot, { ...record, containerName });
  assertProcessing(dataRoot, state.generation);
  return managed;
}
export function containerArgs(record, { tty = false, args = [] } = {}) {
  return ['compose', 'run', '--rm', ...(!tty ? ['-T'] : []), '--name', record.containerName,
    '--label', `mosaic.root=${record.root}`, '--label', `mosaic.runtime=${record.id}`,
    '-e', `MOSAIC_PROCESSING_GENERATION=${record.generation}`, 'mosaic-agent', ...args];
}
export function runContainer(dataRoot, record, { tty = false, args = [], ...options } = {}) {
  const managed = prepareContainer(dataRoot, record);
  assertProcessing(dataRoot, managed.generation);
  // Keep this synchronous. The independent broker sweeps the prepublished name while this waits.
  const proc = spawnSync('docker', containerArgs(managed, { tty, args }), options);
  const current = runtime(dataRoot, managed.id);
  // An interrupted Docker client is not proof its container exited. Keep that receipt for inspection.
  if (!proc.error && proc.status === 0) updateRuntime(dataRoot, managed.id, { phase: 'ended' });
  return { proc, managed, haltGeneration: current.haltGeneration ?? null };
}
export function createContainerStopper({ dataRoot = null, docker = async args => {
  const r = await execute('docker', args, { encoding: 'utf8', timeout: 2000, maxBuffer: 131072 });
  return r.stdout;
} } = {}) {
  return async (record, { graceMs = 5000, killWaitMs = 2000 } = {}) => {
    const signals = [];
    let id = record.containerId ?? null, seen = false;
    const start = Date.now(), termEnd = start + graceMs, end = termEnd + killWaitMs;
    async function inspect(target) {
      const doc = JSON.parse(await docker(['inspect', '--format', '{{json .}}', target]));
      if (!doc || !/^[a-f0-9]{64}$/.test(doc.Id) || doc.Name !== '/' + record.containerName ||
          doc.Config?.Labels?.['mosaic.root'] !== record.root || doc.Config?.Labels?.['mosaic.runtime'] !== record.id ||
          typeof doc.State?.Running !== 'boolean' || (id && doc.Id !== id)) {
        const error = new Error('container identity mismatch'); error.code = 'identity-mismatch'; throw error;
      }
      if (!id) {
        id = doc.Id;
        if (dataRoot) updateRuntime(dataRoot, record.id, { containerId: id });
      }
      seen = true;
      return doc;
    }
    while (Date.now() <= end) {
      try {
        const doc = await inspect(id ?? record.containerName);
        if (!doc.State.Running) return { outcome: signals.includes('SIGKILL') ? 'killed' : signals.length ? 'terminated' : 'already-exited', signals };
        const next = !signals.length ? 'TERM' : Date.now() >= termEnd && !signals.includes('SIGKILL') ? 'KILL' : null;
        if (next) {
          // Revalidate immediately before each signal, targeting the immutable inspected id.
          const checked = await inspect(id);
          if (!checked.State.Running) continue;
          await docker(['kill', '--signal', next, id]);
          signals.push('SIG' + next);
        }
      } catch (e) {
        // A missing name can still be a queued create. Only a container observed before a signal
        // and then removed by --rm is known to have finished. Never signal on an inspect failure.
        if (e.code === 'identity-mismatch') return { outcome: 'unknown', signals };
        if (id && seen && signals.length) {
          try {
            const listed = await docker(['ps', '--all', '--no-trunc', '--filter', `id=${id}`, '--format', '{{.ID}}']);
            if (listed.trim() === '') return { outcome: signals.includes('SIGKILL') ? 'killed' : 'terminated', signals };
          } catch { /* a failed Docker request proves nothing about the workload */ }
        }
        if (dataRoot) {
          try {
            if (runtime(dataRoot, record.id).phase === 'ended')
              return { outcome: signals.includes('SIGKILL') ? 'killed' : signals.length ? 'terminated' : 'already-exited', signals };
          } catch { return { outcome: 'unknown', signals }; }
        }
      }
      await pause(Math.min(20, Math.max(1, end - Date.now())));
    }
    return { outcome: 'unknown', signals };
  };
}