diff --git a/AGENTS.md b/AGENTS.md index 587d940c..f077e2dc 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -39,12 +39,17 @@ Before opening or updating a PR for asynchronous behavior, test every applicable - An abort error fires, then subprocess close arrives later. - Submit starts, then the user edits the composer or switches items, then the response returns. - Referenced code is reassigned or the snapshot changes, then retry or rendering occurs. +- A synchronous caller hook inside a lifecycle (such as a callback that records a spawned process group) throws, blocks past the deadline, or aborts the signal, then the lifecycle continues. Timers and listeners that bound the lifecycle must already be armed when the hook runs, and a signal already aborted when its listener is attached must still take effect. +- A child process exits, but a descendant outside its process group still holds its output pipes. Every reproduced race requires a failing-before and passing-after regression. Assert both the visible result and the durable state when they can diverge. ## Review readiness - Before requesting or re-requesting an automated Copilot review, self-review the full current diff, fix every issue found, and repeat the self-review and fix cycle until a complete pass finds no new issues. Re-run the relevant validation after fixes; only then request Copilot review. +- Each self-review pass rereads every changed function in full against the base, not only the lines changed since the previous round. Code unchanged since the first commit of the PR still gets reviewed in every pass. +- Treat every behavioural claim the change makes, in code comments, the PR body or docs (for example "pauses every 1,000 entries", "settles only after exit", "never throws", "bounded by N seconds"), as something to verify. Trace each claim through every path that can break it, including nested loops, callbacks, error paths and early returns, and give it a test that fails if the claim is false. +- The author's self-review is not enough for concurrency, process, subprocess, timer or resource-cleanup code. Before requesting Copilot review, also run an independent review that does not share the author's context: `/codex review`, a separate review agent, or `/code-review` at `high` effort or above. Fix its findings like any other. - Run final validation against the exact pushed head after the last change. - Report current test counts separately from historical milestone counts. - Before requesting automated review, report the current head, CI state, mergeability, unresolved threads, and deferred follow-up issues. diff --git a/agents/container/storage.ts b/agents/container/storage.ts index b6747eeb..7f61bdc9 100644 --- a/agents/container/storage.ts +++ b/agents/container/storage.ts @@ -1,4 +1,4 @@ -import { execFileSync, spawnSync } from 'node:child_process'; +import { spawnSync } from 'node:child_process'; import { randomUUID } from 'node:crypto'; import { lstatSync, opendirSync, readlinkSync } from 'node:fs'; import { dirname, isAbsolute, join, relative } from 'node:path'; @@ -6,7 +6,8 @@ import type { TaskClone } from '../contract.ts'; import { assertTaskClone } from '../../git/clone.ts'; import { assertBuiltAgentImage } from './image.ts'; import { createOutcomeUnknown, DOCKER_ID } from '../client-outcome.ts'; -import { runDocker } from '../docker.ts'; +import { DockerError, pause, runDocker, type DockerOutcome } from '../docker.ts'; +import { MAXIMUM_TIMER_MS, runInProcessGroup, type ProcessGroup } from '../process-group.ts'; import { ALLOCATION_IN_USE, allocationListCommands, assertResourceOwner, claimAllocationId, hasOwnerLabels, ownerLabelArgs, releaseAllocationId, type ResourceOwner, UUID_V4 } from '../labels.ts'; @@ -48,6 +49,8 @@ const validLimit = (value: number, name: string) => { }; const createDeadline = (timeoutMs: number) => { validLimit(timeoutMs, 'timeoutMs'); + // Both variants share these steps, and the asynchronous one arms Node timers, which cannot wait longer than this. + if (timeoutMs > MAXIMUM_TIMER_MS) throw new Error(`timeoutMs must be at most ${MAXIMUM_TIMER_MS}.`); const deadline = performance.now() + timeoutMs; return () => { const value = Math.ceil(deadline - performance.now()); @@ -55,33 +58,95 @@ const createDeadline = (timeoutMs: number) => { return value; }; }; -const docker = (args: readonly string[], timeoutMs: number) => execFileSync('docker', [...args], { - encoding: 'utf8', timeout: timeoutMs, killSignal: 'SIGKILL', env: dockerEnvironment(), - stdio: ['ignore', 'pipe', 'pipe'], -}).trim(); -const absent = (result: ReturnType) => result.status !== 0 && !result.error - && /No such (?:object|container|volume)/i.test(`${result.stdout ?? ''}\n${result.stderr ?? ''}`); +/** One Docker call a storage step needs run, a wait, or a point where a long walk lets other work (and an abort) in. */ +type StorageStep = { readonly args: readonly string[]; readonly timeoutMs: number; readonly cancellable: boolean } + | { readonly sleepMs: number } | typeof PAUSE; +type Steps = Generator; +const PAUSE = Symbol('pause'); +// Entries walked between pauses, so the asynchronous variant never holds the event loop for a whole checkout. +const PAUSE_EVERY = 1_000; +/** Ask the driver to run one Docker call. Allocation calls are cancellable; cleanup calls never are. */ +function* run(args: readonly string[], timeoutMs: number, cancellable: boolean): Steps { + return (yield { args, timeoutMs, cancellable })!; +} +/** Run one Docker call that must succeed; its failure throws a `DockerError` that `createOutcomeUnknown` reads. */ +function* must(args: readonly string[], timeoutMs: number): Steps { + const outcome = yield* run(args, timeoutMs, true); + if (outcome.status !== 0) throw new DockerError(args, outcome); + return outcome.stdout.trim(); +} + +// The blocking driver, for the synchronous API. +const sleep = (ms: number) => Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms); +function runSteps(steps: Steps): T { + let next = steps.next(); + while (!next.done) { + const step = next.value; + if (step === PAUSE) next = steps.next(); + else if ('sleepMs' in step) { sleep(step.sleepMs); next = steps.next(); } + else { + const result = spawnSync('docker', [...step.args], { encoding: 'utf8', timeout: step.timeoutMs, killSignal: 'SIGKILL', + env: dockerEnvironment(), stdio: ['ignore', 'pipe', 'pipe'], maxBuffer: 16 * 1024 * 1024 }); + const error = result.error ?? (result.status === null + ? new Error(`docker ${step.args[0] ?? ''} was killed by ${result.signal}.`) : undefined); + next = steps.next({ status: error ? null : result.status, stdout: String(result.stdout ?? ''), + stderr: String(result.stderr ?? ''), ...(error ? { error } : {}) }); + } + } + return next.value; +} + +/** How the asynchronous variant runs its Docker calls. */ +export interface PreparationOptions { + /** Aborting stops the running allocation call's process group; cleanup of anything created still runs, uncancelled. */ + readonly signal?: AbortSignal; + /** Called in the same turn as each Docker spawn with its process group, so the caller can record it durably. */ + readonly onProcessGroup?: (group: ProcessGroup) => void; +} +// The non-blocking driver: every Docker call runs in its own process group, and the promise settles only after it has +// exited. An abort reaches the generator as an error at its next cancellable call or pause, so its own cleanup runs. +async function runStepsAsync(steps: Steps, options: PreparationOptions): Promise { + const aborted = () => Object.assign(new Error('Task storage allocation was cancelled.'), + { name: 'AbortError', code: 'ABORT_ERR' }); + let next = steps.next(); + while (!next.done) { + const step = next.value; + if (step === PAUSE) { + await new Promise(resolve => setImmediate(resolve)); + next = options.signal?.aborted ? steps.throw(aborted()) : steps.next(); + } else if ('sleepMs' in step) { + await pause(step.sleepMs); + next = steps.next(); + } else { + next = steps.next(await runInProcessGroup('docker', step.args, { env: dockerEnvironment(), + timeoutMs: step.timeoutMs, signal: step.cancellable ? options.signal : undefined, + onProcessGroup: options.onProcessGroup })); + } + } + return next.value; +} + +const absent = (result: DockerOutcome) => result.status !== 0 && result.status !== null && !result.error + && /No such (?:object|container|volume)/i.test(`${result.stdout}\n${result.stderr}`); /** How long an object whose create client was killed may still materialize in the daemon. */ const CREATE_SETTLE_MS = 10_000; -const sleep = (ms: number) => Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms); // Remove one storage object found by the name this allocation gave it, once it carries all three owner labels. A // container is then removed and confirmed by the ID just inspected, so a same-named replacement created after the // inspect is never touched. A volume's name is its only identity and Docker has no conditional remove, so a volume // replaced between the inspect and the remove could still be deleted; that needs someone to have deleted ours first. -const remove = (object: 'container' | 'volume', name: string, remaining: () => number, kind: string, - owner: ResourceOwner, settleBy = 0) => { - const inspectArgs = [object, 'inspect', name]; - let before: ReturnType; +// Cleanup is never cancelled: it runs to its own deadline so nothing is dropped. +function* remove(object: 'container' | 'volume', name: string, remaining: () => number, kind: string, + owner: ResourceOwner, settleBy = 0): Steps { + let before: DockerOutcome; for (;;) { - before = spawnSync('docker', [...inspectArgs], { encoding: 'utf8', timeout: remaining(), killSignal: 'SIGKILL', - env: dockerEnvironment(), stdio: ['ignore', 'pipe', 'pipe'] }); + before = yield* run([object, 'inspect', name], remaining(), false); if (before.status === 0) break; if (!absent(before)) throw new Error(`Failed to establish ownership of ${kind}.`); // A killed create may still land; only absence after the settle window counts. if (performance.now() >= settleBy) return; - sleep(250); + yield { sleepMs: 250 }; } - const inspected = JSON.parse(String(before.stdout || '[]'))[0] as + const inspected = JSON.parse(before.stdout || '[]')[0] as { Id?: string; Labels?: Record; Config?: { Labels?: Record } } | undefined; const labels = inspected?.Labels ?? inspected?.Config?.Labels; // All three owner labels must match: the allocation ID alone is caller-chosen and could be reused elsewhere. @@ -91,27 +156,26 @@ const remove = (object: 'container' | 'volume', name: string, remaining: () => n if (!inspected?.Id || !DOCKER_ID.test(inspected.Id)) throw new Error(`Failed to establish the identity of ${kind}.`); target = inspected.Id; } - const result = spawnSync('docker', object === 'container' ? ['rm', '--force', target] : ['volume', 'rm', '--force', target], - { encoding: 'utf8', timeout: remaining(), killSignal: 'SIGKILL', env: dockerEnvironment(), stdio: ['ignore', 'pipe', 'pipe'] }); + const result = yield* run(object === 'container' ? ['rm', '--force', target] : ['volume', 'rm', '--force', target], + remaining(), false); if (result.status === 0) return; - const inspect = spawnSync('docker', [object, 'inspect', target], { encoding: 'utf8', timeout: remaining(), - killSignal: 'SIGKILL', env: dockerEnvironment(), stdio: ['ignore', 'pipe', 'pipe'] }); - if (!absent(inspect)) throw new Error(`Failed to confirm removal of ${kind}.`); -}; -const cleanup = (containers: readonly string[], volumes: readonly string[], owner: ResourceOwner, - unsettled: ReadonlySet = new Set(), timeoutMs = 30_000) => { + if (!absent(yield* run([object, 'inspect', target], remaining(), false))) + throw new Error(`Failed to confirm removal of ${kind}.`); +} +function* cleanup(containers: readonly string[], volumes: readonly string[], owner: ResourceOwner, + unsettled: ReadonlySet = new Set(), timeoutMs = 30_000): Steps { const remaining = createDeadline(timeoutMs + (unsettled.size ? CREATE_SETTLE_MS : 0)), failures: unknown[] = []; const settleBy = (name: string) => unsettled.has(name) ? performance.now() + CREATE_SETTLE_MS : 0; for (const container of containers) { - try { remove('container', container, remaining, 'task container', owner, settleBy(container)); } + try { yield* remove('container', container, remaining, 'task container', owner, settleBy(container)); } catch (error) { failures.push(error); } } for (const volume of volumes) { - try { remove('volume', volume, remaining, 'task volume', owner, settleBy(volume)); } + try { yield* remove('volume', volume, remaining, 'task volume', owner, settleBy(volume)); } catch (error) { failures.push(error); } } if (failures.length) throw new AggregateError(failures, 'Task filesystem cleanup did not settle.'); -}; +} export function assertTaskFilesystems(filesystems: TaskFilesystems, clone?: TaskClone): void { const identity = allocations.get(filesystems); @@ -172,12 +236,14 @@ const linkStaysInside = (staging: string, link: string) => { * checkout (for example process environments that hold vendor credentials). Worktree links that stay inside, including * loops and not-yet-existing targets, are allowed. */ -const assertContainedLinks = (staging: string, remaining: () => number) => { +function* containedLinks(staging: string, remaining: () => number): Steps { const metadata = join(staging, '.git'), pending = [staging]; - let count = 0; + // `walked` counts every entry touched, including each one read from a directory, so one huge directory pauses too. + let count = 0, walked = 0; while (pending.length) { remaining(); count++; + if (++walked % PAUSE_EVERY === 0) yield PAUSE; const path = pending.pop()!, stat = lstatSync(path); if (stat.isSymbolicLink()) { const name = JSON.stringify(relative(staging, path)); @@ -195,30 +261,28 @@ const assertContainedLinks = (staging: string, remaining: () => number) => { if (count + pending.length >= LINK_INSPECTION_LIMIT) throw new Error('Repository checkout exceeds the link inspection limit.'); pending.push(join(path, entry.name)); + if (++walked % PAUSE_EVERY === 0) yield PAUSE; } } finally { directory.closeSync(); } } -}; +} // Fails closed: an unanswered list cannot prove the ID is unused. Throws when more than `expected` objects carry it. -const assertAllocationObjects = (allocationId: string, expected: number, remaining: () => number) => { +function* allocationObjects(allocationId: string, expected: number, remaining: () => number): Steps { let found = 0; for (const command of allocationListCommands(allocationId)) { - const result = spawnSync('docker', [...command], { encoding: 'utf8', timeout: remaining(), killSignal: 'SIGKILL', - env: dockerEnvironment(), stdio: ['ignore', 'pipe', 'pipe'] }); + const result = yield* run(command, remaining(), true); + // A client that was stopped (an abort or its deadline) or never started says why; a daemon answer does not. + if (result.status === null) throw new DockerError(command, result); if (result.status !== 0) throw new Error('Could not confirm that allocationId is unused.'); - found += String(result.stdout).split('\n').filter(line => line.trim()).length; + found += result.stdout.split('\n').filter(line => line.trim()).length; if (found > expected) throw new Error(ALLOCATION_IN_USE); } -}; +} -/** - * Copy a staging clone into bounded, engine-owned task storage and keep it mounted. `owner` is recorded by the caller - * before this call: every volume and container is labelled with its runner, attempt and allocation, so recovery after - * a crash finds exactly this storage. An allocation ID is used once; a reused ID is refused before anything is created. - */ -export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimits, - imageId: string, owner: ResourceOwner, timeoutMs = 60_000): TaskFilesystems { +/** The allocation, as a sequence of Docker calls. Both variants run these same steps; only how each call runs differs. */ +function* allocation(clone: TaskClone, limits: TaskStorageLimits, imageId: string, owner: ResourceOwner, + timeoutMs: number): Steps { owner = assertResourceOwner(owner); const labels = ownerLabelArgs(owner); for (const [name, value] of Object.entries(limits)) validLimit(value, name); @@ -227,10 +291,10 @@ export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimi const staging = assertTaskClone(clone), remaining = createDeadline(timeoutMs); if (/[\n,]/.test(staging)) throw new Error('Staging path cannot be represented as a Docker mount.'); if (!lstatSync(`${staging}/.git`).isDirectory()) throw new Error('Staging clone must contain standalone Git metadata.'); - assertContainedLinks(staging, remaining); + yield* containedLinks(staging, remaining); const allocationId = claimAllocationId(owner.allocationId); // Nothing created yet: a reused ID (still labelling objects from any earlier process) is refused here. - try { assertAllocationObjects(allocationId, 0, remaining); } + try { yield* allocationObjects(allocationId, 0, remaining); } catch (error) { releaseAllocationId(allocationId); throw error; } // Live from before the first create: a failed allocation whose cleanup does not settle still owns what it made. liveAllocations.set(allocationId, owner.runnerOwner); @@ -240,13 +304,13 @@ export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimi // create the daemon refused made nothing, and its name may belong to someone else. const made = new Set(), unsettled = new Set(); const mine = (names: readonly string[]) => names.filter(name => made.has(name) || unsettled.has(name)); - // Run one allocation step; a client killed by its deadline leaves the daemon outcome for `name` unknown. A pure - // create the daemon refused made nothing. A `docker run` is different: the daemon can create the container and then - // fail to start it, so after any failure of a run whose client started, `name` may exist and cleanup checks it. - const allocate = (name: string, args: readonly string[]) => { - const timeout = remaining(); + // Run one allocation step; a client killed by its deadline or an abort leaves the daemon outcome for `name` unknown. + // A pure create the daemon refused made nothing. A `docker run` is different: the daemon can create the container + // and then fail to start it, so after any failure of a run whose client started, `name` may exist and cleanup + // checks it. + function* allocate(name: string, args: readonly string[]): Steps { try { - const output = docker(args, timeout); + const output = yield* must(args, remaining()); made.add(name); return output; } catch (error) { @@ -254,16 +318,16 @@ export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimi else if (args[0] === 'run' && typeof (error as { status?: unknown }).status === 'number') made.add(name); throw error; } - }; + } try { for (const [kind, name, bytes, inodes] of [['work', workVolume, limits.workBytes, limits.workInodes], ['metadata', metadataVolume, limits.metadataBytes, limits.metadataInodes]] as const) { - allocate(name, ['volume', 'create', '--driver', 'local', '--opt', 'type=tmpfs', '--opt', 'device=tmpfs', + yield* allocate(name, ['volume', 'create', '--driver', 'local', '--opt', 'type=tmpfs', '--opt', 'device=tmpfs', '--opt', `o=size=${bytes},nr_inodes=${inodes},uid=10001,gid=10001,mode=0755,nosuid,nodev`, '--label', `io.codeboost.task-storage=${kind}`, ...labels, name]); // The check above and this create are not atomic across processes. Once our first object exists, it must be // the only one carrying the ID: of two racing allocations, the later check sees both and backs out. - if (kind === 'work') assertAllocationObjects(allocationId, 1, remaining); + if (kind === 'work') yield* allocationObjects(allocationId, 1, remaining); } // Copy metadata straight to its own volume so the work allocation never holds both at once. const seed = ['set -eu', @@ -272,14 +336,14 @@ export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimi 'cp -a --no-preserve=ownership,timestamps /run/codeboost-staging/.git/. /metadata/', 'mkdir -p /work/.git', 'chown -R 10001:10001 /work /metadata'].join('; '); // Create and start separately: once the create returns, the keeper is ours by ID even if its start fails. - const keeperId = allocate(keeper, ['create', '--name', keeper, '--read-only', '--user', '10001:10001', '--network=none', + const keeperId = yield* allocate(keeper, ['create', '--name', keeper, '--read-only', '--user', '10001:10001', '--network=none', '--cap-drop=ALL', '--security-opt=no-new-privileges', '--security-opt=seccomp=builtin', '--runtime=runc', '--pids-limit=32', '--memory=128m', '--cpus=.25', '--mount', `type=volume,source=${workVolume},target=/work`, '--mount', `type=volume,source=${metadataVolume},target=/metadata`, '--label', 'io.codeboost.task-storage=keeper', ...labels, '--entrypoint', 'sleep', imageId, 'infinity']); if (!DOCKER_ID.test(keeperId)) throw new Error('Docker did not return the created keeper ID.'); - docker(['start', keeperId], remaining()); - allocate(seeder, ['run', '--rm', '--name', seeder, '--label', 'io.codeboost.task-storage=seeder', ...labels, + yield* must(['start', keeperId], remaining()); + yield* allocate(seeder, ['run', '--rm', '--name', seeder, '--label', 'io.codeboost.task-storage=seeder', ...labels, '--read-only', '--user', '0:0', '--network=none', '--cap-drop=ALL', '--cap-add=CHOWN', '--cap-add=DAC_OVERRIDE', '--cap-add=FOWNER', '--security-opt=no-new-privileges', '--security-opt=seccomp=builtin', '--runtime=runc', '--pids-limit=32', '--memory=128m', '--cpus=.25', '--mount', `type=bind,source=${staging},target=/run/codeboost-staging,readonly`, @@ -294,7 +358,7 @@ export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimi releaseAllocationId(allocationId); return filesystems; } catch (error) { - try { cleanup(mine([seeder, keeper]), mine([metadataVolume, workVolume]), owner, unsettled); } + try { yield* cleanup(mine([seeder, keeper]), mine([metadataVolume, workVolume]), owner, unsettled); } // Still live: this process owns whatever cleanup could not remove, and reports it through the error. catch (cleanupError) { throw new AggregateError([error, cleanupError], 'Task allocation failed and cleanup did not settle.'); } liveAllocations.delete(allocationId); @@ -303,6 +367,38 @@ export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimi } } +/** + * Copy a staging clone into bounded, engine-owned task storage and keep it mounted. `owner` is recorded by the caller + * before this call: every volume and container is labelled with its runner, attempt and allocation, so recovery after + * a crash finds exactly this storage. An allocation ID is used once; a reused ID is refused before anything is created. + * Blocks the event loop while it runs; the runner uses `prepareTaskFilesystemsAsync`. + */ +export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimits, + imageId: string, owner: ResourceOwner, timeoutMs = 60_000): TaskFilesystems { + return runSteps(allocation(clone, limits, imageId, owner, timeoutMs)); +} + +/** + * `prepareTaskFilesystems` without blocking the event loop. Each Docker call runs in its own process group, reported + * through `onProcessGroup`. An abort stops the running call's group (SIGTERM, then SIGKILL after 5 s); the allocation + * then removes what it created, uncancelled and within its own cleanup deadline, and the promise rejects only after + * every group has exited. A create cut short by the abort counts as possibly created, as for a killed client. + */ +export async function prepareTaskFilesystemsAsync(clone: TaskClone, limits: TaskStorageLimits, imageId: string, + owner: ResourceOwner, options: PreparationOptions & { readonly timeoutMs?: number } = {}): Promise { + if (options.signal?.aborted) + throw Object.assign(new Error('Task storage allocation was cancelled.'), { name: 'AbortError', code: 'ABORT_ERR' }); + try { return await runStepsAsync(allocation(clone, limits, imageId, owner, options.timeoutMs ?? 60_000), options); } + catch (error) { + // However the abort surfaced, it is a cancel. A cleanup that did not settle stays an AggregateError: the caller + // must still learn what may be left. + if (options.signal?.aborted && !(error instanceof AggregateError) && (error as Error | undefined)?.name !== 'AbortError') + throw Object.assign(new Error('Task storage allocation was cancelled.', { cause: error }), + { name: 'AbortError', code: 'ABORT_ERR' }); + throw error; + } +} + /** * Task storage that `recoverLeftovers` found after a restart, when the `TaskFilesystems` value that allocated it is * gone. A crash during allocation can leave any subset of the parts, so each is optional. D issues a handle only after @@ -355,7 +451,7 @@ export async function adoptRecoveredTaskStorage(owner: ResourceOwner, parts: Tas const target = object === 'container' && keeperId ? keeperId : name; const inspect = await runDocker([object, 'inspect', target], { timeoutMs: remaining() }); if (inspect.status !== 0) { - if (absent(inspect as unknown as ReturnType)) + if (absent(inspect)) throw new RecoveredStorageRejected(`Recovered task ${kind} is gone.`); throw new Error(`Recovered task ${kind} could not be inspected.`); } @@ -392,15 +488,15 @@ export function removeTaskFilesystems(filesystems: TaskFilesystems | RecoveredTa const recovered = recoveredStorage.get(filesystems as RecoveredTaskStorage); if (recovered) { const handle = filesystems as RecoveredTaskStorage; - cleanup(recovered.keeperId ? [recovered.keeperId] : [], - [handle.metadataVolume, handle.workVolume].filter((name): name is string => name !== undefined), recovered.owner); + runSteps(cleanup(recovered.keeperId ? [recovered.keeperId] : [], + [handle.metadataVolume, handle.workVolume].filter((name): name is string => name !== undefined), recovered.owner)); recoveredStorage.delete(handle); return; } const allocated = filesystems as TaskFilesystems; assertTaskFilesystems(allocated); const owner = taskFilesystemOwner(allocated); - cleanup([allocated.keeper], [allocated.metadataVolume, allocated.workVolume], owner); + runSteps(cleanup([allocated.keeper], [allocated.metadataVolume, allocated.workVolume], owner)); allocations.delete(allocated); liveAllocations.delete(owner.allocationId); } diff --git a/agents/process-group.ts b/agents/process-group.ts new file mode 100644 index 00000000..db7bad60 --- /dev/null +++ b/agents/process-group.ts @@ -0,0 +1,193 @@ +import { spawn } from 'node:child_process'; +import { NOT_STARTED, type DockerOutcome } from './docker.ts'; + +/** A subprocess's process group, reported as soon as it exists so the caller can record it durably. */ +export interface ProcessGroup { + /** The group ID, which is the leader's PID. */ + readonly pgid: number; + /** `Date.now()` right after the spawn; with the ID, it tells this group from a later one that reuses the ID. */ + readonly startedAt: number; +} +export interface ProcessGroupOptions { + readonly env: NodeJS.ProcessEnv; + readonly timeoutMs: number; + readonly cwd?: string; + /** Aborting stops the group (SIGTERM, then SIGKILL after the grace period); the call still settles only after exit. */ + readonly signal?: AbortSignal; + /** Called synchronously, in the same turn as the spawn, with the new group. */ + readonly onProcessGroup?: (group: ProcessGroup) => void; + /** How long a stopped group gets between SIGTERM and SIGKILL. Default 5 s. */ + readonly graceMs?: number; + /** Bytes kept of each of stdout and stderr; more stops the group. Default 16 MiB. */ + readonly maxBuffer?: number; +} + +const DEFAULT_GRACE_MS = 5_000; +/** The longest delay a Node timer honours; `setTimeout` treats anything longer as 1 ms. */ +export const MAXIMUM_TIMER_MS = 2 ** 31 - 1; +const DEFAULT_MAX_BUFFER = 16 * 1024 * 1024; +// After the leader exits, how long to keep killing and polling other members of its group before giving up. +const DRAIN_LIMIT_MS = 10_000; +// Once the group is gone, how long to wait for stdout and stderr to reach end of file. Only a process outside the group +// (one that called setsid but kept the inherited pipe) can hold them open after that, and it must not hold this call. +const STDIO_CLOSE_MS = 1_000; +const pause = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); +const notStarted = (message: string): DockerOutcome => + ({ status: null, stdout: '', stderr: '', error: Object.assign(new Error(message), { code: NOT_STARTED }) }); + +// Signal every process in the group; false once the group no longer exists. Never throws: it runs in timers and +// listeners, where a throw would crash the runner. Any failure other than ESRCH (such as EPERM, for a member that can +// no longer be signalled) counts as still alive, so draining gives up at its limit and reports it. +const signalGroup = (pgid: number, signal: NodeJS.Signals | 0) => { + try { process.kill(-pgid, signal); return true; } + catch (error) { return (error as NodeJS.ErrnoException).code !== 'ESRCH'; } +}; + +// The leader can exit while other members of its group still run (for example a child it forked). A group ID stays +// taken while any member exists, so until the group is empty the ID cannot name someone else's group. +const drainGroup = async (pgid: number) => { + const giveUpAt = performance.now() + DRAIN_LIMIT_MS; + while (signalGroup(pgid, 0)) { + signalGroup(pgid, 'SIGKILL'); + if (performance.now() >= giveUpAt) return false; + await pause(20); + } + return true; +}; + +/** + * Run one command as the leader of a new process group. The group is reported through `onProcessGroup` before this + * returns control to the event loop. On abort or at the deadline the whole group gets SIGTERM, then SIGKILL after the + * grace period, so a child that ignores SIGTERM is still bounded. The promise settles only after the leader has + * exited and every other member of its group is gone. Never rejects: failures are in the outcome, shaped like + * `runDocker`'s, so `status` is a number only when the command ran to completion. + */ +export function runInProcessGroup(file: string, args: readonly string[], + options: ProcessGroupOptions): Promise { + if (!Number.isSafeInteger(options.timeoutMs) || options.timeoutMs < 1 || options.timeoutMs > MAXIMUM_TIMER_MS) + return Promise.resolve(notStarted(`Process deadline must be a positive integer of at most ${MAXIMUM_TIMER_MS} ms.`)); + if (options.signal?.aborted) return Promise.resolve(notStarted(`${file} was cancelled before it started.`)); + const graceMs = options.graceMs ?? DEFAULT_GRACE_MS, maxBuffer = options.maxBuffer ?? DEFAULT_MAX_BUFFER; + return new Promise(resolve => { + // `detached` makes the child the leader of a new process group (setsid), so signals to -pgid reach all it starts. + // spawn throws, rather than emitting 'error', for invalid arguments and for spawn failures other than ENOENT, + // EACCES, EAGAIN, EMFILE and ENFILE (for example ENOMEM or E2BIG). No child exists then; report it, never reject. + let child: ReturnType; + try { + child = spawn(file, [...args], { cwd: options.cwd, env: options.env, detached: true, + stdio: ['ignore', 'pipe', 'pipe'] }); + } catch (error) { + resolve({ status: null, stdout: '', stderr: '', error: error as Error }); + return; + } + const pgid = child.pid; + if (pgid === undefined) { + // The spawn itself failed (for example ENOENT); nothing started, and 'error' carries the cause. + child.once('error', error => resolve({ status: null, stdout: '', stderr: '', error })); + return; + } + const out: Buffer[] = [], err: Buffer[] = []; + let outBytes = 0, errBytes = 0, stopped: 'cancelled' | 'timeout' | 'output-limit' | 'unrecorded' | undefined; + let graceTimer: ReturnType | undefined; + // Set once the leader has exited: from then on the group is drained, not stopped, and its ID may be reused. + let exited = false; + const stop = (reason: NonNullable) => { + if (stopped || exited) return; + stopped = reason; + signalGroup(pgid, 'SIGTERM'); + graceTimer = setTimeout(() => signalGroup(pgid, 'SIGKILL'), graceMs); + }; + // Armed at the spawn, before the caller records the group: time spent recording counts against the deadline. + const deadline = setTimeout(() => stop('timeout'), options.timeoutMs); + // If the caller cannot record the group, the child must not outlive this call: it is killed at once below, and + // the call still settles only after the group has exited, with the caller's error. + let unrecorded: unknown; + try { options.onProcessGroup?.(Object.freeze({ pgid, startedAt: Date.now() })); } + catch (error) { unrecorded = error; } + // Output past the limit stops a running group; once the leader has exited it only drops the excess. Either way the + // output is incomplete, so the call settles as ENOBUFS, never as a success. + let truncated = false; + const collect = (chunks: Buffer[], add: (bytes: number) => number) => (chunk: Buffer) => { + if (add(chunk.length) > maxBuffer) { truncated = true; stop('output-limit'); } + else chunks.push(chunk); + }; + child.stdout!.on('data', collect(out, bytes => (outBytes += bytes))); + child.stderr!.on('data', collect(err, bytes => (errBytes += bytes))); + const onAbort = () => stop('cancelled'); + options.signal?.addEventListener('abort', onAbort, { once: true }); + if (unrecorded !== undefined) { stopped = 'unrecorded'; signalGroup(pgid, 'SIGKILL'); } + // An abort raised during onProcessGroup came before the listener existed, and an aborted signal never fires again. + else if (options.signal?.aborted) stop('cancelled'); + let spawnError: Error | undefined; + let pipesClosed = false; + const closed = new Promise(resolveClosed => child.once('close', () => { pipesClosed = true; resolveClosed(); })); + // Settle from the leader's exit, not from 'close': 'close' also waits for every holder of the pipes, which can + // include a process that left the group and so survives the group kill. + let finished = false; + const finish = (code: number | null, signal: NodeJS.Signals | null) => { + if (finished) return; + finished = true; + exited = true; + clearTimeout(deadline); + options.signal?.removeEventListener('abort', onAbort); + void drainGroup(pgid).then(async drained => { + // The group is empty (or given up on): a later SIGKILL could reach a new group that reuses the ID. + clearTimeout(graceTimer); + let stdioTimer: ReturnType | undefined; + await Promise.race([closed, new Promise(done => { stdioTimer = setTimeout(done, STDIO_CLOSE_MS); })]); + // Cleared so a finished call never keeps the process alive. + clearTimeout(stdioTimer); + // Timers run before I/O in each event-loop turn: after a long block the timer can win while the rest of the + // output is already waiting. One more turn lets that I/O (and the end of the pipes) be read first. + if (!pipesClosed) await new Promise(resolveTurn => setImmediate(resolveTurn)); + child.stdout!.destroy(); + child.stderr!.destroy(); + const stdout = Buffer.concat(out).toString('utf8'), stderr = Buffer.concat(err).toString('utf8'); + if (!drained) { + resolve({ status: null, stdout, stderr, error: Object.assign( + new Error(`${file} left processes in its group that did not exit after SIGKILL.`), { code: 'EGROUPALIVE' }) }); + return; + } + if (!pipesClosed) { + // Only a process outside the group can still hold the pipes; the output may be incomplete, so this is not a + // successful result even if the leader exited 0. + resolve({ status: null, stdout, stderr, error: Object.assign(new Error( + `${file} exited, but a process outside its group still holds its output; the output may be incomplete.`), + { code: 'ESTDIOHELD' }) }); + return; + } + if (stopped === 'unrecorded') { + resolve({ status: null, stdout, stderr, error: Object.assign(new Error( + `${file} was stopped because its process group could not be recorded.`, { cause: unrecorded }), + { code: 'EUNRECORDED' }) }); + return; + } + if (stopped === 'cancelled') { + resolve({ status: null, stdout, stderr, error: Object.assign(new Error(`${file} ${args[0] ?? ''} was cancelled.`), + { name: 'AbortError', code: 'ABORT_ERR' }) }); + return; + } + if (truncated && !stopped) { + resolve({ status: null, stdout, stderr, error: Object.assign( + new Error(`${file} ${args[0] ?? ''} exceeded its output limit.`), { code: 'ENOBUFS' }) }); + return; + } + if (stopped) { + resolve({ status: null, stdout, stderr, error: Object.assign(new Error(stopped === 'timeout' + ? `${file} ${args[0] ?? ''} ETIMEDOUT after ${options.timeoutMs} ms.` + : `${file} ${args[0] ?? ''} exceeded its output limit.`), { code: stopped === 'timeout' ? 'ETIMEDOUT' : 'ENOBUFS' }) }); + return; + } + if (spawnError || code === null) { + resolve({ status: null, stdout, stderr, + error: spawnError ?? Object.assign(new Error(`${file} ${args[0] ?? ''} was killed by ${signal}.`), { signal }) }); + return; + } + resolve({ status: code, stdout, stderr }); + }); + }; + child.once('exit', finish); + // A spawn that fails after the child had a PID emits 'error' and may never emit 'exit'. + child.once('error', error => { spawnError = error; finish(null, null); }); + }); +} diff --git a/docs/implementation/agent-isolation.md b/docs/implementation/agent-isolation.md index 0cfc7752..3186b6b7 100644 --- a/docs/implementation/agent-isolation.md +++ b/docs/implementation/agent-isolation.md @@ -57,8 +57,10 @@ metadata, read-only isolation and capacity probes had this defect. Use only these entry points to run an agent: -1. `createTaskClone` creates a committed, standalone staging clone. -2. `prepareTaskFilesystems` copies that clone into bounded task storage, labelled with the owner you pass: your +1. `createTaskClone` creates a committed, standalone staging clone. The runner uses `createTaskCloneAsync`, which + takes an `AbortSignal` and reports each Git process group through `onProcessGroup` (#51 item 5). +2. `prepareTaskFilesystems` (or `prepareTaskFilesystemsAsync`, which takes `signal` and `onProcessGroup` in the same + way) copies that clone into bounded task storage, labelled with the owner you pass: your runner token, the attempt ID, and an allocation ID (a lowercase UUID v4) you record first. Use each allocation ID once: before creating anything, D refuses an ID that another allocation in this process holds, or that any container, volume or network still carries (so a reuse after a restart is caught too). That check and the first diff --git a/docs/implementation/runner-lifecycle.md b/docs/implementation/runner-lifecycle.md index 70a44270..8dbcf4bb 100644 --- a/docs/implementation/runner-lifecycle.md +++ b/docs/implementation/runner-lifecycle.md @@ -16,7 +16,7 @@ - the order of steps when the program shuts down; - the feedback events that the learning lane (J) will read. -**When implementation starts.** The plan starts F implementation after D5 merges, and D5 merged as #50. F1 code can start now. The attempt lifecycle itself uses only the D interface already on `main` (`agents/contract.ts`), which D4 (#47) did not change. Preparation, shutdown and startup recovery need D changes. Bounded settlement (#51 item 1) and asynchronous launch (#51 item 2) are done. Owner labels (#51 item 3) are done: `runnerOwner` on `InvocationInput`, and an owner on every allocator that runs before an invocation. Scoped recovery (#51 item 4) is done: `recoverLeftovers` (`agents/recovery.ts`) returns authenticated storage handles that `removeTaskFilesystems` accepts. This does **not** exist yet: asynchronous, abortable versions of both preparation helpers, `createTaskClone` and `prepareTaskFilesystems` (#51 item 5), listed as a prerequisite under "Shutdown". The F1 implementation may start against fakes, but it must not merge until #51's pre-F1 items land. +**When implementation starts.** The plan starts F implementation after D5 merges, and D5 merged as #50. F1 code can start now. The attempt lifecycle itself uses only the D interface already on `main` (`agents/contract.ts`), which D4 (#47) did not change. Preparation, shutdown and startup recovery need D changes. Bounded settlement (#51 item 1) and asynchronous launch (#51 item 2) are done. Owner labels (#51 item 3) are done: `runnerOwner` on `InvocationInput`, and an owner on every allocator that runs before an invocation. Scoped recovery (#51 item 4) is done: `recoverLeftovers` (`agents/recovery.ts`) returns authenticated storage handles that `removeTaskFilesystems` accepts. Abortable preparation (#51 item 5) is done: `createTaskCloneAsync` (`git/clone.ts`) and `prepareTaskFilesystemsAsync` (`agents/container/storage.ts`) take an `AbortSignal`, run every subprocess in its own process group, report each group through `onProcessGroup` in the same turn as the spawn, and settle only after it has exited. All of #51's pre-F1 items have landed. ## Summary @@ -260,7 +260,7 @@ This belongs to D (D5 or a D follow-up), not F. It changes `agents/contract.ts`, **Also before F1 merges: asynchronous launch.** Done in #51 item 2 (see "Launch setup does not block the server"). -**Third prerequisite before F1 merges: abortable preparation helpers.** Both preparation helpers are synchronous today. `createTaskClone` runs git with `execFileSync`, and `prepareTaskFilesystems` runs `docker` with `execFileSync` and `spawnSync` (`agents/container/storage.ts`). Either would block the event loop, and cancel or shutdown could not abort it. D must add asynchronous variants of both that take an `AbortSignal`, spawn their subprocesses in their own process group, expose the group ID, and settle only after those processes have exited. Storage allocation must also label its volumes with the runner owner and attempt ID before it returns (see "Recovered storage handles"), so that storage allocated just before a crash is found by recovery. +**Third prerequisite before F1 merges: abortable preparation helpers.** Done in #51 item 5: `createTaskCloneAsync` and `prepareTaskFilesystemsAsync` take `signal` and `onProcessGroup`. Each subprocess runs as the leader of a new process group (`agents/process-group.ts`); `onProcessGroup` receives `{ pgid, startedAt }` synchronously after the spawn, so F can save it as `preparation_pgid` and `preparation_started_at` before anything else runs. Every Git or Docker call is a separate group, so F updates the saved group on each call. On abort or at the deadline the group gets `SIGTERM`, then `SIGKILL` after 5 seconds; the call settles only after the leader and every other member of its group have exited. Two bounds apply: a member still alive 10 seconds after the leader exits (repeated `SIGKILL` included) settles the call with `EGROUPALIVE`, which means the group may still be running and F must treat the outcome as unknown; and an output pipe still open 1 second after the group is gone (held by a process that left the group) is closed, so settlement never waits on it, and the call settles with `ESTDIOHELD`: its output may be incomplete, so it is never reported as a success. A spawn that fails outright (for example `ENOMEM`) settles as not started, and the helper never rejects. An aborted clone removes its partial directory; an aborted storage allocation removes what it created, with cleanup that is not cancelled. The synchronous helpers remain for existing callers and share the same steps. What follows is the original requirement. Both preparation helpers were synchronous. `createTaskClone` runs git with `execFileSync`, and `prepareTaskFilesystems` runs `docker` with `execFileSync` and `spawnSync` (`agents/container/storage.ts`). Either would block the event loop, and cancel or shutdown could not abort it. D must add asynchronous variants of both that take an `AbortSignal`, spawn their subprocesses in their own process group, expose the group ID, and settle only after those processes have exited. Storage allocation must also label its volumes with the runner owner and attempt ID before it returns (see "Recovered storage handles"), so that storage allocated just before a crash is found by recovery. **Prerequisite before F2's writable attempts: a bounded task-volume export.** D must add a bounded diff export over the task volumes (see "Partial output and interrupted rebases"). F1 does not need it: F1 runs only read-only attempts, and their task storage is removed without an export. diff --git a/git/clone.ts b/git/clone.ts index edece0c2..39877c4d 100644 --- a/git/clone.ts +++ b/git/clone.ts @@ -3,6 +3,7 @@ import { randomUUID } from 'node:crypto'; import { lstatSync, mkdtempSync, opendirSync, realpathSync, rmSync } from 'node:fs'; import { isAbsolute, join, relative, resolve } from 'node:path'; import type { TaskClone } from '../agents/contract.ts'; +import { runInProcessGroup, type ProcessGroup } from '../agents/process-group.ts'; interface DirectoryIdentity { readonly dev: number; readonly ino: number } interface CloneIdentity { readonly directory: string; readonly root: DirectoryIdentity; readonly metadata: DirectoryIdentity } @@ -25,15 +26,29 @@ export function assertTaskClone(clone: TaskClone): string { return identity.directory; } +/** One Git call the clone steps need run, or a point where a long file walk lets other work (and an abort) in. */ +type CloneStep = { readonly cwd: string; readonly args: readonly string[]; readonly timeoutMs: number } | typeof PAUSE; +const PAUSE = Symbol('pause'); +/** What a Git call returned: `error` is set whenever it did not exit 0. */ +interface GitOutcome { readonly status: number | null; readonly stdout: string; readonly stderr: string; readonly error?: Error } +// Entries walked between pauses, so the asynchronous variant never holds the event loop for a whole object store. +const PAUSE_EVERY = 1_000; +const GIT_OPTIONS = ['--no-pager', '--no-replace-objects', '-c', 'core.hooksPath=/dev/null', '-c', 'init.templateDir=', + '-c', 'protocol.allow=never', '-c', 'submodule.recurse=false']; +// Deliberately do not inherit Git variables or credential/config environment. +const gitEnvironment = () => ({ PATH: process.env.PATH, GIT_CONFIG_NOSYSTEM: '1', GIT_CONFIG_GLOBAL: '/dev/null', + GIT_TERMINAL_PROMPT: '0', GIT_NO_LAZY_FETCH: '1', GIT_GRAFT_FILE: '/dev/null' }); + +export interface CloneOptions { + readonly source: string; readonly parent: string; readonly taskId: string; readonly head: string; + readonly timeoutMs?: number; +} + /** - * Prepare an independent committed snapshot. This is trusted staging, not the - * writable execution filesystem: D2 must reserve bounded storage and separate - * metadata before mounting it. Source must stay quiescent during this operation. - * No hooks, filters from user config, credentials, submodules or network access. + * The clone, as a sequence of Git calls. Both variants run these same steps; only how each call runs differs. Every + * failure after the clone directory exists, an abort included, removes it before the error leaves. */ -export function createTaskClone(options: { - source: string; parent: string; taskId: string; head: string; timeoutMs?: number; -}): TaskClone { +function* cloneSteps(options: CloneOptions): Generator { if (!options.taskId || options.taskId.includes('\0')) throw new Error('Task identity is required.'); if (!/^(?:[a-f0-9]{40}|[a-f0-9]{64})$/.test(options.head)) throw new Error('A full committed head is required.'); const timeout = options.timeoutMs ?? 30_000; @@ -50,29 +65,25 @@ export function createTaskClone(options: { return rel === '' || (!isAbsolute(rel) && rel !== '..' && !rel.startsWith('../')); }; if (within(source, parent)) throw new Error('Task storage must be outside the source repository.'); - // Deliberately do not inherit Git variables or credential/config environment. - const env = { PATH: process.env.PATH, GIT_CONFIG_NOSYSTEM: '1', GIT_CONFIG_GLOBAL: '/dev/null', - GIT_TERMINAL_PROMPT: '0', GIT_NO_LAZY_FETCH: '1', GIT_GRAFT_FILE: '/dev/null' }; - const run = (cwd: string, ...args: string[]) => { - const result = execFileSync('git', [ - '--no-pager', '--no-replace-objects', '-c', 'core.hooksPath=/dev/null', '-c', 'init.templateDir=', - '-c', 'protocol.allow=never', '-c', 'submodule.recurse=false', ...args, - ], { cwd, env, timeout: remaining(), killSignal: 'SIGKILL', maxBuffer: 1024 * 1024, - stdio: ['ignore', 'pipe', 'pipe'] }); + function* run(cwd: string, ...args: string[]): Generator { + const outcome = yield { cwd, args, timeoutMs: remaining() }; + if (!outcome || outcome.status !== 0) throw outcome?.error ?? new Error(`git ${args[0] ?? ''} failed.`); remaining(); - return result.toString().trim(); - }; - const common = realpathSync(resolve(source, run(source, 'rev-parse', '--git-common-dir'))); + return outcome.stdout.trim(); + } + const common = realpathSync(resolve(source, yield* run(source, 'rev-parse', '--git-common-dir'))); if (within(common, parent)) throw new Error('Task storage must be outside source metadata.'); - function audit(metadata: string, independent: boolean) { + function* audit(metadata: string, independent: boolean): Generator { for (const name of ['shallow', 'info/grafts', 'objects/info/alternates', 'objects/info/http-alternates']) { if (lstatSync(join(metadata, name), { throwIfNoEntry: false })) throw new Error(`Unsupported Git storage: ${name}`); } const pending = [join(metadata, 'objects')]; - let count = 0; + // `walked` counts every entry touched, including each one read from a directory, so one huge directory pauses too. + let count = 0, walked = 0; while (pending.length) { remaining(); if (++count > 100_000) throw new Error('Object storage exceeds inspection limit.'); + if (++walked % PAUSE_EVERY === 0) yield PAUSE; const path = pending.pop()!, stat = lstatSync(path); if (stat.isSymbolicLink() || (!stat.isDirectory() && !stat.isFile())) throw new Error('Unsupported object entry.'); if (independent && stat.isFile() && stat.nlink !== 1) throw new Error('Task objects must not be hard-linked.'); @@ -83,23 +94,27 @@ export function createTaskClone(options: { remaining(); if (count + pending.length >= 100_000) throw new Error('Object storage exceeds inspection limit.'); pending.push(join(path, entry.name)); + if (++walked % PAUSE_EVERY === 0) yield PAUSE; } } finally { directory.closeSync(); } } } } - audit(common, false); - if (run(source, 'for-each-ref', '--format=%(refname)', 'refs/replace')) throw new Error('Replacement objects are unsupported.'); - if (run(source, 'rev-parse', '--verify', `${options.head}^{commit}`) !== options.head) throw new Error('Head is not a commit.'); + yield* audit(common, false); + if (yield* run(source, 'for-each-ref', '--format=%(refname)', 'refs/replace')) + throw new Error('Replacement objects are unsupported.'); + if ((yield* run(source, 'rev-parse', '--verify', `${options.head}^{commit}`)) !== options.head) + throw new Error('Head is not a commit.'); const directory = mkdtempSync(join(parent, 'codeboost-task-')); try { - run(parent, '-c', 'protocol.file.allow=always', 'clone', '--local', '--no-hardlinks', '--no-checkout', '--', source, directory); + yield* run(parent, '-c', 'protocol.file.allow=always', 'clone', '--local', '--no-hardlinks', '--no-checkout', '--', + source, directory); const metadata = join(directory, '.git'); if (!lstatSync(metadata).isDirectory()) throw new Error('Task requires standalone Git metadata.'); - audit(metadata, true); - run(directory, 'remote', 'remove', 'origin'); - run(directory, 'checkout', '--detach', options.head); - if (run(directory, 'rev-parse', 'HEAD') !== options.head) throw new Error('Task head changed during clone.'); + yield* audit(metadata, true); + yield* run(directory, 'remote', 'remove', 'origin'); + yield* run(directory, 'checkout', '--detach', options.head); + if ((yield* run(directory, 'rev-parse', 'HEAD')) !== options.head) throw new Error('Task head changed during clone.'); const clone = Object.freeze({ id: randomUUID(), taskId: options.taskId, directory, head: options.head }); const root = directoryIdentity(directory), cloneMetadata = directoryIdentity(metadata); if (!root || !cloneMetadata) throw new Error('Task clone directory is not a real directory.'); @@ -110,3 +125,74 @@ export function createTaskClone(options: { throw error; } } + +/** + * Prepare an independent committed snapshot. This is trusted staging, not the + * writable execution filesystem: D2 must reserve bounded storage and separate + * metadata before mounting it. Source must stay quiescent during this operation. + * No hooks, filters from user config, credentials, submodules or network access. + * Blocks the event loop while it runs; the runner uses `createTaskCloneAsync`. + */ +export function createTaskClone(options: CloneOptions): TaskClone { + const env = gitEnvironment(), steps = cloneSteps(options); + let next = steps.next(); + while (!next.done) { + const step = next.value; + if (step === PAUSE) { next = steps.next(); continue; } + let outcome: GitOutcome; + try { + const stdout = execFileSync('git', [...GIT_OPTIONS, ...step.args], { cwd: step.cwd, env, timeout: step.timeoutMs, + killSignal: 'SIGKILL', maxBuffer: 1024 * 1024, stdio: ['ignore', 'pipe', 'pipe'] }); + outcome = { status: 0, stdout: stdout.toString(), stderr: '' }; + } catch (error) { + const failed = error as { status?: number | null; stdout?: Buffer; stderr?: Buffer }; + outcome = { status: typeof failed.status === 'number' ? failed.status : null, + stdout: String(failed.stdout ?? ''), stderr: String(failed.stderr ?? ''), error: error as Error }; + } + next = steps.next(outcome); + } + return next.value; +} + +export interface AsyncCloneOptions extends CloneOptions { + /** Aborting stops the running Git call's process group and removes the partial clone before the promise rejects. */ + readonly signal?: AbortSignal; + /** Called in the same turn as each Git spawn with its process group, so the caller can record it durably. */ + readonly onProcessGroup?: (group: ProcessGroup) => void; +} + +/** + * `createTaskClone` without blocking the event loop. Each Git call runs in its own process group, reported through + * `onProcessGroup`; an abort sends the group SIGTERM, then SIGKILL after 5 s, and the promise settles only after the + * group has exited and the partial clone is removed. + */ +export async function createTaskCloneAsync(options: AsyncCloneOptions): Promise { + const signal = options.signal; + try { return await runCloneSteps(options); } + catch (error) { + // However the abort surfaced (during a call, at a pause, or as the next call refusing to start), it is a cancel. + if (signal?.aborted && (error as Error | undefined)?.name !== 'AbortError') + throw Object.assign(new Error('Task clone was cancelled.', { cause: error }), { name: 'AbortError', code: 'ABORT_ERR' }); + throw error; + } +} +async function runCloneSteps(options: AsyncCloneOptions): Promise { + const env = gitEnvironment(), steps = cloneSteps(options), signal = options.signal; + const aborted = () => Object.assign(new Error('Task clone was cancelled.'), { name: 'AbortError', code: 'ABORT_ERR' }); + if (signal?.aborted) throw aborted(); + let next = steps.next(); + while (!next.done) { + const step = next.value; + if (step === PAUSE) { + await new Promise(resolve => setImmediate(resolve)); + next = signal?.aborted ? steps.throw(aborted()) : steps.next(); + continue; + } + const outcome = await runInProcessGroup('git', [...GIT_OPTIONS, ...step.args], { cwd: step.cwd, env, + timeoutMs: step.timeoutMs, signal, onProcessGroup: options.onProcessGroup, maxBuffer: 1024 * 1024 }); + const cancelled = (outcome.error as NodeJS.ErrnoException | undefined)?.code === 'ABORT_ERR'; + next = steps.next(outcome.status === 0 ? outcome : { ...outcome, error: cancelled ? aborted() : outcome.error + ?? new Error(`git ${step.args[0] ?? ''} failed (exit ${outcome.status}): ${outcome.stderr.trim().slice(0, 512)}`) }); + } + return next.value; +} diff --git a/test/agent-clone-async.test.ts b/test/agent-clone-async.test.ts new file mode 100644 index 00000000..2ca95141 --- /dev/null +++ b/test/agent-clone-async.test.ts @@ -0,0 +1,108 @@ +import { execFileSync } from 'node:child_process'; +import { existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, describe, expect, it, vi } from 'vitest'; + +// Every Git call passes through; a test can act after one call settles and before the next is requested. +const calls = vi.hoisted(() => ({ made: [] as string[][], afterCall: undefined as undefined | ((args: readonly string[]) => void) })); +vi.mock('../agents/process-group.ts', async importOriginal => { + const actual = await importOriginal(); + return { ...actual, runInProcessGroup: async (...call: Parameters) => { + calls.made.push([...call[1]]); + const outcome = await actual.runInProcessGroup(...call); + calls.afterCall?.(call[1]); + return outcome; + } }; +}); +const { assertTaskClone, createTaskCloneAsync } = await import('../git/clone.ts'); +import type { ProcessGroup } from '../agents/process-group.ts'; + +// The asynchronous clone (#51 item 5): every Git call runs in its own process group, and an abort settles only after +// that group has exited and the partial clone is gone. +const roots: string[] = []; +const git = (cwd: string, ...args: string[]) => execFileSync('git', ['-c', 'core.hooksPath=/dev/null', ...args], + { cwd, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'] }).trim(); +function fixture() { + const root = mkdtempSync(join(tmpdir(), 'clone-async-')); roots.push(root); + const source = join(root, 'source'), parent = join(root, 'tasks'); + mkdirSync(source); mkdirSync(parent); + git(source, 'init'); git(source, 'config', 'user.name', 'Test'); git(source, 'config', 'user.email', 'test@example.com'); + writeFileSync(join(source, 'file.txt'), 'trusted\n'); git(source, 'add', '.'); git(source, 'commit', '-m', 'baseline'); + return { root, source, parent, head: git(source, 'rev-parse', 'HEAD'), taskId: 'task-1' }; +} +const groupAlive = (pgid: number) => { + try { process.kill(-pgid, 0); return true; } catch { return false; } +}; +/** Put a `git` first on PATH whose `clone` ignores SIGTERM, starts a background child, and never finishes. */ +const withStubbornClone = async (root: string, started: string, run: () => Promise) => { + const shim = join(root, 'bin'); mkdirSync(shim); + const realGit = execFileSync('sh', ['-c', 'command -v git'], { encoding: 'utf8' }).trim(); + writeFileSync(join(shim, 'git'), ['#!/bin/sh', + 'for arg in "$@"; do if [ "$arg" = clone ]; then', + ` trap '' TERM; sleep 60 & touch '${started}'; while :; do sleep 1; done`, + 'fi; done', + `exec '${realGit}' "$@"`].join('\n'), { mode: 0o755 }); + const path = process.env.PATH; + process.env.PATH = `${shim}:${path}`; + try { return await run(); } finally { process.env.PATH = path; } +}; +afterEach(() => { + calls.afterCall = undefined; calls.made = []; + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); +}); + +describe('asynchronous task clone', () => { + it('clones like the synchronous helper and reports each Git process group, all gone once it settles', async () => { + const input = fixture(), groups: ProcessGroup[] = []; + const clone = await createTaskCloneAsync({ ...input, onProcessGroup: group => { groups.push(group); } }); + expect(assertTaskClone(clone)).toBe(clone.directory); + expect(readFileSync(join(clone.directory, 'file.txt'), 'utf8')).toBe('trusted\n'); + expect(git(clone.directory, 'rev-parse', 'HEAD')).toBe(input.head); + expect(groups.length).toBeGreaterThanOrEqual(5); + for (const group of groups) { + expect(group.pgid).toBeGreaterThan(1); + expect(groupAlive(group.pgid)).toBe(false); + } + }); + + it('on abort, kills a clone that ignores SIGTERM, waits for its group, and removes the partial clone', async () => { + const input = fixture(), started = join(input.root, 'clone-started'), groups: ProcessGroup[] = []; + const controller = new AbortController(); + // The event loop stays free while Git runs: this keeps ticking until the clone settles. + let ticks = 0; + const ticker = setInterval(() => { ticks += 1; }, 50); + const waitForClone = setInterval(() => { if (existsSync(started)) controller.abort(); }, 20); + const began = performance.now(); + try { + const error = await withStubbornClone(input.root, started, () => + createTaskCloneAsync({ ...input, timeoutMs: 60_000, signal: controller.signal, + onProcessGroup: group => { groups.push(group); } }).then(() => undefined, caught => caught)); + expect((error as Error).message).toContain('cancelled'); + } finally { clearInterval(ticker); clearInterval(waitForClone); } + // SIGTERM was ignored, so settlement waited for the SIGKILL after the 5 s grace period. + expect(performance.now() - began).toBeGreaterThanOrEqual(5_000); + expect(ticks).toBeGreaterThan(50); + for (const group of groups) expect(groupAlive(group.pgid)).toBe(false); + expect(readdirSync(input.parent)).toEqual([]); + }, 30_000); + + it('reports an abort that lands between two Git calls as a cancel, and removes the partial clone', async () => { + const input = fixture(), controller = new AbortController(), groups: ProcessGroup[] = []; + // Aborted after `git clone` settles and before the next Git call is requested: that call never starts. + calls.afterCall = args => { if (args.includes('clone')) controller.abort(); }; + const error = await createTaskCloneAsync({ ...input, signal: controller.signal, + onProcessGroup: group => { groups.push(group); } }).then(() => undefined, caught => caught); + expect(error).toMatchObject({ name: 'AbortError', code: 'ABORT_ERR' }); + expect(calls.made.length - groups.length).toBe(1); + expect(readdirSync(input.parent)).toEqual([]); + }); + + it('spawns nothing and creates nothing when the signal is already aborted', async () => { + const input = fixture(), groups: ProcessGroup[] = []; + await expect(createTaskCloneAsync({ ...input, signal: AbortSignal.abort(), + onProcessGroup: group => { groups.push(group); } })).rejects.toThrow('cancelled'); + expect(groups).toEqual([]); + expect(readdirSync(input.parent)).toEqual([]); + }); +}); diff --git a/test/agent-container.test.ts b/test/agent-container.test.ts index f6c21bac..cff086c4 100644 --- a/test/agent-container.test.ts +++ b/test/agent-container.test.ts @@ -12,7 +12,7 @@ import { createValidatedContainer, disposeValidatedContainer, prepareTaskFilesys startValidatedContainer, hasExactOptions, validateContainer } from '../agents/container/run.ts'; import { createTaskClone } from '../git/clone.ts'; import { hasOwnerLabels } from '../agents/labels.ts'; -import { isRecoveredTaskStorage, taskFilesystemOwner } from '../agents/container/storage.ts'; +import { isRecoveredTaskStorage, prepareTaskFilesystemsAsync, taskFilesystemOwner } from '../agents/container/storage.ts'; import { recoverLeftovers } from '../agents/recovery.ts'; import { createVendorNetwork, removeVendorNetwork, VendorNetworkCreationCleanupError, type VendorNetwork } from '../agents/network/network.ts'; @@ -715,6 +715,30 @@ describe('real Docker agent isolation', () => { expect(byAllocation(owner.allocationId)).toEqual([]); }, 60_000); + it('aborts storage allocation while a Docker call ignores SIGTERM, and removes what it created before settling', async () => { + const data = fixture(), owner = testOwner('abort-allocation'), marker = join(data.root, 'start-began'); + // The keeper really starts, then the client ignores SIGTERM and never returns. + const stubborn = [ + 'result = run(args);', + `require('node:fs').writeFileSync(${JSON.stringify(marker)}, '');`, + "process.on('SIGTERM', () => {});", + 'Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0);', + ].join('\n'); + const controller = new AbortController(), pgids: number[] = []; + const waitForStart = setInterval(() => { if (existsSync(marker)) controller.abort(); }, 20); + const began = performance.now(); + try { + await withDockerShim(['start'], stubborn, () => expect(prepareTaskFilesystemsAsync(data.clone, storageLimits, + imageId, owner, { signal: controller.signal, onProcessGroup: group => { pgids.push(group.pgid); } })) + .rejects.toThrow('cancelled')); + } finally { clearInterval(waitForStart); } + expect(performance.now() - began).toBeGreaterThanOrEqual(5_000); + expect(byAllocation(owner.allocationId)).toEqual([]); + // process.kill with a negative ID signals the group; the kill utility would read "-" as an option. + const groupAlive = (pgid: number) => { try { process.kill(-pgid, 0); return true; } catch { return false; } }; + for (const pgid of pgids) expect(groupAlive(pgid)).toBe(false); + }, 90_000); + it('keeps a failed allocation whose cleanup did not settle live, so recovery in this process refuses its runner', async () => { const data = fixture(), runnerOwner = randomBytes(16).toString('hex'); const owner = { runnerOwner, attemptId: 'unsettled-allocation', allocationId: randomUUID() }; diff --git a/test/agent-process-group.test.ts b/test/agent-process-group.test.ts new file mode 100644 index 00000000..844a2004 --- /dev/null +++ b/test/agent-process-group.test.ts @@ -0,0 +1,143 @@ +import { mkdtempSync, readFileSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { describe, expect, it } from 'vitest'; +import { runInProcessGroup, type ProcessGroup } from '../agents/process-group.ts'; +import { NOT_STARTED } from '../agents/docker.ts'; +import { createOutcomeUnknown } from '../agents/client-outcome.ts'; + +// Preparation subprocesses run in their own process group, and settle only once the whole group has exited (#51 item 5). +const env = { PATH: process.env.PATH }; +const groupAlive = (pgid: number) => { + try { process.kill(-pgid, 0); return true; } catch { return false; } +}; + +describe('runInProcessGroup', () => { + it('reports the group in the same turn as the spawn, and returns a normal exit', async () => { + let reported: ProcessGroup | undefined; + const pending = runInProcessGroup('sh', ['-c', 'echo out; echo err >&2; exit 3'], + { env, timeoutMs: 5_000, onProcessGroup: group => { reported = group; } }); + // Synchronously, before anything else can run: the caller can record it before a crash could lose it. + expect(reported?.pgid).toBeGreaterThan(1); + expect(Math.abs(reported!.startedAt - Date.now())).toBeLessThan(5_000); + expect(await pending).toMatchObject({ status: 3, stdout: 'out\n', stderr: 'err\n' }); + expect(groupAlive(reported!.pgid)).toBe(false); + }); + + it('kills what the leader left in its group before settling, even after a normal exit', async () => { + let group: ProcessGroup | undefined; + const began = performance.now(); + const outcome = await runInProcessGroup('sh', ['-c', 'sleep 60 >/dev/null 2>&1 & echo started'], + { env, timeoutMs: 30_000, onProcessGroup: reported => { group = reported; } }); + expect(outcome).toMatchObject({ status: 0, stdout: 'started\n' }); + expect(groupAlive(group!.pgid)).toBe(false); + expect(performance.now() - began).toBeLessThan(10_000); + }); + + it('bounds a group that ignores SIGTERM: SIGKILL after the grace period, then settles as a timeout', async () => { + let group: ProcessGroup | undefined; + const began = performance.now(); + const outcome = await runInProcessGroup('sh', ['-c', "trap '' TERM; sleep 60 & while :; do sleep 1; done"], + { env, timeoutMs: 200, graceMs: 400, onProcessGroup: reported => { group = reported; } }); + const elapsed = performance.now() - began; + expect(outcome.status).toBeNull(); + expect((outcome.error as NodeJS.ErrnoException).code).toBe('ETIMEDOUT'); + expect(createOutcomeUnknown({ status: outcome.status, cause: outcome.error })).toBe(true); + expect(elapsed).toBeGreaterThanOrEqual(600); + expect(groupAlive(group!.pgid)).toBe(false); + }); + + it('stops the group on abort and reports it as cancelled', async () => { + const controller = new AbortController(); + let group: ProcessGroup | undefined; + setTimeout(() => controller.abort(), 100); + const outcome = await runInProcessGroup('sleep', ['60'], + { env, timeoutMs: 30_000, signal: controller.signal, onProcessGroup: reported => { group = reported; } }); + expect(outcome.status).toBeNull(); + expect((outcome.error as NodeJS.ErrnoException).code).toBe('ABORT_ERR'); + expect(groupAlive(group!.pgid)).toBe(false); + }); + + it('kills the group and still waits for it when the caller cannot record it', async () => { + let group: ProcessGroup | undefined; + const outcome = await runInProcessGroup('sh', ['-c', "trap '' TERM; sleep 60 & while :; do sleep 1; done"], + { env, timeoutMs: 30_000, onProcessGroup: reported => { group = reported; throw new Error('database is locked'); } }); + expect(outcome.status).toBeNull(); + expect((outcome.error as NodeJS.ErrnoException).code).toBe('EUNRECORDED'); + expect(((outcome.error as Error).cause as Error).message).toBe('database is locked'); + expect(groupAlive(group!.pgid)).toBe(false); + }); + + it('stops at once when the signal is aborted during onProcessGroup', async () => { + const controller = new AbortController(); + let group: ProcessGroup | undefined; + const began = performance.now(); + const outcome = await runInProcessGroup('sleep', ['60'], { env, timeoutMs: 30_000, signal: controller.signal, + onProcessGroup: reported => { group = reported; controller.abort(); } }); + expect((outcome.error as NodeJS.ErrnoException).code).toBe('ABORT_ERR'); + expect(performance.now() - began).toBeLessThan(10_000); + expect(groupAlive(group!.pgid)).toBe(false); + }); + + it('settles when a process that left the group still holds its output pipe', async () => { + const dir = mkdtempSync(join(tmpdir(), 'escaped-')), pidFile = join(dir, 'pid'); + // The leader starts `sleep` in a new session (outside the group) that inherits stdout, then exits. + const escape = `const c = require('node:child_process').spawn('sleep', ['60'], { detached: true, ` + + `stdio: ['ignore', 'inherit', 'inherit'] }); require('node:fs').writeFileSync(${JSON.stringify(pidFile)}, ` + + `String(c.pid)); c.unref(); console.log('leader done');`; + try { + const began = performance.now(); + const outcome = await runInProcessGroup(process.execPath, ['-e', escape], { env, timeoutMs: 30_000 }); + // The output may be incomplete, so it is not reported as a success; what was read is kept. + expect(outcome).toMatchObject({ status: null, stdout: 'leader done\n' }); + expect((outcome.error as NodeJS.ErrnoException).code).toBe('ESTDIOHELD'); + expect(performance.now() - began).toBeLessThan(10_000); + } finally { + try { process.kill(Number(readFileSync(pidFile, 'utf8')), 'SIGKILL'); } catch { /* already gone */ } + rmSync(dir, { recursive: true, force: true }); + } + }, 30_000); + + it('counts time spent recording the group against the deadline', async () => { + const began = performance.now(); + const outcome = await runInProcessGroup('sleep', ['60'], { env, timeoutMs: 1_000, graceMs: 100, + // A slow durable write: the deadline must already be running while it blocks. + onProcessGroup: () => { const until = performance.now() + 1_000; while (performance.now() < until); } }); + expect((outcome.error as NodeJS.ErrnoException).code).toBe('ETIMEDOUT'); + expect(performance.now() - began).toBeLessThan(1_700); + }); + + it('stops a group whose output exceeds the limit', async () => { + const outcome = await runInProcessGroup('sh', ['-c', 'while :; do echo xxxxxxxxxxxxxxxx; done'], + { env, timeoutMs: 30_000, maxBuffer: 1024 }); + expect((outcome.error as NodeJS.ErrnoException).code).toBe('ENOBUFS'); + expect(outcome.stdout.length).toBeLessThanOrEqual(1024); + }); + + it('resolves, never rejects, when spawn itself throws', async () => { + let reported = false; + // A NUL byte in an argument makes spawn throw synchronously, as ENOMEM or E2BIG from the spawn itself would. + const outcome = await runInProcessGroup('sh', ['-c', 'echo a\0b'], { env, timeoutMs: 5_000, + onProcessGroup: () => { reported = true; } }); + expect(outcome.status).toBeNull(); + expect(outcome.error).toBeInstanceOf(Error); + expect(reported).toBe(false); + }); + + it('starts nothing for an aborted signal, an invalid deadline, or a missing program', async () => { + let reported = false; + const onProcessGroup = () => { reported = true; }; + for (const outcome of [ + await runInProcessGroup('sleep', ['1'], { env, timeoutMs: 5_000, signal: AbortSignal.abort(), onProcessGroup }), + await runInProcessGroup('sleep', ['1'], { env, timeoutMs: 0, onProcessGroup }), + // Longer than a Node timer can wait: accepting it would time the call out after 1 ms. + await runInProcessGroup('sleep', ['1'], { env, timeoutMs: 2 ** 31, onProcessGroup }), + ]) { + expect((outcome.error as NodeJS.ErrnoException).code).toBe(NOT_STARTED); + expect(createOutcomeUnknown({ status: outcome.status, cause: outcome.error })).toBe(false); + } + const missing = await runInProcessGroup('codeboost-no-such-program', [], { env, timeoutMs: 5_000, onProcessGroup }); + expect((missing.error as NodeJS.ErrnoException).code).toBe('ENOENT'); + expect(reported).toBe(false); + }); +}); diff --git a/test/agent-storage-async.test.ts b/test/agent-storage-async.test.ts new file mode 100644 index 00000000..bc0cb0ed --- /dev/null +++ b/test/agent-storage-async.test.ts @@ -0,0 +1,193 @@ +import { execFileSync } from 'node:child_process'; +import { randomUUID } from 'node:crypto'; +import { existsSync, mkdirSync, mkdtempSync, readdirSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { ProcessGroup } from '../agents/process-group.ts'; + +// The asynchronous storage allocation (#51 item 5), against a fake Docker CLI that keeps its objects as files, so the +// abort path runs without a daemon. The real-Docker suite covers the same path against Docker itself. +// spawn passes through unless a test makes it throw, as it does synchronously for errors such as ENOMEM. +const spawnFault = vi.hoisted(() => ({ on: undefined as undefined | ((args: readonly string[]) => boolean) })); +// Every Docker call passes through; a test can act after one call settles and before the next is requested. +const calls = vi.hoisted(() => ({ made: [] as string[][], afterCall: undefined as undefined | ((args: readonly string[]) => void) })); +vi.mock('../agents/process-group.ts', async importOriginal => { + const actual = await importOriginal(); + return { ...actual, runInProcessGroup: async (...call: Parameters) => { + calls.made.push([...call[1]]); + const outcome = await actual.runInProcessGroup(...call); + calls.afterCall?.(call[1]); + return outcome; + } }; +}); +vi.mock('node:child_process', async importOriginal => { + const actual = await importOriginal(); + return { ...actual, spawn: ((file: string, args: readonly string[], options: object) => { + if (spawnFault.on?.(args)) throw Object.assign(new Error('spawn ENOMEM'), { code: 'ENOMEM', syscall: 'spawn' }); + return actual.spawn(file, args, options); + }) as typeof actual.spawn }; +}); +vi.mock('../agents/container/image.ts', async importOriginal => ({ + ...await importOriginal(), assertBuiltAgentImage: () => {}, +})); +const { hasLiveTaskStorage, prepareTaskFilesystemsAsync, removeTaskFilesystems } = + await import('../agents/container/storage.ts'); +const { createTaskClone } = await import('../git/clone.ts'); + +const RUNNER = '0123456789abcdef0123456789abcdef'; +const IMAGE = `sha256:${'a'.repeat(64)}`; +const LIMITS = { workBytes: 16 * 1024 * 1024, workInodes: 512, metadataBytes: 16 * 1024 * 1024, metadataInodes: 512 }; +let root = '', state = '', path: string | undefined; +// Objects are files named by name, holding { id, labels }. `docker start` hangs ignoring SIGTERM when told to. +const FAKE_DOCKER = String.raw`#!${process.execPath} +const fs = require('node:fs'), path = require('node:path'), crypto = require('node:crypto'); +const state = __STATE__, args = process.argv.slice(2); +const objects = () => fs.readdirSync(state).filter(name => name.endsWith('.json')) + .map(file => ({ name: file.slice(0, -5), ...JSON.parse(fs.readFileSync(path.join(state, file), 'utf8')) })); +const find = ref => objects().find(object => object.name === ref || object.id === ref); +const labels = () => Object.fromEntries(args.flatMap((arg, i) => arg === '--label' ? [args[i + 1].split(/=(.*)/s).slice(0, 2)] : [])); +const save = (name, object) => fs.writeFileSync(path.join(state, name + '.json'), JSON.stringify(object)); +const [a, b] = args; +if ((a === 'ps' || b === 'ls') && fs.existsSync(path.join(state, 'hang-ls'))) { + fs.writeFileSync(path.join(state, 'ls-began'), ''); + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0); +} +if (a === 'ps' || b === 'ls') process.exit(0); +if (a === 'volume' && b === 'create') { save(args.at(-1), { kind: 'volume', labels: labels() }); console.log(args.at(-1)); process.exit(0); } +if (a === 'create') { + const id = crypto.randomBytes(32).toString('hex'); + save(args[args.indexOf('--name') + 1], { kind: 'container', id, labels: labels() }); console.log(id); process.exit(0); +} +if (a === 'start' && fs.existsSync(path.join(state, 'hang-start'))) { + fs.writeFileSync(path.join(state, 'start-began'), String(process.pid)); + process.on('SIGTERM', () => {}); + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0); +} +if (a === 'start') process.exit(0); +if (a === 'run') process.exit(0); +if (b === 'inspect') { + const object = find(args[2]); + if (!object) { console.error('Error: No such object: ' + args[2]); process.exit(1); } + console.log(JSON.stringify([object.kind === 'container' + ? { Id: object.id, Name: '/' + object.name, Config: { Labels: object.labels } } + : { Name: object.name, Labels: object.labels }])); + process.exit(0); +} +if (a === 'rm' || (a === 'volume' && b === 'rm')) { + const object = find(args.at(-1)); + if (object) fs.rmSync(path.join(state, object.name + '.json')); + process.exit(0); +} +console.error('fake docker: unsupported ' + args.join(' ')); process.exit(2); +`; + +const git = (cwd: string, ...args: string[]) => execFileSync('git', ['-c', 'core.hooksPath=/dev/null', ...args], + { cwd, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'] }).trim(); +const clone = () => { + const source = join(root, `source-${randomUUID()}`), parent = join(root, `staging-${randomUUID()}`); + mkdirSync(source); mkdirSync(parent); + git(source, 'init'); git(source, 'config', 'user.name', 'Test'); git(source, 'config', 'user.email', 'test@example.com'); + writeFileSync(join(source, 'file.txt'), 'trusted\n'); git(source, 'add', '.'); git(source, 'commit', '-m', 'baseline'); + return createTaskClone({ source, parent, taskId: 'task-1', head: git(source, 'rev-parse', 'HEAD') }); +}; +const owner = () => ({ runnerOwner: RUNNER, attemptId: `attempt-${randomUUID()}`, allocationId: randomUUID() }); +const stored = () => readdirSync(state).filter(name => name.endsWith('.json')); +const groupAlive = (pgid: number) => { + try { process.kill(-pgid, 0); return true; } catch { return false; } +}; + +beforeEach(() => { + root = mkdtempSync(join(tmpdir(), 'storage-async-')); state = join(root, 'state'); mkdirSync(state); + const bin = join(root, 'bin'); mkdirSync(bin); + // Storage passes Docker only an allowlisted environment, so the fake's state path is written into the script. + writeFileSync(join(bin, 'docker'), FAKE_DOCKER.replace('__STATE__', JSON.stringify(state)), { mode: 0o755 }); + path = process.env.PATH; + process.env.PATH = `${bin}:${path}`; +}); +afterEach(() => { + spawnFault.on = undefined; calls.afterCall = undefined; calls.made = []; + process.env.PATH = path; + rmSync(root, { recursive: true, force: true }); +}); + +describe('asynchronous task storage allocation', () => { + it('allocates the same storage as the synchronous helper, reporting every Docker process group', async () => { + const groups: ProcessGroup[] = []; + const filesystems = await prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), + { onProcessGroup: group => { groups.push(group); } }); + expect(stored().sort()).toEqual([`${filesystems.keeper}.json`, `${filesystems.metadataVolume}.json`, + `${filesystems.workVolume}.json`].sort()); + expect(groups.length).toBeGreaterThanOrEqual(8); + for (const group of groups) expect(groupAlive(group.pgid)).toBe(false); + removeTaskFilesystems(filesystems); + expect(stored()).toEqual([]); + }); + + it('on abort, kills a Docker call that ignores SIGTERM, then removes everything it created before settling', async () => { + writeFileSync(join(state, 'hang-start'), ''); + const groups: ProcessGroup[] = [], controller = new AbortController(); + let ticks = 0; + const ticker = setInterval(() => { ticks += 1; }, 50); + const waitForStart = setInterval(() => { if (existsSync(join(state, 'start-began'))) controller.abort(); }, 20); + const began = performance.now(); + try { + const error = await prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), + { signal: controller.signal, onProcessGroup: group => { groups.push(group); } }).then(() => undefined, caught => caught); + expect((error as Error).message).toContain('cancelled'); + } finally { clearInterval(ticker); clearInterval(waitForStart); } + // SIGTERM was ignored, so the call was killed only after the 5 s grace period. + expect(performance.now() - began).toBeGreaterThanOrEqual(5_000); + expect(ticks).toBeGreaterThan(50); + // The volumes and the keeper it created are gone, and no group it started is still running. + expect(stored()).toEqual([]); + for (const group of groups) expect(groupAlive(group.pgid)).toBe(false); + }, 30_000); + + it('reports an abort during the allocation ID check as cancelled, having created nothing', async () => { + writeFileSync(join(state, 'hang-ls'), ''); + const controller = new AbortController(); + const waitForList = setInterval(() => { if (existsSync(join(state, 'ls-began'))) controller.abort(); }, 20); + try { + await expect(prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), { signal: controller.signal })) + .rejects.toThrow('cancelled'); + } finally { clearInterval(waitForList); } + expect(stored()).toEqual([]); + }, 30_000); + + it('removes what it created and releases the allocation when a later Docker call cannot be spawned', async () => { + // Both volumes are created; spawning the keeper's `docker create` then fails synchronously. + spawnFault.on = args => args[0] === 'create'; + await expect(prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner())).rejects.toThrow('ENOMEM'); + expect(stored()).toEqual([]); + expect(hasLiveTaskStorage(RUNNER)).toBe(false); + }); + + it('reports an abort that lands between two Docker calls as a cancel, and removes what it created', async () => { + const controller = new AbortController(), groups: ProcessGroup[] = []; + // Aborted after the work volume's create settles and before the next call (the reuse check) is requested. + calls.afterCall = args => { if (args[0] === 'volume' && args[1] === 'create') controller.abort(); }; + const error = await prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), + { signal: controller.signal, onProcessGroup: group => { groups.push(group); } }).then(() => undefined, caught => caught); + expect(error).toMatchObject({ name: 'AbortError', code: 'ABORT_ERR' }); + // Exactly one call was refused before it started (the reuse check); every other call ran and reported its group. + expect(calls.made.length - groups.length).toBe(1); + expect(stored()).toEqual([]); + expect(hasLiveTaskStorage(RUNNER)).toBe(false); + }); + + it('refuses a deadline longer than a Node timer can wait, before any Docker call', async () => { + await expect(prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), { timeoutMs: 2 ** 31 })) + .rejects.toThrow('at most'); + expect(calls.made).toEqual([]); + expect(stored()).toEqual([]); + }); + + it('creates nothing when the signal is already aborted', async () => { + const groups: ProcessGroup[] = []; + await expect(prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), + { signal: AbortSignal.abort(), onProcessGroup: group => { groups.push(group); } })).rejects.toThrow('cancelled'); + expect(groups).toEqual([]); + expect(stored()).toEqual([]); + }); +}); diff --git a/test/agent-walk-pauses.test.ts b/test/agent-walk-pauses.test.ts new file mode 100644 index 00000000..52a802bd --- /dev/null +++ b/test/agent-walk-pauses.test.ts @@ -0,0 +1,83 @@ +import { execFileSync } from 'node:child_process'; +import { randomUUID } from 'node:crypto'; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +// The asynchronous clone and storage allocation pause every 1,000 entries they walk, including entries read from one +// huge directory, so the event loop (and an abort) is never held for a whole directory (#51 item 5). +const trace = vi.hoisted(() => ({ run: 0, longest: 0, total: 0 })); +vi.mock('node:fs', async importOriginal => { + const actual = await importOriginal(); + return { ...actual, opendirSync: ((...args: Parameters) => { + const directory = actual.opendirSync(...args), read = directory.readSync.bind(directory); + directory.readSync = () => { + const entry = read(); + if (entry) { trace.run += 1; trace.total += 1; trace.longest = Math.max(trace.longest, trace.run); } + return entry; + }; + return directory; + }) as typeof actual.opendirSync }; +}); +vi.mock('../agents/container/image.ts', async importOriginal => ({ + ...await importOriginal(), assertBuiltAgentImage: () => {}, +})); +const { createTaskClone, createTaskCloneAsync } = await import('../git/clone.ts'); +const { prepareTaskFilesystemsAsync } = await import('../agents/container/storage.ts'); + +const ENTRIES = 2_500; +let root = ''; +const realSetImmediate = globalThis.setImmediate; +const git = (cwd: string, ...args: string[]) => execFileSync('git', ['-c', 'core.hooksPath=/dev/null', ...args], + { cwd, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'] }).trim(); +const repository = () => { + const source = join(root, 'source'), parent = join(root, 'staging'); + mkdirSync(source); mkdirSync(parent); + git(source, 'init'); git(source, 'config', 'user.name', 'Test'); git(source, 'config', 'user.email', 'test@example.com'); + writeFileSync(join(source, 'file.txt'), 'trusted\n'); git(source, 'add', '.'); git(source, 'commit', '-m', 'baseline'); + return { source, parent, head: git(source, 'rev-parse', 'HEAD'), taskId: 'task-1' }; +}; +beforeEach(() => { + root = mkdtempSync(join(tmpdir(), 'walk-pauses-')); + trace.run = 0; trace.longest = 0; trace.total = 0; + // Every pause of the asynchronous drivers goes through setImmediate: the run of entries read restarts there. + vi.spyOn(globalThis, 'setImmediate').mockImplementation(((callback: (...args: unknown[]) => void, ...args: unknown[]) => { + trace.run = 0; + return realSetImmediate(callback, ...args); + }) as typeof setImmediate); +}); +afterEach(() => { vi.restoreAllMocks(); rmSync(root, { recursive: true, force: true }); }); + +describe('pauses while walking one large directory', () => { + it('in the asynchronous clone, reading object storage', async () => { + const input = repository(); + // One object-storage directory with far more entries than one pause interval. + const info = join(input.source, '.git', 'objects', 'info'); + for (let index = 0; index < ENTRIES; index++) writeFileSync(join(info, `entry-${index}`), ''); + // Each Git call is awaited, which yields the event loop too; the object checks before and after the clone call are + // separate walks, so a run only counts entries read without any yield in between. + await createTaskCloneAsync({ ...input, onProcessGroup: () => { trace.run = 0; } }); + expect(trace.total).toBeGreaterThanOrEqual(ENTRIES); + expect(trace.longest).toBeLessThanOrEqual(1_000); + }, 60_000); + + it('in the asynchronous storage allocation, checking the checkout for links', async () => { + const input = repository(); + const clone = createTaskClone(input); + const wide = join(clone.directory, 'wide'); + mkdirSync(wide); + for (let index = 0; index < ENTRIES; index++) writeFileSync(join(wide, `entry-${index}`), ''); + trace.run = 0; trace.longest = 0; trace.total = 0; + // No Docker on PATH: the allocation fails at its first Docker call, after the link walk has finished. + const path = process.env.PATH; + process.env.PATH = '/nonexistent'; + try { + await expect(prepareTaskFilesystemsAsync(clone, { workBytes: 1 << 24, workInodes: 512, metadataBytes: 1 << 24, + metadataInodes: 512 }, `sha256:${'a'.repeat(64)}`, { runnerOwner: '0'.repeat(32), attemptId: 'walk', + allocationId: randomUUID() })).rejects.toThrow(); + } finally { process.env.PATH = path; } + expect(trace.total).toBeGreaterThanOrEqual(ENTRIES); + expect(trace.longest).toBeLessThanOrEqual(1_000); + }, 60_000); +});