commit e44064ed125147d0268e4d98d5f78966b607dcc7
parent bd658275e2b0dd996745b9b63483127d5cbbac12
Author: finwo <finwo@pm.me>
Date: Sat, 19 Sep 2026 01:50:57 +0200
Worker agent: containerised jobs, features and artifacts
Diffstat:
18 files changed, 2138 insertions(+), 9 deletions(-)
diff --git a/README.md b/README.md
@@ -30,10 +30,11 @@ Under construction. Working today:
- local and S3 compatible object storage
- pipeline parsing, matrix and architecture expansion, dependency graphs
- triggers, run scheduling, the worker API, live logs, artifacts, reaping
+ - the worker: containerised jobs, services, features, caches, cleanup
-Not yet built: the worker agent itself, user accounts and the admin API, the
-read-api service, and the dashboard. Until the admin API exists, projects and
-worker tokens are managed with `src/admin-cli.js`.
+Not yet built: user accounts and the admin API, the read-api service, and the
+dashboard. Until the admin API exists, projects and worker tokens are managed
+with `src/admin-cli.js`.
Installation
------------
@@ -88,6 +89,14 @@ and set `CONDUCTOR_URL`, `CONDUCTOR_PROJECT` and `CONDUCTOR_SECRET` for it.
GitHub, Gitea and GitLab webhooks are accepted at the same endpoint, so a
forge can drive it instead.
+Then run a worker, on this machine or any other:
+
+```sh
+node src/worker/agent.js --config worker.json
+```
+
+See [docs/worker.md](docs/worker.md), and `examples/worker.json`.
+
Pipelines
---------
diff --git a/docs/worker.md b/docs/worker.md
@@ -0,0 +1,168 @@
+Running a worker
+================
+
+A worker polls a conductor for jobs and runs them in containers. It needs no
+inbound connectivity, so it can sit behind NAT, and it needs no repository
+credentials, because the conductor serves the source for the exact commit a
+job was handed.
+
+The worker has no npm dependencies. Copying `src/worker` onto a host with
+node, tar and a container runtime is enough.
+
+Requirements
+------------
+
+ - node 24 or newer
+ - a docker compatible CLI: docker, podman or nerdctl
+ - `tar`, to unpack job sources
+ - `git`, only for projects configured to clone rather than download
+
+Getting a token
+---------------
+
+Ask whoever runs the conductor for a worker token. They create one with:
+
+```sh
+node src/admin-cli.js token:add friends-pi
+```
+
+The token is shown once and stored only as a hash, so it cannot be recovered
+later. If it is lost, issue a new one and remove the old.
+
+Configuration
+-------------
+
+JSON is the native format. YAML also works when the `yaml` package is
+installed, which it is in the published image.
+
+```json
+{
+ "conductor_url": "https://ci.example.com",
+ "name": "friends-pi",
+ "token_file": "/run/secrets/worker_token",
+ "arches": ["aarch64"],
+ "concurrency": 2,
+ "features": {
+ "dind": { "privileged": true },
+ "sign-key": {
+ "mounts": ["/srv/keys/build.rsa:/keys/build.rsa:ro"],
+ "env": { "UNOS_SIGN_KEY": "/keys/build.rsa" }
+ }
+ }
+}
+```
+
+```sh
+node src/worker/agent.js --config worker.json
+```
+
+| Key | Default | Description |
+| ---------------- | -------------------- | ----------------------------------------------- |
+| `conductor_url` | `http://127.0.0.1:8080` | Base URL of the conductor. |
+| `name` | hostname | Shown against jobs this worker ran. |
+| `token` | none | The worker token, inline. |
+| `token_file` | none | Path to a file holding the token. Preferred. |
+| `arches` | `[]` | Architectures offered. Empty means no `arch` jobs, only jobs that declare none. |
+| `features` | `{}` | Capabilities offered, see below. |
+| `concurrency` | `1` | Jobs run at once. |
+| `poll_interval` | `5` | Seconds between polls when idle. |
+| `docker` | `docker` | Runtime CLI. Set to `podman` or `nerdctl`. |
+| `shell` | `sh` | Shell used to run a job script in its image. |
+| `workspace_root` | a temp directory | Where job workspaces are created. |
+| `cache_root` | a temp directory | Where job caches are kept between runs. |
+
+Every key can also be set from the environment: `CONDUCTOR_URL`,
+`CONDUCTOR_WORKER_TOKEN`, `CONDUCTOR_WORKER_TOKEN_FILE`,
+`CONDUCTOR_WORKER_ARCHES`, `CONDUCTOR_WORKER_CONCURRENCY`,
+`CONDUCTOR_WORKER_DOCKER`, and so on. The environment wins over the file.
+
+Features
+--------
+
+A feature is a name this worker advertises, together with whatever local
+resources a job asking for it should receive. The conductor only ever learns
+the name. It will not send a job to a worker that does not advertise every
+feature the job's `requires` lists.
+
+```json
+"features": {
+ "sign-key": {
+ "mounts": ["/srv/keys/build.rsa:/keys/build.rsa:ro"],
+ "env": { "SIGN_KEY": "/keys/build.rsa" },
+ "devices": [],
+ "privileged": false
+ }
+}
+```
+
+A pipeline then asks for it by name:
+
+```yaml
+package:
+ requires: [sign-key]
+ script: ['./sign.sh']
+```
+
+This is how a signing key reaches a build without the conductor ever holding
+it. The key stays on the machine that owns it, and only jobs that explicitly
+ask for it are ever scheduled there.
+
+`privileged: true` is what a docker in docker service needs. Grant it only
+to features you intend to be privileged, since it removes the isolation
+between the job and the host.
+
+What a job gets
+---------------
+
+Each job runs in its own container, on its own network, in its own
+workspace:
+
+ - the repository tree at the job's commit, unpacked into `/workspace`
+ - `/workspace` as the working directory
+ - the job's `env`, plus `ARCH` and `MATRIX_*` for a fanned out job
+ - `CONDUCTOR_PROJECT`, `CONDUCTOR_RUN_ID`, `CONDUCTOR_RUN_NUMBER`,
+ `CONDUCTOR_JOB`, `CONDUCTOR_JOB_ID`, `CONDUCTOR_SHA`, `CONDUCTOR_REF`
+ and `CONDUCTOR_ATTEMPT`
+ - any `services`, reachable by their alias on the job network
+ - mounts and environment from the features it requires
+
+Output is streamed to the conductor as it happens. Artifacts are uploaded
+when the job finishes, and the workspace is then removed.
+
+Caches are bind mounted from `cache_root` straight onto their path in the
+workspace. A cache is an optimisation and nothing more: a job must still work
+with an empty one, and a cache is never shared between workers.
+
+Operational notes
+-----------------
+
+**Cleanup.** Containers, networks and workspaces are removed whatever the
+outcome. Most images run as root, so files a job creates are owned by root
+while the worker usually is not. When removing the workspace fails for that
+reason, the worker empties it from inside a container first. Nothing is left
+behind either way.
+
+**Shutdown.** On SIGINT or SIGTERM the worker stops polling and lets running
+jobs finish. A second signal exits immediately, and the conductor will
+eventually reap whatever was left running.
+
+**Failures.** A worker that stops reporting has its jobs requeued or failed
+by the conductor after `scheduler.heartbeat_timeout`. Losing a worker never
+strands a run.
+
+**Log loss.** If the conductor cannot be reached while a job runs, log output
+for that window is dropped rather than failing the job. The job's outcome is
+still reported.
+
+Trust
+-----
+
+A worker is trusted with the source of the commits it builds and with
+whatever its features grant it. It is not trusted with anything else: it
+holds no repository credentials, it cannot read other refs, it cannot reach
+another project's jobs, and it cannot write to storage except by uploading
+artifacts for a job it currently holds.
+
+Running a job means running code from the repository. Only accept work from
+a conductor whose projects you are willing to execute, and keep features
+narrow.
diff --git a/examples/worker.json b/examples/worker.json
@@ -0,0 +1,20 @@
+{
+ "conductor_url": "https://ci.example.com",
+ "name": "friends-pi",
+ "token_file": "/run/secrets/worker_token",
+ "arches": ["aarch64"],
+ "concurrency": 2,
+ "poll_interval": 5,
+ "docker": "docker",
+ "workspace_root": "/var/lib/conductor/work",
+ "cache_root": "/var/lib/conductor/cache",
+ "features": {
+ "dind": {
+ "privileged": true
+ },
+ "sign-key": {
+ "mounts": ["/srv/keys/build.rsa:/keys/build.rsa:ro"],
+ "env": { "SIGN_KEY": "/keys/build.rsa" }
+ }
+ }
+}
diff --git a/src/conductor/routes/workers.js b/src/conductor/routes/workers.js
@@ -73,9 +73,13 @@ export default async function workerRoutes(fastify, { cfg, db, git, logs, storag
job: {
id: job.id,
run_id: job.run_id,
+ project_id: job.project_id,
name: job.name,
arch: job.arch,
image: job.image,
+ // The worker maps these onto its own local mounts, environment and
+ // privileges. Without them no feature is applied at all.
+ requires: job.requires,
script: job.script,
env: job.env,
services: job.services,
@@ -84,6 +88,13 @@ export default async function workerRoutes(fastify, { cfg, db, git, logs, storag
timeout: job.timeout,
attempt: job.attempt,
sha: job.sha,
+ ref: job.ref,
+ // Values the worker must redact from the log stream. Populated
+ // once project variables exist; masking also happens on ingest.
+ masked: job.masked ?? [],
+ // Derived from the server's own timeout so a worker never has to
+ // guess how often it is expected to check in.
+ heartbeat_interval: Math.max(5, Math.floor(cfg.scheduler.heartbeat_timeout / 4)),
// A worker fetches the tree from the conductor by default, so it
// never needs repository credentials. Projects set to clone mode
// get the URL instead.
diff --git a/src/conductor/scheduler.js b/src/conductor/scheduler.js
@@ -33,7 +33,8 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, logger
// satisfied, which is the whole point of the flag.
const ELIGIBLE = `
SELECT j.id, j.run_id, j.name, j.arch, j.image, j.requires, j.spec,
- j.timeout, j.attempt, j.max_attempts, r.head_sha, r.project_id
+ j.timeout, j.attempt, j.max_attempts,
+ r.head_sha, r.project_id, r.ref, r.number
FROM jobs j
JOIN runs r ON r.id = j.run_id
WHERE j.state = 'queued'
@@ -233,6 +234,7 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, logger
if (claimed.changes !== 1) continue;
const spec = JSON.parse(candidate.spec);
+ const attempt = candidate.attempt + 1;
return {
id: candidate.id,
run_id: candidate.run_id,
@@ -241,12 +243,26 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, logger
arch: candidate.arch,
image: candidate.image,
sha: candidate.head_sha,
+ ref: candidate.ref,
requires,
timeout: candidate.timeout,
- attempt: candidate.attempt + 1,
+ attempt,
max_attempts: candidate.max_attempts,
script: spec.script,
- env: spec.env,
+ // Pipeline env, plus the facts about this run that a script
+ // should not have to be told twice. These are set last so a
+ // pipeline cannot quietly redefine them.
+ env: {
+ ...spec.env,
+ CONDUCTOR_PROJECT: candidate.project_id,
+ CONDUCTOR_RUN_ID: candidate.run_id,
+ CONDUCTOR_RUN_NUMBER: String(candidate.number),
+ CONDUCTOR_JOB: candidate.name,
+ CONDUCTOR_JOB_ID: candidate.id,
+ CONDUCTOR_SHA: candidate.head_sha,
+ CONDUCTOR_REF: candidate.ref ?? '',
+ CONDUCTOR_ATTEMPT: String(attempt),
+ },
services: spec.services,
artifacts: spec.artifacts,
cache: spec.cache,
diff --git a/src/worker/agent.js b/src/worker/agent.js
@@ -0,0 +1,137 @@
+#!/usr/bin/env node
+// src/worker/agent.js - the worker
+//
+// Polls a conductor for work and runs it in containers. Pull based, so the
+// worker needs no inbound connectivity and can sit behind any amount of NAT.
+//
+// Deliberately free of npm dependencies: copy src/worker onto a host with
+// node, tar and a container runtime and it runs. That matters because the
+// point of this design is accepting build capacity from machines you do not
+// administer.
+//
+// Usage:
+// node src/worker/agent.js [--config worker.json] [--once]
+
+import { loadWorkerConfig, featureNames } from './config.js';
+import { createClient } from './client.js';
+import { createRuntime } from './docker.js';
+import { hasTar, hasGit } from './source.js';
+import { runJob } from './job.js';
+
+const stamp = () => new Date().toISOString().replace('T', ' ').slice(0, 19);
+
+const logger = {
+ info: (m) => console.log(`${stamp()} ${m}`),
+ warn: (m) => console.warn(`${stamp()} warning: ${m}`),
+ error: (m) => console.error(`${stamp()} error: ${m}`),
+ debug: () => {},
+};
+
+function parseArgv(argv) {
+ const out = { config: null, once: false };
+ for (let i = 0; i < argv.length; i += 1) {
+ if (argv[i] === '--config' || argv[i] === '-c') {
+ out.config = argv[i + 1];
+ i += 1;
+ } else if (argv[i] === '--once') {
+ out.once = true;
+ } else if (argv[i] === '--help' || argv[i] === '-h') {
+ console.log('usage: agent.js [--config <file>] [--once]');
+ process.exit(0);
+ }
+ }
+ return out;
+}
+
+export async function main(argv = process.argv.slice(2)) {
+ const args = parseArgv(argv);
+ const cfg = await loadWorkerConfig(args.config);
+ const client = createClient(cfg);
+ const runtime = createRuntime(cfg, { logger });
+
+ // Fail at startup rather than on the first job, when a missing tool would
+ // otherwise look like a broken pipeline.
+ if (!(await runtime.available())) {
+ throw new Error(`container runtime ${JSON.stringify(cfg.docker)} is not usable; is the daemon running?`);
+ }
+ if (!(await hasTar())) {
+ throw new Error('tar is required to unpack job sources but was not found on PATH');
+ }
+ if (!(await hasGit())) {
+ logger.warn('git was not found; projects configured for clone mode will fail');
+ }
+
+ const features = featureNames(cfg);
+ logger.info(`worker ${cfg.name} polling ${cfg.conductor_url}`);
+ logger.info(` arches: ${cfg.arches.length > 0 ? cfg.arches.join(', ') : '(any)'}`);
+ logger.info(` features: ${features.length > 0 ? features.join(', ') : '(none)'}`);
+ logger.info(` concurrency: ${cfg.concurrency}`);
+
+ let stopping = false;
+ const running = new Set();
+
+ const shutdown = (signal) => {
+ if (stopping) {
+ logger.warn('second signal, exiting now');
+ process.exit(1);
+ }
+ stopping = true;
+ logger.info(`${signal} received, finishing ${running.size} running job(s)`);
+ };
+ for (const signal of ['SIGINT', 'SIGTERM']) process.once(signal, () => shutdown(signal));
+
+ const sleep = (ms) => new Promise((resolve) => { setTimeout(resolve, ms).unref?.(); });
+
+ // One loop per concurrency slot. Each polls independently, so a slow job
+ // in one slot does not stall the others.
+ async function slot(index) {
+ let backoff = cfg.poll_interval;
+
+ while (!stopping) {
+ let job = null;
+ try {
+ job = await client.poll({ arches: cfg.arches, features, name: cfg.name });
+ backoff = cfg.poll_interval;
+ } catch (e) {
+ // A conductor that is down or restarting should not become a busy
+ // loop, so back off up to a minute.
+ logger.warn(`poll failed: ${e.message}`);
+ await sleep(backoff * 1000);
+ backoff = Math.min(backoff * 2, 60);
+ continue;
+ }
+
+ if (!job) {
+ if (args.once) return;
+ await sleep(cfg.poll_interval * 1000);
+ continue;
+ }
+
+ running.add(job.id);
+ logger.info(`[${index}] ${job.name} (${job.image})`);
+ const started = Date.now();
+ try {
+ const result = await runJob({ cfg, client, runtime, job, logger });
+ const seconds = ((Date.now() - started) / 1000).toFixed(1);
+ logger.info(`[${index}] ${job.name} ${result.success ? 'succeeded' : 'failed'} in ${seconds}s`);
+ } catch (e) {
+ logger.error(`[${index}] ${job.name} crashed the worker loop: ${e.stack ?? e.message}`);
+ } finally {
+ running.delete(job.id);
+ }
+
+ if (args.once) return;
+ }
+ }
+
+ await Promise.all(Array.from({ length: cfg.concurrency }, (_, i) => slot(i + 1)));
+ logger.info('stopped');
+}
+
+// Only run when executed directly, so the module stays importable by tests.
+if (process.argv[1] && import.meta.url === `file://${process.argv[1]}`) {
+ main().catch((e) => {
+ logger.error(e.message);
+ process.exit(1);
+ });
+}
diff --git a/src/worker/artifacts.js b/src/worker/artifacts.js
@@ -0,0 +1,84 @@
+// src/worker/artifacts.js - collecting and uploading job output
+//
+// Patterns are matched with the glob support built into node, so there is
+// nothing to install. Matches are confined to the workspace: a job can write
+// a symlink pointing anywhere it likes, and following one would upload a
+// file the job was never given.
+
+import fs from 'node:fs/promises';
+import path from 'node:path';
+import { Readable } from 'node:stream';
+import { createReadStream } from 'node:fs';
+import { SCRIPT_DIR } from './script.js';
+
+const MAX_FILES = 2000;
+
+export async function collectArtifacts(workspace, patterns, { logger = console } = {}) {
+ const root = await fs.realpath(workspace);
+ const seen = new Map();
+
+ for (const pattern of patterns) {
+ let matches;
+ try {
+ matches = await Array.fromAsync(fs.glob(pattern, { cwd: root }));
+ } catch (e) {
+ logger.warn?.(`artifact pattern ${pattern} failed: ${e.message}`);
+ continue;
+ }
+
+ for (const relative of matches) {
+ if (seen.size >= MAX_FILES) {
+ logger.warn?.(`artifact limit of ${MAX_FILES} files reached, ignoring the rest`);
+ return [...seen.values()];
+ }
+
+ const normalized = relative.split(path.sep).join('/');
+ // The generated script lives in the workspace and is not output.
+ if (normalized === SCRIPT_DIR || normalized.startsWith(`${SCRIPT_DIR}/`)) continue;
+ if (seen.has(normalized)) continue;
+
+ const absolute = path.join(root, relative);
+ let resolved;
+ let stat;
+ try {
+ resolved = await fs.realpath(absolute);
+ stat = await fs.stat(resolved);
+ } catch {
+ // Vanished between globbing and stat, or a broken symlink.
+ continue;
+ }
+
+ if (!stat.isFile()) continue;
+ if (resolved !== root && !resolved.startsWith(root + path.sep)) {
+ logger.warn?.(`refusing artifact ${normalized}: resolves outside the workspace`);
+ continue;
+ }
+
+ seen.set(normalized, { path: normalized, absolute: resolved, size: stat.size });
+ }
+ }
+
+ return [...seen.values()];
+}
+
+export function shouldCollect(when, success) {
+ if (when === 'always') return true;
+ if (when === 'on_failure') return !success;
+ return success;
+}
+
+export async function uploadArtifacts(client, url, files, { logger = console } = {}) {
+ const uploaded = [];
+ for (const file of files) {
+ try {
+ const body = Readable.toWeb(createReadStream(file.absolute));
+ const result = await client.uploadArtifact(url, file.path, body, file.size);
+ uploaded.push(result);
+ } catch (e) {
+ // An artifact that cannot be stored is worth reporting, but the job
+ // itself already succeeded or failed on its own merits.
+ logger.warn?.(`could not upload ${file.path}: ${e.message}`);
+ }
+ }
+ return uploaded;
+}
diff --git a/src/worker/client.js b/src/worker/client.js
@@ -0,0 +1,114 @@
+// src/worker/client.js - talking to the conductor
+//
+// Built on fetch, so there is nothing to install. Every call carries the
+// worker token. Network errors are surfaced rather than retried here; the
+// agent decides what a failure means, since a failed poll and a failed log
+// append want very different handling.
+
+export class ConductorError extends Error {
+ constructor(message, { status = 0, body = null } = {}) {
+ super(message);
+ this.name = 'ConductorError';
+ this.status = status;
+ this.body = body;
+ }
+}
+
+export function createClient(cfg) {
+ const base = cfg.conductor_url;
+ const auth = { authorization: `Bearer ${cfg.token}` };
+
+ async function readError(res, action) {
+ let body = null;
+ let detail = '';
+ try {
+ const text = await res.text();
+ detail = text.slice(0, 512);
+ body = JSON.parse(text);
+ } catch {
+ // A non JSON body is still worth reporting as text.
+ }
+ return new ConductorError(`${action} failed: ${res.status} ${res.statusText} ${detail}`.trim(), {
+ status: res.status,
+ body,
+ });
+ }
+
+ return {
+ // Returns a job, or null when there is nothing to do.
+ async poll({ arches, features, name }) {
+ const query = new URLSearchParams();
+ if (arches.length > 0) query.set('arches', arches.join(','));
+ if (features.length > 0) query.set('features', features.join(','));
+ if (name) query.set('name', name);
+
+ const res = await fetch(`${base}/api/workers/poll?${query}`, { headers: auth });
+ if (res.status === 204) return null;
+ if (!res.ok) throw await readError(res, 'poll');
+ return (await res.json()).job;
+ },
+
+ // The tree at the job's commit. Returned as a web stream so it can be
+ // piped straight into tar without buffering.
+ async source(url) {
+ const res = await fetch(url, { headers: auth });
+ if (!res.ok) throw await readError(res, 'source download');
+ return res.body;
+ },
+
+ // Returns the new log size. A 409 means the conductor and the worker
+ // disagree about the offset, and carries the offset to resume from.
+ async appendLog(url, chunk, offset) {
+ const res = await fetch(url, {
+ method: 'POST',
+ headers: { ...auth, 'content-type': 'application/octet-stream', 'x-log-offset': String(offset) },
+ body: chunk,
+ });
+
+ if (res.status === 409) {
+ const body = await res.json().catch(() => ({}));
+ return { conflict: true, expected: body.expected_offset ?? 0 };
+ }
+ if (!res.ok) throw await readError(res, 'log append');
+ return await res.json();
+ },
+
+ async uploadArtifact(url, relPath, body, size) {
+ const res = await fetch(url, {
+ method: 'POST',
+ headers: {
+ ...auth,
+ 'content-type': 'application/octet-stream',
+ 'content-length': String(size),
+ 'x-artifact-path': relPath,
+ },
+ body,
+ duplex: 'half',
+ });
+ if (!res.ok) throw await readError(res, `artifact upload (${relPath})`);
+ return await res.json();
+ },
+
+ // Reports that the job is alive, and learns whether it should stop.
+ async heartbeat(url) {
+ const res = await fetch(url, {
+ method: 'POST',
+ headers: { ...auth, 'content-type': 'application/json' },
+ body: '{}',
+ });
+ if (res.status === 404) return { cancelled: true, gone: true };
+ if (!res.ok) throw await readError(res, 'heartbeat');
+ return await res.json();
+ },
+
+ async done(url, { success, exitCode, error }) {
+ const res = await fetch(url, {
+ method: 'POST',
+ headers: { ...auth, 'content-type': 'application/json' },
+ body: JSON.stringify({ success, exit_code: exitCode ?? null, error: error ?? null }),
+ });
+ if (!res.ok) throw await readError(res, 'done');
+ return await res.json();
+ },
+ };
+}
diff --git a/src/worker/config.js b/src/worker/config.js
@@ -0,0 +1,206 @@
+// src/worker/config.js - worker configuration
+//
+// The worker deliberately has no npm dependencies, so that a third party can
+// copy src/worker into a host and run it with nothing but node and a
+// container runtime. JSON is therefore the native config format. YAML is
+// accepted too, but only when the yaml package happens to be resolvable,
+// which it is in the official image.
+//
+// Features are the important part. A feature is a name the worker
+// advertises, plus whatever local resources a job asking for it should get:
+// mounts, environment, or a privileged container. The conductor only ever
+// sees the name, so a signing key can be handed to a build without the
+// server holding it.
+
+import fs from 'node:fs';
+import path from 'node:path';
+import os from 'node:os';
+
+const DEFAULTS = {
+ conductor_url: 'http://127.0.0.1:8080',
+ name: os.hostname(),
+ token: null,
+ token_file: null,
+ arches: [],
+ features: {},
+ concurrency: 1,
+ poll_interval: 5,
+ // Container runtime CLI. Podman and nerdctl are argument compatible.
+ docker: 'docker',
+ // Shell used to run a job script inside its image.
+ shell: 'sh',
+ workspace_root: null,
+ cache_root: null,
+ // Applied when a job does not ask for something longer.
+ default_timeout: 3600,
+};
+
+const ENV_MAP = [
+ ['CONDUCTOR_URL', 'conductor_url', String],
+ ['CONDUCTOR_WORKER_NAME', 'name', String],
+ ['CONDUCTOR_WORKER_TOKEN', 'token', String],
+ ['CONDUCTOR_WORKER_TOKEN_FILE', 'token_file', String],
+ ['CONDUCTOR_WORKER_ARCHES', 'arches', (v) => splitList(v)],
+ ['CONDUCTOR_WORKER_CONCURRENCY', 'concurrency', toInt],
+ ['CONDUCTOR_WORKER_POLL_INTERVAL', 'poll_interval', toInt],
+ ['CONDUCTOR_WORKER_DOCKER', 'docker', String],
+ ['CONDUCTOR_WORKER_SHELL', 'shell', String],
+ ['CONDUCTOR_WORKER_WORKSPACE', 'workspace_root', String],
+ ['CONDUCTOR_WORKER_CACHE', 'cache_root', String],
+];
+
+function splitList(value) {
+ return String(value).split(',').map((s) => s.trim()).filter(Boolean);
+}
+
+function toInt(v) {
+ const n = parseInt(v, 10);
+ if (!Number.isFinite(n)) throw new Error(`expected an integer, got ${JSON.stringify(v)}`);
+ return n;
+}
+
+function isPlainObject(v) {
+ return v !== null && typeof v === 'object' && !Array.isArray(v);
+}
+
+async function readConfigFile(file) {
+ const text = fs.readFileSync(file, 'utf8');
+ if (file.endsWith('.json')) return JSON.parse(text);
+
+ try {
+ const YAML = (await import('yaml')).default;
+ return YAML.parse(text);
+ } catch (e) {
+ if (e instanceof SyntaxError || e.name === 'YAMLParseError') throw e;
+ throw new Error(
+ `cannot read ${file}: the yaml package is not installed. ` +
+ 'Install it, or use a .json config file.',
+ { cause: e }
+ );
+ }
+}
+
+// A feature maps a name onto local resources. Everything is optional, so
+// `{ "dind": {} }` is a valid way to advertise a capability that needs no
+// setup of its own.
+function normalizeFeature(name, raw, problems) {
+ if (raw === null || raw === undefined) return { mounts: [], env: {}, privileged: false, devices: [] };
+ if (!isPlainObject(raw)) {
+ problems.push(`features.${name}: expected a mapping`);
+ return null;
+ }
+
+ const allowed = ['mounts', 'env', 'privileged', 'devices'];
+ for (const key of Object.keys(raw)) {
+ if (!allowed.includes(key)) problems.push(`features.${name}.${key}: unknown key`);
+ }
+
+ const mounts = [];
+ for (const [i, mount] of (raw.mounts ?? []).entries()) {
+ if (typeof mount !== 'string' || !mount.includes(':')) {
+ problems.push(`features.${name}.mounts[${i}]: expected "host:container" or "host:container:ro"`);
+ continue;
+ }
+ mounts.push(mount);
+ }
+
+ const env = {};
+ for (const [key, value] of Object.entries(raw.env ?? {})) {
+ if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(key)) {
+ problems.push(`features.${name}.env.${key}: not a valid environment variable name`);
+ continue;
+ }
+ env[key] = String(value);
+ }
+
+ return {
+ mounts,
+ env,
+ privileged: raw.privileged === true,
+ devices: (raw.devices ?? []).map(String),
+ };
+}
+
+export async function loadWorkerConfig(explicitPath) {
+ const file = explicitPath || process.env.CONDUCTOR_WORKER_CONFIG || null;
+ let cfg = { ...DEFAULTS };
+
+ if (file) {
+ if (!fs.existsSync(file)) throw new Error(`worker config not found: ${file}`);
+ const parsed = await readConfigFile(file);
+ if (!isPlainObject(parsed)) throw new Error(`${file} must contain a mapping at the top level`);
+ cfg = { ...cfg, ...parsed };
+ }
+
+ for (const [env, key, parse] of ENV_MAP) {
+ const raw = process.env[env];
+ if (raw === undefined || raw === '') continue;
+ try {
+ cfg[key] = parse(raw);
+ } catch (e) {
+ throw new Error(`invalid value for ${env}: ${e.message}`);
+ }
+ }
+
+ const problems = [];
+
+ if (typeof cfg.conductor_url !== 'string') problems.push('conductor_url must be a string');
+ else {
+ try {
+ new URL(cfg.conductor_url);
+ } catch {
+ problems.push(`conductor_url must be an absolute URL, got ${JSON.stringify(cfg.conductor_url)}`);
+ }
+ }
+
+ if (!Array.isArray(cfg.arches)) {
+ problems.push('arches must be a list');
+ cfg.arches = [];
+ }
+ if (!Number.isInteger(cfg.concurrency) || cfg.concurrency < 1 || cfg.concurrency > 64) {
+ problems.push(`concurrency must be between 1 and 64, got ${cfg.concurrency}`);
+ }
+ if (!Number.isInteger(cfg.poll_interval) || cfg.poll_interval < 1) {
+ problems.push(`poll_interval must be at least 1 second, got ${cfg.poll_interval}`);
+ }
+
+ const features = {};
+ if (!isPlainObject(cfg.features)) {
+ problems.push('features must be a mapping of feature name to definition');
+ } else {
+ for (const [name, raw] of Object.entries(cfg.features)) {
+ if (!/^[A-Za-z0-9][A-Za-z0-9_.-]*$/.test(name)) {
+ problems.push(`features.${name}: invalid feature name`);
+ continue;
+ }
+ const feature = normalizeFeature(name, raw, problems);
+ if (feature) features[name] = feature;
+ }
+ }
+ cfg.features = features;
+
+ // The token may be inline, in a file, or in the environment. A file is
+ // preferred so it never appears in the process list or in `docker inspect`.
+ if (!cfg.token && cfg.token_file) {
+ try {
+ cfg.token = fs.readFileSync(cfg.token_file, 'utf8').trim();
+ } catch (e) {
+ problems.push(`cannot read token_file ${cfg.token_file}: ${e.message}`);
+ }
+ }
+ if (!cfg.token) problems.push('no worker token: set token, token_file, or CONDUCTOR_WORKER_TOKEN');
+
+ cfg.workspace_root = path.resolve(cfg.workspace_root || path.join(os.tmpdir(), 'conductor-work'));
+ cfg.cache_root = path.resolve(cfg.cache_root || path.join(os.tmpdir(), 'conductor-cache'));
+ cfg.conductor_url = String(cfg.conductor_url).replace(/\/+$/, '');
+
+ if (problems.length > 0) {
+ throw new Error(`invalid worker configuration:\n - ${problems.join('\n - ')}`);
+ }
+
+ return cfg;
+}
+
+export function featureNames(cfg) {
+ return Object.keys(cfg.features).sort();
+}
diff --git a/src/worker/docker.js b/src/worker/docker.js
@@ -0,0 +1,133 @@
+// src/worker/docker.js - container runtime
+//
+// Shells out to a docker compatible CLI. Podman and nerdctl accept the same
+// arguments, so the binary is configurable and nothing here is Docker
+// specific beyond the command names.
+//
+// Arguments are always passed as an array and never through a shell, since
+// image names, environment values and mount paths all originate from a
+// repository.
+
+import crypto from 'node:crypto';
+import { execFile, spawn } from 'node:child_process';
+import { promisify } from 'node:util';
+
+const execFileAsync = promisify(execFile);
+
+// Docker requires [a-zA-Z0-9][a-zA-Z0-9_.-]*, while job names contain
+// colons, commas and equals signs. A hash suffix keeps names unique after
+// the unsafe characters are folded away.
+export function containerName(prefix, id, suffix = '') {
+ const digest = crypto.createHash('sha256').update(id).digest('hex').slice(0, 8);
+ const safe = id.replace(/[^a-zA-Z0-9_.-]/g, '-').replace(/^[^a-zA-Z0-9]+/, '').slice(0, 40);
+ return [prefix, safe, digest, suffix].filter(Boolean).join('-');
+}
+
+export function createRuntime(cfg, { logger = console } = {}) {
+ const bin = cfg.docker;
+
+ async function run(args, { timeout = 120000 } = {}) {
+ try {
+ const { stdout } = await execFileAsync(bin, args, { timeout, maxBuffer: 8 * 1024 * 1024 });
+ return stdout.trim();
+ } catch (e) {
+ const detail = (e.stderr || e.message || '').toString().trim().split('\n').slice(0, 3).join('; ');
+ const err = new Error(`${bin} ${args[0]} failed: ${detail}`);
+ err.cause = e;
+ throw err;
+ }
+ }
+
+ // Turns a job and the worker's feature definitions into run arguments.
+ function buildRunArgs({ name, image, network, workspace, env = {}, mounts = [], privileged = false, devices = [], command = [], entrypoint = null, detach = false, networkAlias = null }) {
+ const args = ['run', '--rm', '--name', name];
+ if (detach) args.push('--detach');
+ else args.push('--attach', 'stdout', '--attach', 'stderr');
+
+ if (network) args.push('--network', network);
+ if (networkAlias) args.push('--network-alias', networkAlias);
+ if (privileged) args.push('--privileged');
+
+ for (const device of devices) args.push('--device', device);
+ for (const mount of mounts) args.push('--volume', mount);
+
+ if (workspace) {
+ args.push('--volume', `${workspace}:/workspace`);
+ args.push('--workdir', '/workspace');
+ }
+
+ for (const [key, value] of Object.entries(env)) {
+ args.push('--env', `${key}=${value}`);
+ }
+
+ if (entrypoint !== null) args.push('--entrypoint', entrypoint);
+ args.push(image);
+ args.push(...command);
+ return args;
+ }
+
+ return {
+ buildRunArgs,
+
+ async available() {
+ try {
+ await run(['version', '--format', '{{.Server.Version}}'], { timeout: 15000 });
+ return true;
+ } catch {
+ return false;
+ }
+ },
+
+ async createNetwork(name) {
+ await run(['network', 'create', name]);
+ return name;
+ },
+
+ async removeNetwork(name) {
+ try {
+ await run(['network', 'rm', name]);
+ } catch (e) {
+ logger.warn?.(`could not remove network ${name}: ${e.message}`);
+ }
+ },
+
+ // Services run detached; their output is not part of the job log.
+ async startService(options) {
+ return run(buildRunArgs({ ...options, detach: true }));
+ },
+
+ // Starts the job container and hands back the process. The caller reads
+ // stdout and stderr, and awaits the exit code.
+ start(options) {
+ const args = buildRunArgs({ ...options, detach: false });
+ const child = spawn(bin, args, { stdio: ['ignore', 'pipe', 'pipe'] });
+ return { child, args };
+ },
+
+ // Runs a container to completion and returns its output. Used for
+ // short housekeeping tasks rather than for jobs.
+ async runOnce(options, { timeout = 120000 } = {}) {
+ return run(buildRunArgs({ ...options, detach: false }), { timeout });
+ },
+
+ async kill(name, signal = 'KILL') {
+ try {
+ await run(['kill', '--signal', signal, name], { timeout: 30000 });
+ } catch {
+ // Already gone, which is the desired state.
+ }
+ },
+
+ async remove(name) {
+ try {
+ await run(['rm', '--force', name], { timeout: 30000 });
+ } catch {
+ // Already gone.
+ }
+ },
+
+ async pull(image) {
+ return run(['pull', image], { timeout: 30 * 60 * 1000 });
+ },
+ };
+}
diff --git a/src/worker/job.js b/src/worker/job.js
@@ -0,0 +1,251 @@
+// src/worker/job.js - running one job
+//
+// Sequence: prepare a workspace, fetch the tree, start any services on a
+// private network, run the job container, ship its output, upload
+// artifacts, report the outcome, and clean up whatever was created.
+//
+// Cleanup runs whatever happened. A worker that leaks containers or
+// networks stops being usable after a few dozen jobs.
+
+import fs from 'node:fs/promises';
+import path from 'node:path';
+import { containerName } from './docker.js';
+import { createLogStream } from './logstream.js';
+import { fetchSource } from './source.js';
+import { buildScript, SCRIPT_DIR, SCRIPT_NAME, SCRIPT_PATH } from './script.js';
+import { collectArtifacts, uploadArtifacts, shouldCollect } from './artifacts.js';
+
+// Resolves the features a job asked for into local resources.
+export function resolveFeatures(job, available) {
+ const mounts = [];
+ const env = {};
+ const devices = [];
+ let privileged = false;
+ const missing = [];
+
+ for (const name of job.requires ?? []) {
+ const feature = available[name];
+ if (!feature) {
+ missing.push(name);
+ continue;
+ }
+ mounts.push(...feature.mounts);
+ devices.push(...feature.devices);
+ Object.assign(env, feature.env);
+ if (feature.privileged) privileged = true;
+ }
+
+ return { mounts, env, devices, privileged, missing };
+}
+
+// Cache directories are bind mounted straight onto their path in the
+// workspace, which is cheaper than copying in and out around every job.
+export async function prepareCache(cfg, job) {
+ if (!job.cache || !Array.isArray(job.cache.paths)) return [];
+
+ const key = String(job.cache.key ?? 'default').replace(/[^A-Za-z0-9_.-]/g, '-').slice(0, 128);
+ const project = String(job.project_id ?? 'project').replace(/[^A-Za-z0-9_.-]/g, '-');
+ const mounts = [];
+
+ for (const relative of job.cache.paths) {
+ const normalized = relative.split('/').filter((s) => s && s !== '.' && s !== '..').join('/');
+ if (!normalized) continue;
+ const host = path.join(cfg.cache_root, project, key, normalized);
+ await fs.mkdir(host, { recursive: true });
+ mounts.push(`${host}:/workspace/${normalized}`);
+ }
+
+ return mounts;
+}
+
+export async function runJob({ cfg, client, runtime, job, logger = console }) {
+ const workspace = path.join(cfg.workspace_root, containerName('ws', job.id));
+ const network = containerName('conductor-net', job.id);
+ const jobContainer = containerName('conductor', job.id);
+
+ const log = createLogStream(client, job.endpoints.log, { masked: job.masked ?? [], logger });
+
+ const services = [];
+ let networkCreated = false;
+ let heartbeat = null;
+ let timeoutTimer = null;
+ let cancelled = false;
+ let timedOut = false;
+ let exitCode = null;
+ let failure = null;
+
+ try {
+ await fs.mkdir(workspace, { recursive: true });
+
+ log.note(`[conductor] job ${job.name} on ${cfg.name}`);
+ log.note(`[conductor] commit ${String(job.sha).slice(0, 12)} attempt ${job.attempt}`);
+ log.note(`[conductor] image ${job.image}`);
+
+ const features = resolveFeatures(job, cfg.features);
+ if (features.missing.length > 0) {
+ // The conductor should never send such a job, so this means the two
+ // sides disagree about what this worker offers.
+ throw new Error(`worker does not provide required feature(s): ${features.missing.join(', ')}`);
+ }
+
+ await fetchSource(client, job, workspace);
+
+ await fs.mkdir(path.join(workspace, SCRIPT_DIR), { recursive: true });
+ await fs.writeFile(path.join(workspace, SCRIPT_PATH), buildScript(job.script), { mode: 0o755 });
+
+ const cacheMounts = await prepareCache(cfg, job);
+
+ await runtime.createNetwork(network);
+ networkCreated = true;
+
+ for (const service of job.services ?? []) {
+ const name = containerName('conductor-svc', job.id, service.alias);
+ log.note(`[conductor] starting service ${service.image} as ${service.alias}`);
+ await runtime.startService({
+ name,
+ image: service.image,
+ network,
+ networkAlias: service.alias,
+ env: service.env ?? {},
+ entrypoint: service.entrypoint?.length ? service.entrypoint[0] : null,
+ command: service.entrypoint?.length ? [...service.entrypoint.slice(1), ...(service.command ?? [])] : (service.command ?? []),
+ });
+ services.push(name);
+ }
+
+ const { child } = runtime.start({
+ name: jobContainer,
+ image: job.image,
+ network,
+ workspace,
+ env: { ...features.env, ...job.env },
+ mounts: [...features.mounts, ...cacheMounts],
+ devices: features.devices,
+ privileged: features.privileged,
+ entrypoint: cfg.shell,
+ command: [`/workspace/${SCRIPT_PATH}`],
+ });
+
+ child.stdout.on('data', (chunk) => log.write(chunk));
+ child.stderr.on('data', (chunk) => log.write(chunk));
+
+ // Report in, and find out whether the run was cancelled underneath us.
+ const beatEvery = Math.max(5, Number(job.heartbeat_interval) || 30) * 1000;
+ heartbeat = setInterval(async () => {
+ try {
+ const result = await client.heartbeat(job.endpoints.heartbeat);
+ if (result.cancelled && !cancelled) {
+ cancelled = true;
+ log.note('[conductor] cancelled, stopping the container');
+ await runtime.kill(jobContainer);
+ }
+ } catch (e) {
+ logger.warn?.(`heartbeat failed for ${job.name}: ${e.message}`);
+ }
+ }, beatEvery);
+ heartbeat.unref?.();
+
+ const limit = (Number(job.timeout) || cfg.default_timeout) * 1000;
+ timeoutTimer = setTimeout(async () => {
+ timedOut = true;
+ log.note(`[conductor] timed out after ${Math.round(limit / 1000)}s, stopping the container`);
+ await runtime.kill(jobContainer);
+ }, limit);
+ timeoutTimer.unref?.();
+
+ exitCode = await new Promise((resolve, reject) => {
+ child.on('error', reject);
+ child.on('close', resolve);
+ });
+ } catch (e) {
+ failure = e;
+ log.note(`[conductor] ${e.message}`);
+ } finally {
+ if (heartbeat) clearInterval(heartbeat);
+ if (timeoutTimer) clearTimeout(timeoutTimer);
+ }
+
+ const success = failure === null && exitCode === 0 && !cancelled && !timedOut;
+
+ // Artifacts are collected before teardown, and even after a failure when
+ // the pipeline asked for that.
+ if (job.artifacts && shouldCollect(job.artifacts.when, success)) {
+ try {
+ const files = await collectArtifacts(workspace, job.artifacts.paths, { logger });
+ if (files.length > 0) {
+ log.note(`[conductor] uploading ${files.length} artifact(s)`);
+ await uploadArtifacts(client, job.endpoints.artifact, files, { logger });
+ } else {
+ log.note('[conductor] no files matched the artifact patterns');
+ }
+ } catch (e) {
+ log.note(`[conductor] artifact collection failed: ${e.message}`);
+ logger.warn?.(`artifact collection failed for ${job.name}: ${e.message}`);
+ }
+ }
+
+ const error = failure
+ ? failure.message
+ : timedOut
+ ? 'job timed out'
+ : cancelled
+ ? 'job cancelled'
+ : exitCode === 0
+ ? null
+ : `job exited ${exitCode}`;
+
+ if (error) log.note(`[conductor] ${error}`);
+ else log.note('[conductor] done');
+
+ await log.close();
+
+ // Teardown first, so a slow conductor cannot hold containers open.
+ await cleanup();
+
+ try {
+ await client.done(job.endpoints.done, { success, exitCode, error });
+ } catch (e) {
+ logger.error?.(`could not report completion of ${job.name}: ${e.message}`);
+ }
+
+ return { success, exitCode, error, cancelled, timedOut };
+
+ async function cleanup() {
+ for (const name of services) await runtime.remove(name);
+ await runtime.remove(jobContainer);
+ if (networkCreated) await runtime.removeNetwork(network);
+ await removeWorkspace();
+ }
+
+ // Most images run as root, so files the job created in the bind mounted
+ // workspace are owned by root while the worker usually is not. Removing
+ // them from the host then fails with EACCES. The directory itself belongs
+ // to the worker, so emptying it from inside a container as root and then
+ // removing the empty directory works without forcing every job image to
+ // run as the worker's uid.
+ async function removeWorkspace() {
+ try {
+ await fs.rm(workspace, { recursive: true, force: true });
+ return;
+ } catch (e) {
+ if (e.code !== 'EACCES' && e.code !== 'EPERM' && e.code !== 'ENOTEMPTY') {
+ logger.warn?.(`could not remove workspace ${workspace}: ${e.message}`);
+ return;
+ }
+ }
+
+ try {
+ // The job image is already present locally, so nothing is pulled.
+ await runtime.runOnce({
+ name: containerName('conductor-clean', job.id),
+ image: job.image,
+ workspace,
+ entrypoint: cfg.shell,
+ command: ['-c', 'rm -rf /workspace/..?* /workspace/.[!.]* /workspace/* 2>/dev/null || true'],
+ });
+ await fs.rm(workspace, { recursive: true, force: true });
+ } catch (e) {
+ logger.warn?.(`could not remove workspace ${workspace}: ${e.message}`);
+ }
+ }
+}
diff --git a/src/worker/logstream.js b/src/worker/logstream.js
@@ -0,0 +1,162 @@
+// src/worker/logstream.js - batching log shipper with masking
+//
+// Job output arrives as many small writes; shipping each one would be a
+// request per line. Output is accumulated and flushed on a timer, or as soon
+// as it grows past a threshold.
+//
+// The offset the conductor has accepted is tracked locally, so a retry after
+// a dropped connection resends from the right place rather than duplicating
+// or losing output. A 409 carries the conductor's view of the offset and
+// wins, since it is the side holding the file.
+//
+// Masking happens here, before anything leaves the host. Two details make it
+// actually work:
+//
+// Masking is applied when a batch is assembled, not as each write
+// arrives, and the last few characters are held back until more output
+// appears. A secret written across two chunks is therefore still matched.
+// Masking each write on its own would defeat the hold back entirely.
+//
+// While masking is active the stream is decoded as UTF-8 with a
+// StringDecoder, so a multi byte character split across writes is not
+// mangled. With no secrets configured the bytes are passed through
+// untouched and the stream stays binary safe.
+
+import { StringDecoder } from 'node:string_decoder';
+
+const FLUSH_BYTES = 16 * 1024;
+const FLUSH_MS = 1000;
+const REDACTION = '[masked]';
+
+// Below this length a value is too common to redact usefully, and would
+// turn ordinary output into noise.
+const MIN_SECRET_LENGTH = 4;
+
+export function createLogStream(client, url, { masked = [], flushMs = FLUSH_MS, logger = console } = {}) {
+ // Longest first, so an overlapping shorter secret cannot partially
+ // reveal a longer one.
+ const secrets = [...new Set(masked.filter((s) => typeof s === 'string' && s.length >= MIN_SECRET_LENGTH))]
+ .sort((a, b) => b.length - a.length);
+ const masking = secrets.length > 0;
+ const holdBack = masking ? Math.max(...secrets.map((s) => s.length)) - 1 : 0;
+
+ const decoder = masking ? new StringDecoder('utf8') : null;
+ let text = '';
+ let bytes = Buffer.alloc(0);
+ let outbox = Buffer.alloc(0);
+
+ let offset = 0;
+ let timer = null;
+ let flushing = null;
+ let ended = false;
+ let failed = null;
+
+ function mask(value) {
+ let out = value;
+ for (const secret of secrets) out = out.split(secret).join(REDACTION);
+ return out;
+ }
+
+ function buffered() {
+ return outbox.length + (masking ? text.length : bytes.length);
+ }
+
+ // Moves everything that is safe to send into the outbox, masking on the
+ // way. Anything within holdBack of the end waits for more input, unless
+ // this is the final flush.
+ function assemble(final) {
+ if (!masking) {
+ if (bytes.length === 0) return;
+ outbox = Buffer.concat([outbox, bytes]);
+ bytes = Buffer.alloc(0);
+ return;
+ }
+
+ if (final) text += decoder.end();
+ const keep = final ? 0 : Math.min(holdBack, text.length);
+ const take = text.slice(0, text.length - keep);
+ text = text.slice(text.length - keep);
+ if (take.length === 0) return;
+ outbox = Buffer.concat([outbox, Buffer.from(mask(take), 'utf8')]);
+ }
+
+ function schedule() {
+ if (timer !== null || ended) return;
+ timer = setTimeout(() => { timer = null; void flush(); }, flushMs);
+ timer.unref?.();
+ }
+
+ async function flush(final = false) {
+ if (timer !== null) {
+ clearTimeout(timer);
+ timer = null;
+ }
+ // Serialize, so two flushes cannot race on the offset.
+ while (flushing) await flushing;
+
+ assemble(final);
+ if (outbox.length === 0) return;
+
+ flushing = (async () => {
+ while (outbox.length > 0) {
+ const chunk = outbox;
+ outbox = Buffer.alloc(0);
+ try {
+ const result = await client.appendLog(url, chunk, offset);
+ if (result.conflict) {
+ // The conductor knows what it already has; drop that much and
+ // resend the remainder.
+ const skip = result.expected - offset;
+ offset = result.expected;
+ const remainder = skip > 0 ? chunk.subarray(Math.min(skip, chunk.length)) : chunk;
+ outbox = Buffer.concat([remainder, outbox]);
+ continue;
+ }
+ offset = result.size;
+ } catch (e) {
+ // Losing log output must not fail an otherwise good job.
+ failed = e;
+ logger.warn?.(`log shipping failed, dropping ${chunk.length} bytes: ${e.message}`);
+ }
+ }
+ })();
+
+ try {
+ await flushing;
+ } finally {
+ flushing = null;
+ }
+ }
+
+ return {
+ write(chunk) {
+ if (ended) return;
+ const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk), 'utf8');
+ if (masking) text += decoder.write(buffer);
+ else bytes = Buffer.concat([bytes, buffer]);
+
+ if (buffered() >= FLUSH_BYTES) void flush();
+ else schedule();
+ },
+
+ // A line from the worker rather than the job, so the log can explain a
+ // timeout, a cancellation or a setup failure.
+ note(line) {
+ this.write(`${line}\n`);
+ },
+
+ async close() {
+ if (ended) return;
+ await flush(true);
+ ended = true;
+ },
+
+ get offset() {
+ return offset;
+ },
+
+ get error() {
+ return failed;
+ },
+ };
+}
diff --git a/src/worker/script.js b/src/worker/script.js
@@ -0,0 +1,29 @@
+// src/worker/script.js - turning a job's commands into a shell script
+//
+// The script is written into the workspace and run by path rather than piped
+// to the shell's stdin, so that a command which itself reads stdin cannot
+// swallow the rest of the script.
+
+// Single quoting is the only form that is safe for arbitrary text in POSIX
+// sh: everything inside is literal, and an embedded quote is closed,
+// escaped and reopened.
+export function shellQuote(value) {
+ return `'${String(value).replace(/'/g, "'\\''")}'`;
+}
+
+// Each command is echoed before it runs, so a reader can tell which one
+// produced what. set -e stops at the first failure, and the exit code of
+// that command becomes the exit code of the script.
+export function buildScript(commands) {
+ const lines = ['#!/bin/sh', 'set -e', ''];
+ for (const command of commands) {
+ lines.push(`printf '%s\\n' ${shellQuote(`$ ${command}`)}`);
+ lines.push(command);
+ lines.push('');
+ }
+ return lines.join('\n');
+}
+
+export const SCRIPT_DIR = '.conductor';
+export const SCRIPT_NAME = 'script.sh';
+export const SCRIPT_PATH = `${SCRIPT_DIR}/${SCRIPT_NAME}`;
diff --git a/src/worker/source.js b/src/worker/source.js
@@ -0,0 +1,77 @@
+// src/worker/source.js - getting the tree into the workspace
+//
+// Two modes, chosen per project by the conductor:
+//
+// archive download a tarball of the exact commit from the conductor and
+// unpack it. The worker needs no repository credentials and can
+// reach no ref other than the one it was given work for.
+// clone git clone the repository directly, for repositories large
+// enough that a tarball per job is wasteful.
+//
+// Unpacking shells out to the host tar. Both GNU and BSD tar accept these
+// arguments; availability is checked once at startup.
+
+import { spawn, execFile } from 'node:child_process';
+import { promisify } from 'node:util';
+import { Readable } from 'node:stream';
+import { pipeline } from 'node:stream/promises';
+
+const execFileAsync = promisify(execFile);
+
+export async function hasTar() {
+ try {
+ await execFileAsync('tar', ['--version'], { timeout: 10000 });
+ return true;
+ } catch {
+ return false;
+ }
+}
+
+export async function hasGit() {
+ try {
+ await execFileAsync('git', ['--version'], { timeout: 10000 });
+ return true;
+ } catch {
+ return false;
+ }
+}
+
+// Streams a gzipped tar into a directory without staging it on disk.
+export async function extractTarGz(stream, directory) {
+ const child = spawn('tar', ['-xzf', '-', '-C', directory, '--no-same-owner'], {
+ stdio: ['pipe', 'ignore', 'pipe'],
+ });
+
+ let stderr = '';
+ child.stderr.on('data', (chunk) => { stderr += chunk.toString().slice(0, 4096); });
+
+ const exited = new Promise((resolve, reject) => {
+ child.on('error', reject);
+ child.on('close', (code) => {
+ if (code === 0) resolve();
+ else reject(new Error(`tar exited ${code}: ${stderr.trim()}`));
+ });
+ });
+
+ const source = stream instanceof Readable ? stream : Readable.fromWeb(stream);
+ await Promise.all([pipeline(source, child.stdin), exited]);
+}
+
+export async function fetchSource(client, job, workspace, { timeout = 600000 } = {}) {
+ const source = job.source ?? {};
+
+ if (source.mode === 'clone') {
+ const env = { ...process.env, GIT_TERMINAL_PROMPT: '0', GIT_ASKPASS: '' };
+ await execFileAsync('git', ['clone', '--quiet', '--no-checkout', '--', source.repo_url, workspace], { timeout, env });
+ await execFileAsync('git', ['-C', workspace, 'checkout', '--quiet', source.sha], { timeout, env });
+ return { mode: 'clone' };
+ }
+
+ if (source.mode === 'archive') {
+ const body = await client.source(source.url);
+ await extractTarGz(body, workspace);
+ return { mode: 'archive' };
+ }
+
+ throw new Error(`unsupported source mode: ${JSON.stringify(source.mode)}`);
+}
diff --git a/test/conductor.test.js b/test/conductor.test.js
@@ -224,6 +224,40 @@ test('a claimed job carries everything the worker needs', async () => {
assert.equal(job.source.mode, 'archive');
assert.match(job.source.url, /^http:\/\/conductor\.test\/api\/workers\/jobs\//);
assert.ok(job.endpoints.log.endsWith('/log'));
+
+ // The worker resolves features and cache keys from these two, so a
+ // missing field silently disables both.
+ assert.deepEqual(job.requires, []);
+ assert.equal(job.project_id, 'demo');
+ assert.ok(Number.isInteger(job.heartbeat_interval) && job.heartbeat_interval > 0);
+ assert.deepEqual(job.masked, []);
+ });
+});
+
+test('a job that requires a feature reports it to the worker', async () => {
+ await withHarness({}, async (h) => {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const roots = await drain(h, { arches: 'x86_64,aarch64' });
+ for (const job of Object.values(roots)) await finish(h, job.id, { success: true });
+
+ const pkg = (await drain(h, { arches: 'x86_64', features: 'sign-key' }))['package:arch=x86_64,pkg=musl'];
+ assert.ok(pkg);
+ assert.deepEqual(pkg.requires, ['sign-key']);
+ });
+});
+
+test('the standard conductor variables are present in the job environment', async () => {
+ await withHarness({}, async (h) => {
+ const run = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().run_id;
+ const job = await claimNamed(h, 'lint', {});
+
+ assert.equal(job.env.CONDUCTOR_PROJECT, 'demo');
+ assert.equal(job.env.CONDUCTOR_RUN_ID, run);
+ assert.equal(job.env.CONDUCTOR_RUN_NUMBER, '1');
+ assert.equal(job.env.CONDUCTOR_JOB, 'lint');
+ assert.equal(job.env.CONDUCTOR_SHA, h.sha);
+ assert.equal(job.env.CONDUCTOR_REF, 'refs/heads/main');
+ assert.equal(job.env.CONDUCTOR_ATTEMPT, '1');
});
});
diff --git a/test/helpers/harness.js b/test/helpers/harness.js
@@ -6,6 +6,7 @@
import fs from 'node:fs/promises';
import os from 'node:os';
+import net from 'node:net';
import path from 'node:path';
import crypto from 'node:crypto';
import { execFile } from 'node:child_process';
@@ -65,8 +66,10 @@ export async function startHarness(options = {}) {
const configFile = path.join(root, 'conductor.yaml');
await fs.writeFile(configFile, [
'server:',
- ' port: 0',
- ' public_url: http://conductor.test',
+ // A real port and a matching public_url are needed whenever something
+ // outside the process, such as a worker, has to reach back in.
+ ` port: ${options.port ?? 0}`,
+ ` public_url: ${options.publicUrl ?? 'http://conductor.test'}`,
'database:',
' path: ./state/conductor.db',
'storage:',
@@ -154,7 +157,9 @@ export async function startHarness(options = {}) {
async stop() {
await app.close();
await services.db.close();
- await fs.rm(root, { recursive: true, force: true });
+ // Best effort: a container that ran as root may have left files the
+ // test process cannot remove, and that must not mask a real failure.
+ await fs.rm(root, { recursive: true, force: true }).catch(() => {});
},
};
}
@@ -162,3 +167,17 @@ export async function startHarness(options = {}) {
function silentLogger() {
return { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} };
}
+
+// Asks the kernel for a free port and releases it. Only needed because the
+// public_url has to be known before the server is built.
+export async function freePort() {
+ return new Promise((resolve, reject) => {
+ const server = net.createServer();
+ server.unref();
+ server.on('error', reject);
+ server.listen(0, '127.0.0.1', () => {
+ const { port } = server.address();
+ server.close(() => resolve(port));
+ });
+ });
+}
diff --git a/test/worker-docker.test.js b/test/worker-docker.test.js
@@ -0,0 +1,340 @@
+// test/worker-docker.test.js - the whole loop, for real
+//
+// Runs actual containers against an actual conductor: trigger, claim, fetch
+// source, run, stream logs, upload artifacts, report, and clean up.
+// Skipped when no container runtime is available.
+//
+// The conductor is bound to a real port here rather than driven through
+// inject, because the worker is a separate process talking over HTTP.
+
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { execFile } from 'node:child_process';
+import { promisify } from 'node:util';
+import { startHarness, freePort } from './helpers/harness.js';
+import { createClient } from '../src/worker/client.js';
+import { createRuntime, containerName } from '../src/worker/docker.js';
+import { runJob } from '../src/worker/job.js';
+
+const execFileAsync = promisify(execFile);
+
+const IMAGE = process.env.CONDUCTOR_TEST_IMAGE || 'alpine:3';
+
+async function dockerUsable() {
+ if (process.env.CONDUCTOR_TEST_NO_DOCKER) return false;
+ try {
+ await execFileAsync('docker', ['version', '--format', '{{.Server.Version}}'], { timeout: 15000 });
+ await execFileAsync('docker', ['image', 'inspect', IMAGE], { timeout: 15000 })
+ .catch(() => execFileAsync('docker', ['pull', IMAGE], { timeout: 300000 }));
+ return true;
+ } catch {
+ return false;
+ }
+}
+
+const available = await dockerUsable();
+const opts = { skip: available ? false : 'docker is not available' };
+
+// Starts the harness on a real port so the worker can reach it over HTTP.
+// The port has to be settled before the server is built, because the job
+// payload embeds absolute callback URLs derived from public_url.
+async function listening(options = {}) {
+ const port = await freePort();
+ const url = `http://127.0.0.1:${port}`;
+ const h = await startHarness({ ...options, port, publicUrl: url });
+ await h.app.listen({ port, host: '127.0.0.1' });
+
+ const workerCfg = {
+ conductor_url: url,
+ name: 'docker-test-worker',
+ token: h.worker.token,
+ arches: [],
+ features: {},
+ concurrency: 1,
+ poll_interval: 1,
+ docker: 'docker',
+ shell: 'sh',
+ workspace_root: `${h.root}/work`,
+ cache_root: `${h.root}/cache`,
+ default_timeout: 300,
+ ...(options?.worker ?? {}),
+ };
+
+ return {
+ ...h,
+ url,
+ workerCfg,
+ client: createClient(workerCfg),
+ runtime: createRuntime(workerCfg, { logger: quiet }),
+ };
+}
+
+const quiet = { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} };
+
+async function logOf(h, jobId) {
+ const res = await h.app.inject({ method: 'GET', url: `/api/jobs/${encodeURIComponent(jobId)}/log` });
+ return res.body;
+}
+
+// Anchored, because a substring filter would also match unrelated
+// containers that happen to share the prefix.
+async function exists(kind, name) {
+ const args = kind === 'network'
+ ? ['network', 'ls', '--filter', `name=^${name}$`, '--format', '{{.Name}}']
+ : ['ps', '-a', '--filter', `name=^${name}$`, '--format', '{{.Names}}'];
+ const { stdout } = await execFileAsync('docker', args);
+ return stdout.trim().length > 0;
+}
+
+test('a job runs in a container and its output reaches the conductor', opts, async () => {
+ const pipeline = `
+version: 1
+jobs:
+ hello:
+ image: ${IMAGE}
+ script:
+ - echo "hello from the job"
+ - echo "commit is $CONDUCTOR_SHA"
+ - echo "job is $CONDUCTOR_JOB"
+ - cat README.md
+`;
+ const h = await listening({ pipeline });
+ try {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const job = await h.client.poll({ arches: [], features: [], name: 'test' });
+ assert.ok(job, 'expected a job');
+
+ const result = await runJob({ cfg: h.workerCfg, client: h.client, runtime: h.runtime, job, logger: quiet });
+ assert.equal(result.success, true, `job failed: ${result.error}`);
+ assert.equal(result.exitCode, 0);
+
+ const log = await logOf(h, job.id);
+ assert.match(log, /\$ echo "hello from the job"/, 'each command should be echoed');
+ assert.match(log, /hello from the job/);
+ assert.match(log, new RegExp(`commit is ${h.sha}`), 'CONDUCTOR_SHA should be set');
+ assert.match(log, /job is hello/);
+ // Proves the source tarball was fetched and unpacked.
+ assert.match(log, /test repository/);
+
+ const detail = await h.app.inject({ method: 'GET', url: `/api/runs/${job.run_id}` });
+ assert.equal(detail.json().run.state, 'success');
+ } finally {
+ await h.stop();
+ }
+});
+
+test('a failing command fails the job and preserves the log', opts, async () => {
+ const pipeline = `
+version: 1
+jobs:
+ boom:
+ image: ${IMAGE}
+ script:
+ - echo before
+ - exit 3
+ - echo after
+`;
+ const h = await listening({ pipeline });
+ try {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const job = await h.client.poll({ arches: [], features: [], name: 'test' });
+
+ const result = await runJob({ cfg: h.workerCfg, client: h.client, runtime: h.runtime, job, logger: quiet });
+ assert.equal(result.success, false);
+ assert.equal(result.exitCode, 3);
+
+ const log = await logOf(h, job.id);
+ assert.match(log, /before/);
+ // set -e must stop the script at the failure.
+ assert.ok(!log.includes('after'), 'commands after a failure must not run');
+
+ const detail = await h.app.inject({ method: 'GET', url: `/api/runs/${job.run_id}` });
+ assert.equal(detail.json().run.state, 'failed');
+ } finally {
+ await h.stop();
+ }
+});
+
+test('artifacts produced in the container are uploaded and downloadable', opts, async () => {
+ const pipeline = `
+version: 1
+jobs:
+ produce:
+ image: ${IMAGE}
+ script:
+ - mkdir -p dist/nested
+ - printf 'binary-content' > dist/app.bin
+ - printf 'deep' > dist/nested/deep.txt
+ artifacts:
+ paths: [dist/**]
+`;
+ const h = await listening({ pipeline });
+ try {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const job = await h.client.poll({ arches: [], features: [], name: 'test' });
+
+ const result = await runJob({ cfg: h.workerCfg, client: h.client, runtime: h.runtime, job, logger: quiet });
+ assert.equal(result.success, true, `job failed: ${result.error}`);
+
+ const detail = await h.app.inject({ method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}` });
+ const artifacts = detail.json().artifacts;
+ const paths = artifacts.map((a) => a.path).sort();
+ assert.deepEqual(paths, ['dist/app.bin', 'dist/nested/deep.txt']);
+
+ const binary = artifacts.find((a) => a.path === 'dist/app.bin');
+ const download = await h.app.inject({ method: 'GET', url: `/api/artifacts/${binary.id}` });
+ assert.equal(download.body, 'binary-content');
+ } finally {
+ await h.stop();
+ }
+});
+
+test('a worker feature injects a mount and environment the conductor never sees', opts, async () => {
+ const pipeline = `
+version: 1
+jobs:
+ signed:
+ image: ${IMAGE}
+ requires: [sign-key]
+ script:
+ - echo "key path is $SIGN_KEY"
+ - cat "$SIGN_KEY"
+`;
+ const h = await listening({
+ pipeline,
+ worker: {
+ features: {
+ 'sign-key': {
+ mounts: [],
+ env: { SIGN_KEY: '/keys/build.key' },
+ devices: [],
+ privileged: false,
+ },
+ },
+ },
+ });
+
+ try {
+ // The key lives on the worker host and is mounted in by the feature.
+ const fs = await import('node:fs/promises');
+ await fs.mkdir(`${h.root}/keys`, { recursive: true });
+ await fs.writeFile(`${h.root}/keys/build.key`, 'PRIVATE-KEY-MATERIAL');
+ h.workerCfg.features['sign-key'].mounts = [`${h.root}/keys/build.key:/keys/build.key:ro`];
+
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+
+ // Without the feature the job is not offered at all.
+ const denied = await createClient({ ...h.workerCfg, token: h.worker.token })
+ .poll({ arches: [], features: [], name: 'test' });
+ assert.equal(denied, null, 'a worker lacking the feature must not be offered the job');
+
+ const job = await h.client.poll({ arches: [], features: ['sign-key'], name: 'test' });
+ assert.ok(job, 'expected the job once the feature is advertised');
+
+ const result = await runJob({ cfg: h.workerCfg, client: h.client, runtime: h.runtime, job, logger: quiet });
+ assert.equal(result.success, true, `job failed: ${result.error}`);
+
+ const log = await logOf(h, job.id);
+ assert.match(log, /key path is \/keys\/build\.key/);
+ assert.match(log, /PRIVATE-KEY-MATERIAL/);
+
+ // The conductor stored the requirement, never the key itself.
+ const stored = await h.services.db.get(
+ 'SELECT requires, spec FROM jobs WHERE id = {id}', { id: job.id }
+ );
+ assert.equal(stored.requires, '["sign-key"]');
+ assert.ok(!stored.spec.includes('PRIVATE-KEY-MATERIAL'));
+ } finally {
+ await h.stop();
+ }
+});
+
+test('a service container is reachable from the job by its alias', opts, async () => {
+ const pipeline = `
+version: 1
+jobs:
+ talks:
+ image: ${IMAGE}
+ services:
+ - image: ${IMAGE}
+ alias: sidecar
+ command: [sleep, '60']
+ script:
+ - ping -c1 -W2 sidecar
+`;
+ const h = await listening({ pipeline });
+ try {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const job = await h.client.poll({ arches: [], features: [], name: 'test' });
+
+ const result = await runJob({ cfg: h.workerCfg, client: h.client, runtime: h.runtime, job, logger: quiet });
+ const log = await logOf(h, job.id);
+ assert.equal(result.success, true, `job failed: ${result.error}\n${log}`);
+ assert.match(log, /1 packets received|1 received/);
+ } finally {
+ await h.stop();
+ }
+});
+
+test('a job that exceeds its timeout is stopped', opts, async () => {
+ const pipeline = `
+version: 1
+jobs:
+ slow:
+ image: ${IMAGE}
+ timeout: 2
+ script:
+ - echo starting
+ - sleep 60
+`;
+ const h = await listening({ pipeline });
+ try {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const job = await h.client.poll({ arches: [], features: [], name: 'test' });
+
+ const started = Date.now();
+ const result = await runJob({ cfg: h.workerCfg, client: h.client, runtime: h.runtime, job, logger: quiet });
+ const elapsed = (Date.now() - started) / 1000;
+
+ assert.equal(result.success, false);
+ assert.equal(result.timedOut, true);
+ assert.ok(elapsed < 45, `should have been stopped promptly, took ${elapsed}s`);
+ assert.match(await logOf(h, job.id), /timed out/);
+ } finally {
+ await h.stop();
+ }
+});
+
+test('containers, networks and workspaces are cleaned up', opts, async () => {
+ const pipeline = `
+version: 1
+jobs:
+ tidy:
+ image: ${IMAGE}
+ services:
+ - image: ${IMAGE}
+ alias: sidecar
+ command: [sleep, '60']
+ script:
+ - echo work
+`;
+ const h = await listening({ pipeline });
+ try {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const job = await h.client.poll({ arches: [], features: [], name: 'test' });
+ await runJob({ cfg: h.workerCfg, client: h.client, runtime: h.runtime, job, logger: quiet });
+
+ assert.equal(await exists('container', containerName('conductor', job.id)), false,
+ 'the job container should be gone');
+ assert.equal(await exists('container', containerName('conductor-svc', job.id, 'sidecar')), false,
+ 'the service container should be gone');
+ assert.equal(await exists('network', containerName('conductor-net', job.id)), false,
+ 'the job network should be gone');
+
+ const fs = await import('node:fs/promises');
+ const left = await fs.readdir(`${h.root}/work`).catch(() => []);
+ assert.deepEqual(left, [], 'the workspace should be removed');
+ } finally {
+ await h.stop();
+ }
+});
diff --git a/test/worker.test.js b/test/worker.test.js
@@ -0,0 +1,319 @@
+// test/worker.test.js - worker units that need no container runtime
+//
+// The full loop against real Docker lives in worker-docker.test.js.
+
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import fs from 'node:fs/promises';
+import os from 'node:os';
+import path from 'node:path';
+import { loadWorkerConfig, featureNames } from '../src/worker/config.js';
+import { buildScript, shellQuote } from '../src/worker/script.js';
+import { containerName, createRuntime } from '../src/worker/docker.js';
+import { createLogStream } from '../src/worker/logstream.js';
+import { resolveFeatures, prepareCache } from '../src/worker/job.js';
+import { collectArtifacts, shouldCollect } from '../src/worker/artifacts.js';
+
+async function tempDir(prefix = 'conductor-worker-') {
+ return fs.mkdtemp(path.join(os.tmpdir(), prefix));
+}
+
+async function withConfigEnv(env, fn) {
+ const saved = {};
+ for (const key of Object.keys(process.env)) {
+ if (key.startsWith('CONDUCTOR_')) {
+ saved[key] = process.env[key];
+ delete process.env[key];
+ }
+ }
+ Object.assign(process.env, env);
+ try {
+ return await fn();
+ } finally {
+ for (const key of Object.keys(process.env)) {
+ if (key.startsWith('CONDUCTOR_')) delete process.env[key];
+ }
+ Object.assign(process.env, saved);
+ }
+}
+
+test('worker config loads from json and reports features', async () => {
+ const dir = await tempDir();
+ const file = path.join(dir, 'worker.json');
+ await fs.writeFile(file, JSON.stringify({
+ conductor_url: 'https://ci.example.com/',
+ name: 'pi',
+ token: 'abc',
+ arches: ['aarch64'],
+ features: {
+ dind: { privileged: true },
+ 'sign-key': { mounts: ['/srv/k:/keys/k:ro'], env: { KEY: '/keys/k' } },
+ },
+ }));
+
+ const cfg = await withConfigEnv({}, () => loadWorkerConfig(file));
+ assert.equal(cfg.conductor_url, 'https://ci.example.com');
+ assert.deepEqual(featureNames(cfg), ['dind', 'sign-key']);
+ assert.equal(cfg.features.dind.privileged, true);
+ assert.deepEqual(cfg.features['sign-key'].mounts, ['/srv/k:/keys/k:ro']);
+ await fs.rm(dir, { recursive: true, force: true });
+});
+
+test('worker config reads the token from a file', async () => {
+ const dir = await tempDir();
+ await fs.writeFile(path.join(dir, 'token'), 'secret-token\n');
+ const file = path.join(dir, 'worker.json');
+ await fs.writeFile(file, JSON.stringify({ token_file: path.join(dir, 'token') }));
+
+ const cfg = await withConfigEnv({}, () => loadWorkerConfig(file));
+ assert.equal(cfg.token, 'secret-token');
+ await fs.rm(dir, { recursive: true, force: true });
+});
+
+test('environment overrides the worker config file', async () => {
+ const dir = await tempDir();
+ const file = path.join(dir, 'worker.json');
+ await fs.writeFile(file, JSON.stringify({ token: 'x', arches: ['x86_64'], concurrency: 1 }));
+
+ const cfg = await withConfigEnv({
+ CONDUCTOR_WORKER_ARCHES: 'aarch64,riscv64',
+ CONDUCTOR_WORKER_CONCURRENCY: '4',
+ }, () => loadWorkerConfig(file));
+
+ assert.deepEqual(cfg.arches, ['aarch64', 'riscv64']);
+ assert.equal(cfg.concurrency, 4);
+ await fs.rm(dir, { recursive: true, force: true });
+});
+
+test('worker config without a token is rejected', async () => {
+ const dir = await tempDir();
+ const file = path.join(dir, 'worker.json');
+ await fs.writeFile(file, JSON.stringify({ conductor_url: 'http://x' }));
+ await assert.rejects(withConfigEnv({}, () => loadWorkerConfig(file)), /no worker token/);
+ await fs.rm(dir, { recursive: true, force: true });
+});
+
+test('a malformed feature is reported with its path', async () => {
+ const dir = await tempDir();
+ const file = path.join(dir, 'worker.json');
+ await fs.writeFile(file, JSON.stringify({
+ token: 'x',
+ features: { bad: { mounts: ['no-colon'], nonsense: 1 } },
+ }));
+ // Both problems are reported together; the order between them is not
+ // part of the contract.
+ await assert.rejects(withConfigEnv({}, () => loadWorkerConfig(file)), (e) => {
+ assert.match(e.message, /features\.bad\.mounts\[0\]/);
+ assert.match(e.message, /features\.bad\.nonsense/);
+ return true;
+ });
+ await fs.rm(dir, { recursive: true, force: true });
+});
+
+test('shell quoting survives quotes and metacharacters', () => {
+ assert.equal(shellQuote('simple'), "'simple'");
+ assert.equal(shellQuote("it's"), "'it'\\''s'");
+ assert.equal(shellQuote('$(rm -rf /)'), "'$(rm -rf /)'");
+});
+
+test('a generated script echoes each command and stops on failure', () => {
+ const script = buildScript(['npm ci', "echo 'hi'"]);
+ assert.match(script, /^#!\/bin\/sh\nset -e\n/);
+ assert.ok(script.includes("printf '%s\\n' '$ npm ci'"));
+ assert.ok(script.includes('npm ci\n'));
+ // The echo of a command containing quotes stays a single argument.
+ assert.ok(script.includes("printf '%s\\n' '$ echo '\\''hi'\\'''"));
+});
+
+test('container names are valid for docker and stay unique', () => {
+ const a = containerName('conductor', 'run1:package:arch=x86_64,pkg=musl');
+ const b = containerName('conductor', 'run1:package:arch=x86_64,pkg=busybox');
+ assert.match(a, /^[a-zA-Z0-9][a-zA-Z0-9_.-]*$/);
+ assert.notEqual(a, b);
+ // Distinct jobs that differ only past the truncation point still differ.
+ const long = 'x'.repeat(200);
+ assert.notEqual(containerName('c', `${long}a`), containerName('c', `${long}b`));
+});
+
+test('run arguments carry mounts, env, network and privilege', () => {
+ const runtime = createRuntime({ docker: 'docker', shell: 'sh' });
+ const args = runtime.buildRunArgs({
+ name: 'job1',
+ image: 'alpine:3',
+ network: 'net1',
+ workspace: '/tmp/ws',
+ env: { FOO: 'bar' },
+ mounts: ['/srv/k:/keys/k:ro'],
+ privileged: true,
+ entrypoint: 'sh',
+ command: ['/workspace/.conductor/script.sh'],
+ });
+
+ assert.equal(args[0], 'run');
+ assert.ok(args.includes('--rm'));
+ assert.deepEqual(args.slice(args.indexOf('--network'), args.indexOf('--network') + 2), ['--network', 'net1']);
+ assert.ok(args.includes('--privileged'));
+ assert.ok(args.includes('/srv/k:/keys/k:ro'));
+ assert.ok(args.includes('/tmp/ws:/workspace'));
+ assert.ok(args.includes('FOO=bar'));
+ // The image must come before its command.
+ assert.ok(args.indexOf('alpine:3') < args.indexOf('/workspace/.conductor/script.sh'));
+});
+
+test('features resolve into mounts, env and privilege', () => {
+ const available = {
+ dind: { mounts: [], env: {}, devices: [], privileged: true },
+ 'sign-key': { mounts: ['/srv/k:/keys/k:ro'], env: { KEY: '/keys/k' }, devices: [], privileged: false },
+ };
+ const resolved = resolveFeatures({ requires: ['sign-key', 'dind'] }, available);
+ assert.deepEqual(resolved.mounts, ['/srv/k:/keys/k:ro']);
+ assert.deepEqual(resolved.env, { KEY: '/keys/k' });
+ assert.equal(resolved.privileged, true);
+ assert.deepEqual(resolved.missing, []);
+});
+
+test('an unknown required feature is reported rather than ignored', () => {
+ const resolved = resolveFeatures({ requires: ['nope'] }, {});
+ assert.deepEqual(resolved.missing, ['nope']);
+});
+
+test('cache paths become bind mounts and are created', async () => {
+ const dir = await tempDir();
+ const mounts = await prepareCache(
+ { cache_root: dir },
+ { project_id: 'demo', cache: { key: 'npm', paths: ['.npm', '../escape/attempt'] } }
+ );
+
+ assert.equal(mounts.length, 2);
+ assert.ok(mounts[0].endsWith(':/workspace/.npm'));
+ // Traversal is stripped, not honoured.
+ assert.ok(mounts[1].endsWith(':/workspace/escape/attempt'));
+ for (const mount of mounts) {
+ const host = mount.split(':')[0];
+ assert.ok((await fs.stat(host)).isDirectory());
+ assert.ok(host.startsWith(dir));
+ }
+ await fs.rm(dir, { recursive: true, force: true });
+});
+
+test('artifact collection globs, and refuses to leave the workspace', async () => {
+ const dir = await tempDir();
+ const outside = await tempDir('conductor-outside-');
+ await fs.writeFile(path.join(outside, 'secret.txt'), 'do not upload me');
+
+ await fs.mkdir(path.join(dir, 'dist/nested'), { recursive: true });
+ await fs.mkdir(path.join(dir, '.conductor'), { recursive: true });
+ await fs.writeFile(path.join(dir, 'dist/app.bin'), 'binary');
+ await fs.writeFile(path.join(dir, 'dist/nested/deep.bin'), 'deep');
+ await fs.writeFile(path.join(dir, '.conductor/script.sh'), 'set -e');
+ await fs.symlink(path.join(outside, 'secret.txt'), path.join(dir, 'dist/escape.txt'));
+
+ const found = await collectArtifacts(dir, ['dist/**'], { logger: { warn: () => {} } });
+ const names = found.map((f) => f.path).sort();
+
+ assert.ok(names.includes('dist/app.bin'));
+ assert.ok(names.includes('dist/nested/deep.bin'));
+ assert.ok(!names.includes('dist/escape.txt'), 'symlink out of the workspace must be refused');
+ assert.ok(!names.some((n) => n.startsWith('.conductor')));
+ assert.equal(found.find((f) => f.path === 'dist/app.bin').size, 6);
+
+ await fs.rm(dir, { recursive: true, force: true });
+ await fs.rm(outside, { recursive: true, force: true });
+});
+
+test('artifact when controls collection', () => {
+ assert.equal(shouldCollect('on_success', true), true);
+ assert.equal(shouldCollect('on_success', false), false);
+ assert.equal(shouldCollect('on_failure', false), true);
+ assert.equal(shouldCollect('on_failure', true), false);
+ assert.equal(shouldCollect('always', false), true);
+});
+
+// A client stub that records what the log stream sends.
+function recordingClient() {
+ const calls = [];
+ let size = 0;
+ return {
+ calls,
+ get text() {
+ return calls.map((c) => c.chunk.toString('utf8')).join('');
+ },
+ async appendLog(url, chunk, offset) {
+ calls.push({ chunk, offset });
+ size = offset + chunk.length;
+ return { size };
+ },
+ };
+}
+
+test('log output is batched and offsets advance', async () => {
+ const client = recordingClient();
+ const log = createLogStream(client, 'http://x/log', { flushMs: 5 });
+
+ log.write('hello ');
+ log.write('world\n');
+ await log.close();
+
+ assert.equal(client.text, 'hello world\n');
+ assert.equal(client.calls[0].offset, 0);
+ assert.equal(log.offset, 12);
+});
+
+test('a conflicting offset resynchronises instead of duplicating', async () => {
+ const calls = [];
+ let first = true;
+ const client = {
+ async appendLog(url, chunk, offset) {
+ calls.push({ text: chunk.toString('utf8'), offset });
+ if (first) {
+ first = false;
+ // The conductor already has the first four bytes.
+ return { conflict: true, expected: 4 };
+ }
+ return { size: offset + chunk.length };
+ },
+ };
+
+ const log = createLogStream(client, 'http://x/log', { flushMs: 5 });
+ log.write('abcdefgh');
+ await log.close();
+
+ assert.equal(calls.length, 2);
+ assert.equal(calls[1].offset, 4);
+ assert.equal(calls[1].text, 'efgh', 'only the bytes the conductor lacks should be resent');
+});
+
+test('secrets are masked before anything leaves the host', async () => {
+ const client = recordingClient();
+ const log = createLogStream(client, 'http://x/log', { masked: ['hunter2'], flushMs: 5 });
+
+ log.write('password is hunter2 ok\n');
+ await log.close();
+
+ assert.ok(!client.text.includes('hunter2'));
+ assert.ok(client.text.includes('[masked]'));
+});
+
+test('a secret split across two writes is still masked', async () => {
+ const client = recordingClient();
+ const log = createLogStream(client, 'http://x/log', { masked: ['supersecret'], flushMs: 5 });
+
+ log.write('token=super');
+ log.write('secret done\n');
+ await log.close();
+
+ assert.ok(!client.text.includes('supersecret'), `leaked: ${client.text}`);
+ assert.ok(client.text.includes('[masked]'));
+});
+
+test('a failing conductor does not break the job', async () => {
+ const client = {
+ async appendLog() {
+ throw new Error('conductor unreachable');
+ },
+ };
+ const log = createLogStream(client, 'http://x/log', { flushMs: 5, logger: { warn: () => {} } });
+ log.write('some output\n');
+ await log.close();
+ assert.match(log.error.message, /unreachable/);
+});
