commit bd658275e2b0dd996745b9b63483127d5cbbac12
parent 5d11ffb12e6835d1516157fbe2e0c39446c4717d
Author: finwo <finwo@pm.me>
Date: Sat, 19 Sep 2026 00:59:25 +0200
Conductor service: triggers, scheduling and worker API
Diffstat:
25 files changed, 3551 insertions(+), 9 deletions(-)
diff --git a/README.md b/README.md
@@ -0,0 +1,180 @@
+conductor
+=========
+
+A stand-alone CI system. Pipelines live in the repository, jobs run in
+containers, and workers pull work rather than being pushed to.
+
+Summary
+-------
+
+The conductor accepts a push notification, reads `.conductor.yml` at the
+pushed commit, and turns it into a graph of jobs. Workers poll for work,
+advertising the architectures and features they can offer. Any job whose
+dependencies have succeeded is eligible, so a wide graph runs wide.
+
+Workers need no repository credentials: the conductor serves the source tree
+for the exact commit a job was handed. A worker on someone else's hardware
+can therefore build for you without being able to read anything else.
+
+Nothing has to be set up before it will run. With no configuration it uses
+sqlite on disk, stores artifacts on the local filesystem, and manages its own
+user accounts. Point it at a database url, an S3 bucket or an OIDC issuer and
+it uses those instead.
+
+Status
+------
+
+Under construction. Working today:
+
+ - configuration, with sqlite, mysql/tidb and postgres backends
+ - local and S3 compatible object storage
+ - pipeline parsing, matrix and architecture expansion, dependency graphs
+ - triggers, run scheduling, the worker API, live logs, artifacts, reaping
+
+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`.
+
+Installation
+------------
+
+Requires node 24 or newer. No native modules.
+
+```sh
+npm install
+cp conductor.example.yaml conductor.yaml
+npm start
+```
+
+`mysql2` and `pg` are optional and only needed when `database.url` points at
+one of those servers.
+
+Configuration
+-------------
+
+Every setting has a default, so an empty config file is valid. Values are
+read from defaults, then `conductor.yaml`, then the environment. See
+`conductor.example.yaml` for the annotated list.
+
+Three capabilities switch on when configured and fall back when not:
+
+| Setting | Configured | Not configured |
+| --------------------- | -------------------- | --------------------- |
+| `database.url` | mysql/tidb, postgres | sqlite at a file path |
+| `storage.s3.bucket` | S3 compatible store | local filesystem |
+| `auth.oidc.issuer` | OIDC | built-in accounts |
+
+Usage
+-----
+
+Register a project and mint a worker token:
+
+```sh
+node src/admin-cli.js project:add demo https://git.example.com/demo.git
+node src/admin-cli.js token:add builder-1
+```
+
+Both print a secret once. Put the trigger secret in the git hook and the
+worker token in the worker configuration.
+
+Install `hooks/post-receive` into the bare repository:
+
+```sh
+cp hooks/post-receive /srv/git/demo.git/hooks/
+chmod +x /srv/git/demo.git/hooks/post-receive
+```
+
+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.
+
+Pipelines
+---------
+
+```yaml
+version: 1
+
+defaults:
+ image: debian:bookworm-slim
+ timeout: 30m
+
+jobs:
+ test:
+ script:
+ - make test
+
+ build:
+ arch: [x86_64, aarch64]
+ script:
+ - ./build.sh $ARCH
+ artifacts:
+ paths: [dist/**]
+
+ package:
+ needs: [build]
+ arch: [x86_64, aarch64]
+ matrix:
+ pkg: [musl, busybox]
+ requires: [sign-key]
+ script:
+ - ./package.sh $MATRIX_PKG $ARCH
+
+ publish:
+ needs: [package, test]
+ script:
+ - ./publish.sh
+```
+
+`arch` and `matrix` fan a job out into one job per combination. A dependency
+that shares a dimension is matched on it, so the aarch64 package waits for
+the aarch64 build rather than for every build, while `publish`, which has no
+architecture of its own, waits for all of them.
+
+`requires` names features a worker must offer. A worker holding a signing key
+advertises `sign-key`, and only jobs asking for it are sent there. The key
+stays on the worker and is never known to the conductor.
+
+See `docs/pipeline.md` for the full reference.
+
+Architecture
+------------
+
+```
+git push -> post-receive -> conductor -> job graph in the database
+ |
+ worker polls
+ |
+ source tarball, container, log stream, artifacts
+```
+
+ - `src/conductor` the write side: triggers, scheduling, the worker API
+ - `src/read-api` the public read side, deployable separately
+ - `src/worker` the agent that runs jobs
+ - `src/lib` configuration, database, storage, pipelines, git
+
+Queries are written once against all three databases using named `{key}`
+markers, which each driver compiles to its own placeholder syntax.
+Timestamps are epoch milliseconds everywhere.
+
+Testing
+-------
+
+```sh
+npm test
+```
+
+S3 coverage is skipped unless a server is available:
+
+```sh
+docker run -d -p 9000:9000 -e MINIO_ROOT_USER=testkey \
+ -e MINIO_ROOT_PASSWORD=testsecret123 quay.io/minio/minio server /data
+
+CONDUCTOR_TEST_S3_ENDPOINT=http://127.0.0.1:9000 \
+CONDUCTOR_TEST_S3_KEY=testkey \
+CONDUCTOR_TEST_S3_SECRET=testsecret123 npm test
+```
+
+LICENSE
+-------
+
+See [LICENSE.md](LICENSE.md).
diff --git a/docs/pipeline.md b/docs/pipeline.md
@@ -0,0 +1,292 @@
+.conductor.yml reference
+========================
+
+The pipeline file is read from the repository at the commit being built, so
+it is versioned with the code it builds. The conductor never executes
+anything from the repository while parsing it; the file is data.
+
+Every unknown key is an error rather than a warning, and all problems in a
+file are reported together.
+
+Top level
+---------
+
+| Key | Required | Description |
+| ---------- | -------- | ------------------------------------ |
+| `version` | yes | Must be `1`. |
+| `defaults` | no | Values inherited by every job. |
+| `jobs` | yes | Mapping of job name to job. |
+
+Job names may contain letters, digits, underscore, dot and hyphen, and must
+start with a letter or digit.
+
+```yaml
+version: 1
+defaults:
+ image: debian:bookworm-slim
+jobs:
+ test:
+ script: [make test]
+```
+
+Jobs
+----
+
+| Key | Type | Default | Inheritable |
+| --------------- | ----------------- | -------------- | ----------- |
+| `image` | string | none, required | yes |
+| `script` | list of strings | none, required | no |
+| `needs` | list | `[]` | no |
+| `arch` | string or list | none | yes |
+| `matrix` | mapping | none | no |
+| `requires` | list of strings | `[]` | yes |
+| `services` | list | `[]` | yes |
+| `env` | mapping | `{}` | merged |
+| `artifacts` | list or mapping | none | no |
+| `cache` | list or mapping | none | yes |
+| `allow_failure` | boolean | `false` | yes |
+| `timeout` | duration | server default | yes |
+| `max_attempts` | integer, 1 to 10 | `1` | yes |
+
+`image` and `script` are the only required fields, and `image` may come from
+`defaults`.
+
+### script
+
+A list of shell commands, run in order in the job's container. The job fails
+on the first command that exits non-zero.
+
+Take care with YAML scalars. Unquoted `true`, `false`, `yes`, `no` and `on`
+parse as booleans, not strings, and are rejected:
+
+```yaml
+script: ['true'] # correct
+script: [true] # rejected, this is a boolean
+```
+
+### timeout
+
+Accepts whole seconds, or a duration with units `s`, `m`, `h`, `d` and `w`,
+optionally combined: `90`, `90s`, `30m`, `1h`, `1h30m`. A job timeout may not
+exceed seven days; `artifacts.expire` may not exceed a year.
+
+Fanning out
+-----------
+
+### arch
+
+Expands the job into one job per architecture, and restricts each to a worker
+that offers it. The value is available to the script as `$ARCH`.
+
+```yaml
+build:
+ arch: [x86_64, aarch64]
+ script: ['./build.sh $ARCH']
+```
+
+produces `build:arch=x86_64` and `build:arch=aarch64`.
+
+### matrix
+
+Expands over the cartesian product of its dimensions. Each value is available
+as `$MATRIX_<NAME>` in upper case. Dimension order follows the file, so
+generated names are predictable.
+
+```yaml
+package:
+ matrix:
+ pkg: [musl, busybox]
+ mode: [debug, release]
+```
+
+produces four jobs, named `package:pkg=musl,mode=debug` and so on.
+
+Matrix values become part of the job name, so they are limited to letters,
+digits, underscore, dot and hyphen. Use `arch` rather than a matrix dimension
+called `arch`.
+
+Dependencies
+------------
+
+`needs` lists jobs that must succeed first. A cycle is rejected, naming the
+cycle.
+
+When a dependency shares a dimension with the dependent, it is matched on
+that dimension rather than fanned out across it:
+
+```yaml
+build:
+ arch: [x86_64, aarch64]
+ script: ['./build.sh $ARCH']
+
+package:
+ needs: [build]
+ arch: [x86_64, aarch64]
+ script: ['./package.sh $ARCH']
+
+publish:
+ needs: [package]
+ script: ['./publish.sh']
+```
+
+`package:arch=x86_64` waits only for `build:arch=x86_64`. `publish` declares
+no architecture, so it waits for every `package`.
+
+Three forms are accepted:
+
+```yaml
+needs: [build] # match shared dimensions
+needs: ['build:arch=x86_64'] # one specific job
+needs:
+ - job: build
+ match: all # every instance, ignore dimensions
+```
+
+If a shared dimension has no counterpart, that is an error rather than a
+silently dropped dependency.
+
+A dependency that fails normally causes its dependents, and everything
+reachable from them, to be skipped. A dependency with `allow_failure: true`
+satisfies its dependents whether it passes or not.
+
+Worker features
+---------------
+
+`requires` names capabilities the worker must advertise. The conductor only
+offers a job to a worker that advertises all of them.
+
+```yaml
+package:
+ requires: [sign-key]
+ script: ['./sign.sh']
+```
+
+What a feature provides is defined on the worker, not here: a mount, some
+environment, a privileged container. This is how a signing key reaches a
+build without the conductor ever holding it, and how a job asks for docker in
+docker.
+
+Services
+--------
+
+Sidecar containers started alongside the job on the same network, reachable
+by their alias.
+
+```yaml
+test:
+ image: node:22
+ services:
+ - postgres:16
+ - image: docker:dind
+ alias: docker
+ env:
+ DOCKER_TLS_CERTDIR: ''
+ script: [npm test]
+```
+
+The alias defaults to the image name without registry, path or tag, so
+`postgres:16` is reachable as `postgres`. Aliases must be unique within a job.
+
+Artifacts
+---------
+
+```yaml
+artifacts:
+ paths:
+ - dist/**
+ - build/*.tar.gz
+ when: on_success
+ expire: 30d
+```
+
+A bare list is shorthand for `paths`. Paths are relative to the workspace;
+absolute paths are rejected. `when` is `on_success`, `on_failure` or
+`always`.
+
+Cache
+-----
+
+```yaml
+cache:
+ key: deps-$MATRIX_PKG
+ paths: [.npm, vendor]
+```
+
+A directory preserved between jobs on the same worker, keyed by `key`. A
+cache is an optimisation, never a correctness guarantee: a job must work with
+an empty one.
+
+Environment
+-----------
+
+`env` values must be scalars, and names must be valid environment variable
+names. Job `env` is merged over `defaults.env`.
+
+Injected automatically:
+
+| Variable | Present when |
+| ----------------- | -------------------------------- |
+| `ARCH` | the job declares `arch` |
+| `MATRIX_<NAME>` | for each matrix dimension |
+
+Interpolation
+-------------
+
+`${{ ... }}` is substituted before the job is stored, in `image`, `script`,
+`env` values, `requires`, `services` and `cache`. Two expressions are
+available: `arch` and `matrix.<name>`.
+
+```yaml
+build:
+ arch: [x86_64]
+ matrix:
+ pkg: [musl]
+ image: 'builder:${{ arch }}'
+ script: ['./build.sh ${{ matrix.pkg }}']
+```
+
+Referring to a dimension the job does not declare is an error. Inside
+`script` the `$ARCH` and `$MATRIX_*` variables usually read better, since the
+shell expands them; interpolation exists for the fields the shell never sees,
+such as `image`.
+
+Full example
+------------
+
+```yaml
+version: 1
+
+defaults:
+ image: debian:bookworm-slim
+ timeout: 30m
+
+jobs:
+ lint:
+ script: [make lint]
+
+ build:
+ arch: [x86_64, aarch64]
+ script: ['./mk/build.sh $ARCH']
+ artifacts:
+ paths: [build/out/**]
+ expire: 7d
+
+ package:
+ needs: [build]
+ arch: [x86_64, aarch64]
+ matrix:
+ pkg: [musl, busybox]
+ requires: [sign-key]
+ script: ['./mk/package.sh $MATRIX_PKG $ARCH']
+ artifacts:
+ paths: ['build/repo/$ARCH/*.apk']
+
+ publish:
+ needs: [package, lint]
+ requires: [publish-key]
+ script: ['./mk/publish.sh']
+```
+
+This yields nine jobs: one `lint`, two `build`, four `package` and one
+`publish`. The two `build` jobs run at the same time as `lint`; each
+`package` starts as soon as its own architecture's `build` finishes.
diff --git a/examples/distro.conductor.yml b/examples/distro.conductor.yml
@@ -0,0 +1,64 @@
+# A source distribution building signed packages for two architectures.
+#
+# This is the shape the conductor was extracted from. Two details matter:
+#
+# Each package job waits only for the build of its own architecture,
+# because the dependency shares the arch dimension. The aarch64 packages
+# do not sit behind the x86_64 build.
+#
+# Signing happens on the worker. The worker holds the key and advertises
+# the sign-key feature; the job asks for it by name. The conductor never
+# sees the key, which is what makes it safe to accept build capacity from
+# someone else's hardware.
+#
+# The package list is written out rather than discovered, since the pipeline
+# is data and the conductor does not run repository code to build the graph.
+
+version: 1
+
+defaults:
+ image: debian:bookworm-slim
+ timeout: 2h
+
+jobs:
+ check:
+ timeout: 10m
+ script:
+ - ./mk/deps.sh check
+
+ toolchain:
+ arch: [x86_64, aarch64]
+ script:
+ - ./mk/bootstrap-host.sh --check
+ - ./mk/sysroot.sh --arch $ARCH
+ artifacts:
+ paths: ['build/sysroot-$ARCH.tar.zst']
+ expire: 7d
+
+ package:
+ needs: [toolchain]
+ arch: [x86_64, aarch64]
+ matrix:
+ pkg: [musl, busybox, glibc, grub, tinyssh, base-files]
+ requires: [sign-key]
+ script:
+ - ./mk/build.sh $MATRIX_PKG $ARCH
+ artifacts:
+ paths: ['build/repo/$ARCH/$MATRIX_PKG-*.apk']
+
+ image:
+ needs: [package]
+ arch: [x86_64, aarch64]
+ script:
+ - ./mk/rootfs.sh $ARCH
+ - ./mk/make-img.sh $ARCH
+ artifacts:
+ paths: ['build/images/*-$ARCH.img']
+
+ publish:
+ # No arch of its own, so this waits for every image and every package.
+ needs: [image, check]
+ requires: [publish-key]
+ script:
+ - ./mk/repo-index.sh
+ - ./mk/publish.sh
diff --git a/examples/node-app.conductor.yml b/examples/node-app.conductor.yml
@@ -0,0 +1,55 @@
+# A typical node service: install, check, build, publish an image.
+#
+# The publish job asks for the docker feature, so it only runs on a worker
+# configured to provide a docker socket or a docker in docker service.
+
+version: 1
+
+defaults:
+ image: node:22-bookworm-slim
+ timeout: 20m
+
+jobs:
+ install:
+ script:
+ - npm ci
+ artifacts:
+ paths: [node_modules/**]
+ expire: 1h
+ cache:
+ key: npm
+ paths: [.npm]
+
+ lint:
+ needs: [install]
+ script:
+ - npm run lint
+
+ test:
+ needs: [install]
+ matrix:
+ suite: [unit, integration]
+ services:
+ - image: postgres:16
+ env:
+ POSTGRES_PASSWORD: test
+ env:
+ DATABASE_URL: postgres://postgres:test@postgres:5432/postgres
+ script:
+ - npm run test:$MATRIX_SUITE
+ artifacts:
+ paths: [coverage/**]
+ when: always
+
+ image:
+ needs: [lint, test]
+ image: docker:27-cli
+ requires: [docker]
+ services:
+ - image: docker:27-dind
+ alias: docker
+ env:
+ DOCKER_HOST: tcp://docker:2375
+ script:
+ - docker build -t example/app:$CONDUCTOR_SHA .
+ - docker push example/app:$CONDUCTOR_SHA
diff --git a/hooks/post-receive b/hooks/post-receive
@@ -0,0 +1,118 @@
+#!/bin/sh
+# hooks/post-receive - notify a conductor of pushed commits
+#
+# Install into a bare repository as hooks/post-receive, chmod +x.
+# Reads "<old> <new> <ref>" per pushed ref on stdin.
+#
+# Required:
+# CONDUCTOR_URL base url, for example https://ci.example.com
+# CONDUCTOR_PROJECT project id registered with the conductor
+#
+# Optional:
+# CONDUCTOR_SECRET trigger secret, or a path to a file holding it.
+# Required whenever the project has one set.
+# CONDUCTOR_REFS space separated refs to build. Default: every
+# branch. Example: "refs/heads/main refs/heads/dev"
+# CONDUCTOR_TIMEOUT seconds to wait per request, default 10
+#
+# The hook decides nothing about the build. It reports which commit was
+# pushed and the conductor reads the pipeline from the repository itself.
+# A failure here never blocks the push.
+
+set -eu
+
+CONDUCTOR_URL="${CONDUCTOR_URL:-}"
+CONDUCTOR_PROJECT="${CONDUCTOR_PROJECT:-}"
+CONDUCTOR_SECRET="${CONDUCTOR_SECRET:-}"
+CONDUCTOR_REFS="${CONDUCTOR_REFS:-}"
+CONDUCTOR_TIMEOUT="${CONDUCTOR_TIMEOUT:-10}"
+
+ZERO="0000000000000000000000000000000000000000"
+
+log() { printf '[conductor] %s\n' "$*" >&2; }
+
+if [ -z "${CONDUCTOR_URL}" ] || [ -z "${CONDUCTOR_PROJECT}" ]; then
+ log "CONDUCTOR_URL and CONDUCTOR_PROJECT must be set; not triggering"
+ exit 0
+fi
+
+# The secret may be given inline or as a file, so it can stay out of the
+# process environment on shared hosts.
+if [ -n "${CONDUCTOR_SECRET}" ] && [ -f "${CONDUCTOR_SECRET}" ]; then
+ CONDUCTOR_SECRET=$(cat "${CONDUCTOR_SECRET}")
+fi
+
+if command -v curl >/dev/null 2>&1; then
+ http_client=curl
+elif command -v wget >/dev/null 2>&1; then
+ http_client=wget
+else
+ log "neither curl nor wget is available; not triggering"
+ exit 0
+fi
+
+want_ref() {
+ [ -z "${CONDUCTOR_REFS}" ] && case "$1" in
+ refs/heads/*) return 0 ;;
+ *) return 1 ;;
+ esac
+ for candidate in ${CONDUCTOR_REFS}; do
+ [ "${candidate}" = "$1" ] && return 0
+ done
+ return 1
+}
+
+while read -r old new ref; do
+ if [ "${new}" = "${ZERO}" ]; then
+ log "skipping ${ref}, branch deleted"
+ continue
+ fi
+
+ if ! want_ref "${ref}"; then
+ log "skipping ${ref}"
+ continue
+ fi
+
+ # A new branch reports an all zero parent, which is not a commit.
+ if [ "${old}" = "${ZERO}" ]; then
+ base=""
+ else
+ base="${old}"
+ fi
+
+ payload=$(printf '{"sha":"%s","base":"%s","ref":"%s","actor":"%s"}' \
+ "${new}" "${base}" "${ref}" "${USER:-git}")
+
+ if [ -n "${CONDUCTOR_SECRET}" ]; then
+ digest=$(printf '%s' "${payload}" | openssl dgst -sha256 -hmac "${CONDUCTOR_SECRET}" | sed 's/^.*[= ]//')
+ signature="sha256=${digest}"
+ else
+ signature=""
+ log "warning: no CONDUCTOR_SECRET set, sending unsigned"
+ fi
+
+ url="${CONDUCTOR_URL%/}/api/trigger/${CONDUCTOR_PROJECT}"
+ log "triggering ${ref} at $(echo "${new}" | cut -c1-12)"
+
+ if [ "${http_client}" = curl ]; then
+ response=$(curl -fsS -m "${CONDUCTOR_TIMEOUT}" -X POST "${url}" \
+ -H 'Content-Type: application/json' \
+ -H "X-Hub-Signature-256: ${signature}" \
+ -d "${payload}" 2>&1) || {
+ log "trigger failed for ${ref}: ${response}"
+ continue
+ }
+ else
+ response=$(wget -qO- --timeout="${CONDUCTOR_TIMEOUT}" \
+ --header='Content-Type: application/json' \
+ --header="X-Hub-Signature-256: ${signature}" \
+ --post-data="${payload}" "${url}" 2>&1) || {
+ log "trigger failed for ${ref}: ${response}"
+ continue
+ }
+ fi
+
+ log "${response}"
+done
+
+exit 0
diff --git a/src/admin-cli.js b/src/admin-cli.js
@@ -0,0 +1,213 @@
+// src/admin-cli.js - local administration
+//
+// Usage:
+// node src/admin-cli.js project:add <id> <repo-url> [--name n] [--branch b]
+// [--config p] [--source archive|clone]
+// [--secret s]
+// node src/admin-cli.js project:list
+// node src/admin-cli.js project:secret <id> [secret]
+// node src/admin-cli.js project:remove <id>
+// node src/admin-cli.js token:add <name>
+// node src/admin-cli.js token:list
+// node src/admin-cli.js token:remove <id>
+// node src/admin-cli.js run:trigger <project> <sha> [--ref r] [--base b]
+// node src/admin-cli.js run:cancel <run-id>
+//
+// The HTTP admin API arrives with the auth work. Until then this is the
+// supported way to register projects and mint worker tokens.
+
+import crypto from 'node:crypto';
+import { loadConfig } from './lib/config.js';
+import { createServices } from './conductor/app.js';
+
+const [command, ...rest] = process.argv.slice(2);
+
+// Splits --flag value pairs out of the positional arguments.
+function parseArgs(argv) {
+ const flags = {};
+ const positional = [];
+ for (let i = 0; i < argv.length; i += 1) {
+ if (argv[i].startsWith('--')) {
+ const name = argv[i].slice(2);
+ const next = argv[i + 1];
+ if (next === undefined || next.startsWith('--')) {
+ flags[name] = true;
+ } else {
+ flags[name] = next;
+ i += 1;
+ }
+ } else {
+ positional.push(argv[i]);
+ }
+ }
+ return { flags, positional };
+}
+
+const { flags, positional } = parseArgs(rest);
+
+function usage(message) {
+ if (message) console.error(`error: ${message}\n`);
+ console.error(
+ [
+ 'usage:',
+ ' project:add <id> <repo-url> [--name n] [--branch b] [--config p] [--source archive|clone] [--secret s]',
+ ' project:list',
+ ' project:secret <id> [secret]',
+ ' project:remove <id>',
+ ' token:add <name>',
+ ' token:list',
+ ' token:remove <id>',
+ ' run:trigger <project> <sha> [--ref r] [--base b]',
+ ' run:cancel <run-id>',
+ ].join('\n')
+ );
+ process.exit(message ? 1 : 0);
+}
+
+if (!command || command === 'help' || command === '--help') usage();
+
+const cfg = loadConfig();
+const services = await createServices(cfg, { migrationLogger: () => {} });
+const { projects, workerTokens, scheduler, db } = services;
+
+function table(rows, columns) {
+ if (rows.length === 0) {
+ console.log('(none)');
+ return;
+ }
+ const widths = columns.map((c) => Math.max(c.length, ...rows.map((r) => String(r[c] ?? '').length)));
+ console.log(columns.map((c, i) => c.padEnd(widths[i])).join(' '));
+ for (const row of rows) {
+ console.log(columns.map((c, i) => String(row[c] ?? '').padEnd(widths[i])).join(' '));
+ }
+}
+
+const when = (ms) => (ms ? new Date(ms).toISOString().replace('T', ' ').slice(0, 19) : '');
+
+try {
+ switch (command) {
+ case 'project:add': {
+ const [id, repoUrl] = positional;
+ if (!id || !repoUrl) usage('project:add needs an id and a repository url');
+
+ // Generated when not supplied, since an unsigned trigger endpoint is
+ // rarely what anyone actually wants.
+ const secret = flags.secret === true || flags.secret === undefined
+ ? crypto.randomBytes(24).toString('hex')
+ : String(flags.secret);
+
+ const project = await projects.create({
+ id,
+ name: flags.name ? String(flags.name) : id,
+ repo_url: repoUrl,
+ default_branch: flags.branch ? String(flags.branch) : 'main',
+ config_path: flags.config ? String(flags.config) : '.conductor.yml',
+ source_mode: flags.source ? String(flags.source) : 'archive',
+ trigger_secret: secret,
+ });
+
+ console.log(`created project ${project.id}`);
+ console.log(` repository: ${project.repo_url}`);
+ console.log(` pipeline: ${project.config_path}`);
+ console.log(` source mode: ${project.source_mode}`);
+ console.log(` trigger url: ${cfg.server.public_url.replace(/\/+$/, '')}/api/trigger/${project.id}`);
+ console.log(` secret: ${secret}`);
+ break;
+ }
+
+ case 'project:list': {
+ const rows = (await projects.list()).map((p) => ({
+ id: p.id,
+ repo_url: p.repo_url,
+ branch: p.default_branch,
+ config: p.config_path,
+ source: p.source_mode,
+ enabled: p.enabled ? 'yes' : 'no',
+ runs: p.run_counter,
+ }));
+ table(rows, ['id', 'repo_url', 'branch', 'config', 'source', 'enabled', 'runs']);
+ break;
+ }
+
+ case 'project:secret': {
+ const [id, value] = positional;
+ if (!id) usage('project:secret needs a project id');
+ if (!(await projects.get(id))) usage(`unknown project ${id}`);
+ const secret = value ?? crypto.randomBytes(24).toString('hex');
+ await projects.setTriggerSecret(id, secret);
+ console.log(`trigger secret for ${id}: ${secret}`);
+ break;
+ }
+
+ case 'project:remove': {
+ const [id] = positional;
+ if (!id) usage('project:remove needs a project id');
+ await projects.remove(id);
+ console.log(`removed project ${id}`);
+ break;
+ }
+
+ case 'token:add': {
+ const [name] = positional;
+ if (!name) usage('token:add needs a name');
+ const created = await workerTokens.create(name);
+ console.log(`created worker token ${created.id} for ${created.name}`);
+ console.log(` token: ${created.token}`);
+ console.log(' This is shown once. Store it in the worker configuration now.');
+ break;
+ }
+
+ case 'token:list': {
+ const rows = (await workerTokens.list()).map((t) => ({
+ id: t.id,
+ name: t.name,
+ enabled: t.enabled ? 'yes' : 'no',
+ created: when(t.created_at),
+ last_seen: when(t.last_seen_at),
+ last_ip: t.last_ip ?? '',
+ }));
+ table(rows, ['id', 'name', 'enabled', 'created', 'last_seen', 'last_ip']);
+ break;
+ }
+
+ case 'token:remove': {
+ const [id] = positional;
+ if (!id) usage('token:remove needs a token id');
+ console.log(await workerTokens.remove(id) ? `removed token ${id}` : `no such token ${id}`);
+ break;
+ }
+
+ case 'run:trigger': {
+ const [projectId, sha] = positional;
+ if (!projectId || !sha) usage('run:trigger needs a project and a commit');
+ const project = await projects.get(projectId);
+ if (!project) usage(`unknown project ${projectId}`);
+
+ const result = await scheduler.createRun(project, {
+ headSha: sha,
+ ref: flags.ref ? String(flags.ref) : null,
+ baseSha: flags.base ? String(flags.base) : null,
+ trigger: 'manual',
+ actor: 'admin-cli',
+ });
+ console.log(`created run ${result.runId} with ${result.jobCount} job(s)`);
+ break;
+ }
+
+ case 'run:cancel': {
+ const [runId] = positional;
+ if (!runId) usage('run:cancel needs a run id');
+ const result = await scheduler.cancelRun(runId);
+ console.log(result.ok ? `cancelled ${runId}` : `could not cancel: ${result.reason}`);
+ break;
+ }
+
+ default:
+ usage(`unknown command ${command}`);
+ }
+} catch (e) {
+ console.error(`error: ${e.message}`);
+ process.exitCode = 1;
+} finally {
+ await db.close();
+}
diff --git a/src/conductor/app.js b/src/conductor/app.js
@@ -0,0 +1,91 @@
+// src/conductor/app.js - assembles the conductor
+//
+// Separated from index.js so tests can build a server on a temporary
+// database without starting a listener or installing signal handlers.
+
+import Fastify from 'fastify';
+import cors from '@fastify/cors';
+
+import { ensureStateDirs, parseKey } from '../lib/config.js';
+import { openDatabase, runMigrations } from '../lib/db/index.js';
+import { createStorage } from '../lib/storage/index.js';
+import { createSecretBox } from '../lib/secretbox.js';
+import { createLogStore } from '../lib/log.js';
+import { createGit } from '../lib/git.js';
+import { createProjects } from '../lib/projects.js';
+import { createWorkerTokens } from '../lib/workers.js';
+import { createScheduler } from './scheduler.js';
+
+import workerRoutes from './routes/workers.js';
+import triggerRoutes from './routes/trigger.js';
+import runRoutes from './routes/runs.js';
+
+export async function createServices(cfg, options = {}) {
+ ensureStateDirs(cfg);
+
+ const db = await openDatabase(cfg);
+ if (options.migrate !== false) {
+ await runMigrations(db, { logger: options.migrationLogger });
+ }
+
+ const secrets = createSecretBox(parseKey(cfg.secrets.encryption_key, 'secrets.encryption_key'));
+ const storage = createStorage(cfg);
+ const logs = createLogStore(cfg);
+ const git = createGit(cfg);
+ const projects = createProjects({ db, secrets });
+ const workerTokens = createWorkerTokens({ db });
+
+ const logger = options.logger ?? console;
+ const scheduler = createScheduler({ cfg, db, git, logs, storage, projects, logger });
+
+ return { cfg, db, secrets, storage, logs, git, projects, workerTokens, scheduler, logger };
+}
+
+export async function buildServer(services, options = {}) {
+ const { cfg } = services;
+ const fastify = Fastify({
+ logger: options.logger ?? { level: process.env.LOG_LEVEL || 'info' },
+ // Workers upload artifacts as raw streams; this only bounds parsed
+ // bodies such as the trigger payload.
+ bodyLimit: 1024 * 1024,
+ trustProxy: options.trustProxy ?? true,
+ });
+
+ await fastify.register(cors, { origin: true });
+
+ fastify.get('/health', async () => ({
+ ok: true,
+ service: 'conductor',
+ database: services.db.dialect,
+ storage: services.storage.driver,
+ auth: cfg.auth.mode,
+ }));
+
+ await fastify.register(triggerRoutes, { ...services, prefix: '/api/trigger' });
+ await fastify.register(workerRoutes, { ...services, prefix: '/api/workers' });
+ await fastify.register(runRoutes, { ...services, prefix: '/api' });
+
+ return fastify;
+}
+
+// Requeues or fails jobs whose worker went away. Returns a stop function.
+export function startReaper(services) {
+ const { cfg, scheduler, logger } = services;
+ let running = false;
+
+ const timer = setInterval(async () => {
+ // Skip a tick rather than overlapping if a sweep runs long.
+ if (running) return;
+ running = true;
+ try {
+ await scheduler.reap();
+ } catch (e) {
+ logger.error?.(`reaper failed: ${e.message}`);
+ } finally {
+ running = false;
+ }
+ }, cfg.scheduler.reap_interval * 1000);
+
+ timer.unref();
+ return () => clearInterval(timer);
+}
diff --git a/src/conductor/index.js b/src/conductor/index.js
@@ -0,0 +1,39 @@
+// src/conductor/index.js - conductor entry point
+//
+// The write side: accepts triggers, schedules runs, and serves the worker
+// API. Run the read-api separately when the public read surface should not
+// sit on the same port as the write surface.
+
+import { loadConfig } from '../lib/config.js';
+import { createServices, buildServer, startReaper } from './app.js';
+
+const cfg = loadConfig();
+const services = await createServices(cfg);
+const fastify = await buildServer(services);
+
+if (cfg.auth.session_secret_ephemeral) {
+ fastify.log.warn('auth.session_secret is unset; a random one was generated and sessions will not survive a restart');
+}
+if (!cfg.secrets.encryption_key) {
+ fastify.log.warn('secrets.encryption_key is unset; stored secrets are kept in the clear');
+}
+
+const stopReaper = startReaper({ ...services, logger: fastify.log });
+
+async function shutdown(signal) {
+ fastify.log.info(`${signal} received, shutting down`);
+ stopReaper();
+ try {
+ await fastify.close();
+ await services.db.close();
+ } catch (e) {
+ fastify.log.error({ err: e }, 'shutdown failed');
+ process.exitCode = 1;
+ }
+}
+
+for (const signal of ['SIGINT', 'SIGTERM']) {
+ process.once(signal, () => { shutdown(signal); });
+}
+
+await fastify.listen({ port: cfg.server.port, host: cfg.server.host });
diff --git a/src/conductor/routes/runs.js b/src/conductor/routes/runs.js
@@ -0,0 +1,164 @@
+// src/conductor/routes/runs.js - read access to runs, jobs and logs
+//
+// These routes are read only and carry no secrets, so the read-api service
+// will mount the same handlers in phase 6. Anything that mutates state lives
+// under the admin routes instead.
+
+import { StorageNotFound } from '../../lib/storage/index.js';
+
+const MAX_TAIL = 1024 * 1024;
+
+export default async function runRoutes(fastify, { db, logs, storage }) {
+ fastify.get('/runs', async (req, reply) => {
+ const limit = clamp(req.query.limit, 1, 200, 50);
+ const project = typeof req.query.project === 'string' ? req.query.project : null;
+
+ const rows = project
+ ? await db.all(
+ `SELECT id, project_id, number, ref, base_sha, head_sha, trigger_type, actor,
+ title, state, created_at, started_at, finished_at
+ FROM runs WHERE project_id = {project}
+ ORDER BY created_at DESC LIMIT {limit}`,
+ { project, limit }
+ )
+ : await db.all(
+ `SELECT id, project_id, number, ref, base_sha, head_sha, trigger_type, actor,
+ title, state, created_at, started_at, finished_at
+ FROM runs ORDER BY created_at DESC LIMIT {limit}`,
+ { limit }
+ );
+
+ return reply.send({ runs: rows });
+ });
+
+ fastify.get('/runs/:id', async (req, reply) => {
+ const run = await db.get(
+ `SELECT id, project_id, number, ref, base_sha, head_sha, trigger_type, actor,
+ title, state, error, created_at, started_at, finished_at
+ FROM runs WHERE id = {id}`,
+ { id: req.params.id }
+ );
+ if (!run) return reply.code(404).send({ error: 'unknown run' });
+
+ const jobs = await db.all(
+ `SELECT id, name, base_name, arch, image, state, allow_failure, attempt, max_attempts,
+ exit_code, error, log_size, worker_name, timeout,
+ created_at, started_at, finished_at
+ FROM jobs WHERE run_id = {run} ORDER BY name`,
+ { run: run.id }
+ );
+
+ const deps = await db.all(
+ `SELECT d.job_id, d.depends_on_id
+ FROM job_deps d JOIN jobs j ON j.id = d.job_id
+ WHERE j.run_id = {run}`,
+ { run: run.id }
+ );
+ const needs = new Map(jobs.map((j) => [j.id, []]));
+ for (const d of deps) needs.get(d.job_id)?.push(d.depends_on_id);
+
+ return reply.send({
+ run,
+ jobs: jobs.map((j) => ({ ...j, needs: needs.get(j.id) ?? [] })),
+ });
+ });
+
+ fastify.get('/jobs/:id', async (req, reply) => {
+ const job = await db.get(
+ `SELECT id, run_id, name, base_name, arch, image, requires, spec, state, allow_failure,
+ attempt, max_attempts, exit_code, error, log_size, worker_name, timeout,
+ created_at, started_at, finished_at
+ FROM jobs WHERE id = {id}`,
+ { id: req.params.id }
+ );
+ if (!job) return reply.code(404).send({ error: 'unknown job' });
+
+ const artifacts = await db.all(
+ 'SELECT id, path, size, sha256, created_at FROM artifacts WHERE job_id = {job} ORDER BY path',
+ { job: job.id }
+ );
+
+ return reply.send({
+ job: { ...job, requires: safeJson(job.requires, []), spec: safeJson(job.spec, {}) },
+ artifacts,
+ });
+ });
+
+ // Incremental log tail. While a job runs the bytes come from the local
+ // spool; once it has finished they come from object storage.
+ fastify.get('/jobs/:id/log', async (req, reply) => {
+ const job = await db.get(
+ 'SELECT id, run_id, state, log_key, log_size FROM jobs WHERE id = {id}',
+ { id: req.params.id }
+ );
+ if (!job) return reply.code(404).send({ error: 'unknown job' });
+
+ const offset = clamp(req.query.offset, 0, Number.MAX_SAFE_INTEGER, 0);
+ const limit = clamp(req.query.limit, 1, MAX_TAIL, 64 * 1024);
+
+ if (!job.log_key) {
+ const chunk = await logs.read(job.run_id, job.id, { offset, limit });
+ reply.header('content-type', 'text/plain; charset=utf-8');
+ reply.header('x-log-offset', String(chunk.offset));
+ reply.header('x-log-size', String(chunk.size));
+ // Tells a follower whether to keep polling.
+ reply.header('x-log-complete', 'false');
+ return reply.send(chunk.data);
+ }
+
+ try {
+ const end = offset + limit - 1;
+ const object = await storage.get(job.log_key, { range: { start: offset, end } });
+ reply.header('content-type', 'text/plain; charset=utf-8');
+ reply.header('x-log-offset', String(offset));
+ reply.header('x-log-size', String(job.log_size));
+ reply.header('x-log-complete', 'true');
+ return reply.send(object.stream);
+ } catch (e) {
+ if (e instanceof StorageNotFound) return reply.code(404).send({ error: 'log is no longer stored' });
+ throw e;
+ }
+ });
+
+ fastify.get('/artifacts/:id', async (req, reply) => {
+ const artifact = await db.get(
+ 'SELECT id, job_id, run_id, path, storage_key, size, sha256 FROM artifacts WHERE id = {id}',
+ { id: req.params.id }
+ );
+ if (!artifact) return reply.code(404).send({ error: 'unknown artifact' });
+
+ // Hand the client straight to the object store when that is possible,
+ // rather than proxying the bytes through the conductor.
+ const url = await storage.presign(artifact.storage_key, { expires: 300 });
+ if (url) return reply.redirect(url, 302);
+
+ try {
+ const object = await storage.get(artifact.storage_key);
+ reply.header('content-type', 'application/octet-stream');
+ reply.header('content-length', String(artifact.size));
+ reply.header('content-disposition', `attachment; filename="${encodeURIComponent(basename(artifact.path))}"`);
+ return reply.send(object.stream);
+ } catch (e) {
+ if (e instanceof StorageNotFound) return reply.code(404).send({ error: 'artifact is no longer stored' });
+ throw e;
+ }
+ });
+}
+
+function clamp(raw, min, max, fallback) {
+ const n = Number(raw);
+ if (!Number.isFinite(n)) return fallback;
+ return Math.min(max, Math.max(min, Math.trunc(n)));
+}
+
+function basename(p) {
+ return p.split('/').pop() || 'artifact';
+}
+
+function safeJson(text, fallback) {
+ try {
+ return JSON.parse(text);
+ } catch {
+ return fallback;
+ }
+}
diff --git a/src/conductor/routes/trigger.js b/src/conductor/routes/trigger.js
@@ -0,0 +1,147 @@
+// src/conductor/routes/trigger.js - push notifications
+//
+// Accepts the generic payload produced by hooks/post-receive, and the push
+// payloads of GitHub, Gitea and GitLab, since they all carry the same three
+// facts. Nothing in the payload is trusted beyond which commit to look at:
+// the pipeline is always read from the repository at that commit.
+//
+// A project with a trigger secret requires a valid signature. A project
+// without one accepts unsigned triggers, which is convenient on a private
+// network and a bad idea anywhere else, so it is logged.
+
+import crypto from 'node:crypto';
+import { SHA_PATTERN } from '../../lib/git.js';
+import { PipelineError } from '../../lib/pipeline/index.js';
+
+const ZERO_SHA = '0'.repeat(40);
+
+export default async function triggerRoutes(fastify, { projects, scheduler, logger }) {
+ // HMAC is computed over the exact bytes received, so the raw body has to
+ // survive JSON parsing.
+ fastify.addContentTypeParser('application/json', { parseAs: 'buffer' }, (req, body, done) => {
+ req.rawBody = body;
+ if (body.length === 0) return done(null, {});
+ try {
+ done(null, JSON.parse(body.toString('utf8')));
+ } catch (e) {
+ e.statusCode = 400;
+ done(e);
+ }
+ });
+
+ fastify.post('/:project', async (req, reply) => {
+ const project = await projects.get(req.params.project);
+ if (!project) return reply.code(404).send({ error: 'unknown project' });
+ if (project.enabled !== 1) return reply.code(409).send({ error: 'project is disabled' });
+
+ const secret = projects.triggerSecret(project);
+ if (secret) {
+ if (!verifySignature(req, secret)) {
+ return reply.code(401).send({ error: 'invalid or missing signature' });
+ }
+ } else {
+ logger.warn?.(`project ${project.id} accepted an unsigned trigger; set a trigger secret`);
+ }
+
+ const push = normalizePush(req.body);
+ if (!push) {
+ return reply.code(400).send({
+ error: 'could not read a commit from the payload; expected sha, or a GitHub, Gitea or GitLab push event',
+ });
+ }
+ if (push.sha === ZERO_SHA) {
+ return reply.send({ status: 'ignored', reason: 'branch deletion' });
+ }
+ if (!SHA_PATTERN.test(push.sha)) {
+ return reply.code(400).send({ error: `not a full commit id: ${JSON.stringify(push.sha)}` });
+ }
+
+ try {
+ const { runId, jobCount } = await scheduler.createRun(project, {
+ ref: push.ref,
+ baseSha: push.base,
+ headSha: push.sha,
+ trigger: 'push',
+ actor: push.actor,
+ });
+ return reply.send({ status: 'created', run_id: runId, jobs: jobCount });
+ } catch (e) {
+ if (e instanceof PipelineError) {
+ // A broken pipeline is the pusher's problem, not a server fault.
+ return reply.code(422).send({ error: 'invalid pipeline', detail: e.message, problems: e.errors });
+ }
+ req.log.error({ err: e }, `trigger failed for ${project.id}`);
+ return reply.code(500).send({ error: 'could not create run', detail: String(e.message ?? e) });
+ }
+ });
+}
+
+function verifySignature(req, secret) {
+ const raw = req.rawBody ?? Buffer.alloc(0);
+
+ // GitHub and Gitea: sha256=<hex> over the body.
+ const hubSignature = req.headers['x-hub-signature-256'];
+ if (typeof hubSignature === 'string' && hubSignature.length > 0) {
+ const expected = `sha256=${crypto.createHmac('sha256', secret).update(raw).digest('hex')}`;
+ return timingSafeEqual(expected, hubSignature);
+ }
+
+ // GitLab: the secret itself, compared rather than signed.
+ const gitlabToken = req.headers['x-gitlab-token'];
+ if (typeof gitlabToken === 'string' && gitlabToken.length > 0) {
+ return timingSafeEqual(secret, gitlabToken);
+ }
+
+ return false;
+}
+
+function timingSafeEqual(a, b) {
+ const left = Buffer.from(String(a), 'utf8');
+ const right = Buffer.from(String(b), 'utf8');
+ if (left.length !== right.length) return false;
+ return crypto.timingSafeEqual(left, right);
+}
+
+// Reduces the supported payload shapes to { sha, base, ref, actor }.
+function normalizePush(body) {
+ if (!body || typeof body !== 'object') return null;
+
+ // hooks/post-receive, and anything else driving the API directly.
+ if (typeof body.sha === 'string') {
+ return {
+ sha: body.sha,
+ base: usableSha(body.base),
+ ref: typeof body.ref === 'string' ? body.ref : null,
+ actor: typeof body.actor === 'string' ? body.actor : null,
+ };
+ }
+
+ // GitLab sends both checkout_sha and after; checkout_sha is the one that
+ // refers to the tip of the pushed branch.
+ if (typeof body.checkout_sha === 'string') {
+ return {
+ sha: body.checkout_sha,
+ base: usableSha(body.before),
+ ref: typeof body.ref === 'string' ? body.ref : null,
+ actor: body.user_username ?? body.user_name ?? null,
+ };
+ }
+
+ // GitHub and Gitea.
+ if (typeof body.after === 'string') {
+ return {
+ sha: body.after,
+ base: usableSha(body.before),
+ ref: typeof body.ref === 'string' ? body.ref : null,
+ actor: body.pusher?.name ?? body.pusher?.login ?? body.sender?.login ?? null,
+ };
+ }
+
+ return null;
+}
+
+// A new branch reports an all zero parent, which is not a commit.
+function usableSha(value) {
+ if (typeof value !== 'string' || value === ZERO_SHA || !SHA_PATTERN.test(value)) return null;
+ return value;
+}
diff --git a/src/conductor/routes/workers.js b/src/conductor/routes/workers.js
@@ -0,0 +1,251 @@
+// src/conductor/routes/workers.js - the API workers poll and report to
+//
+// Every route here requires a worker token. A worker may only touch a job it
+// currently holds, which is checked against worker_token_id rather than
+// trusting the job id it sends.
+
+import { pipeline as streamPipeline } from 'node:stream/promises';
+import { LogOffsetError } from '../../lib/log.js';
+import { hashingTransform } from '../../lib/stream.js';
+import { sanitizeRelativePath, keys as storageKeys } from '../../lib/storage/index.js';
+import { newArtifactId } from '../../lib/ids.js';
+
+// Log chunks are small and frequent; artifacts are streamed, so this only
+// bounds a single log append.
+const LOG_CHUNK_LIMIT = 1024 * 1024;
+
+export default async function workerRoutes(fastify, { cfg, db, git, logs, storage, projects, scheduler, workerTokens }) {
+ // Raw bodies: log chunks and artifacts arrive as octet streams and must
+ // not be parsed.
+ fastify.addContentTypeParser('application/octet-stream', (req, payload, done) => done(null, payload));
+
+ async function requireWorker(req, reply) {
+ const header = req.headers.authorization || '';
+ const presented = header.startsWith('Bearer ') ? header.slice(7).trim() : '';
+ const token = await workerTokens.verify(presented, { ip: req.ip });
+ if (!token) {
+ reply.code(401).send({ error: 'invalid or missing worker token' });
+ return;
+ }
+ req.worker = token;
+ }
+
+ // Confirms the job exists and is held by the calling worker.
+ async function heldJob(req, reply) {
+ const job = await db.get(
+ `SELECT j.id, j.run_id, j.name, j.state, j.worker_token_id, j.log_size, r.project_id, r.head_sha
+ FROM jobs j JOIN runs r ON r.id = j.run_id
+ WHERE j.id = {id}`,
+ { id: req.params.id }
+ );
+ if (!job) {
+ reply.code(404).send({ error: 'unknown job' });
+ return null;
+ }
+ if (job.worker_token_id !== req.worker.id) {
+ reply.code(403).send({ error: 'job is not held by this worker' });
+ return null;
+ }
+ return job;
+ }
+
+ fastify.addHook('preHandler', requireWorker);
+
+ // Ask for work. The worker describes what it can run; the scheduler
+ // decides. Returns 204 when there is nothing to do.
+ fastify.get('/poll', async (req, reply) => {
+ const arches = splitList(req.query.arches);
+ const features = splitList(req.query.features);
+ const name = typeof req.query.name === 'string' ? req.query.name.slice(0, 255) : req.worker.name;
+
+ const job = await scheduler.claim({
+ arches,
+ features,
+ tokenId: req.worker.id,
+ workerName: name,
+ });
+ if (!job) return reply.code(204).send();
+
+ const project = await projects.get(job.project_id);
+ const base = cfg.server.public_url.replace(/\/+$/, '');
+
+ return reply.send({
+ job: {
+ id: job.id,
+ run_id: job.run_id,
+ name: job.name,
+ arch: job.arch,
+ image: job.image,
+ script: job.script,
+ env: job.env,
+ services: job.services,
+ artifacts: job.artifacts,
+ cache: job.cache,
+ timeout: job.timeout,
+ attempt: job.attempt,
+ sha: job.sha,
+ // 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.
+ source: project.source_mode === 'clone'
+ ? { mode: 'clone', repo_url: project.repo_url, sha: job.sha }
+ : { mode: 'archive', url: `${base}/api/workers/jobs/${encodeURIComponent(job.id)}/source.tar.gz` },
+ endpoints: {
+ log: `${base}/api/workers/jobs/${encodeURIComponent(job.id)}/log`,
+ artifact: `${base}/api/workers/jobs/${encodeURIComponent(job.id)}/artifact`,
+ heartbeat: `${base}/api/workers/jobs/${encodeURIComponent(job.id)}/heartbeat`,
+ done: `${base}/api/workers/jobs/${encodeURIComponent(job.id)}/done`,
+ },
+ },
+ });
+ });
+
+ // The tree at the job's commit, as a gzipped tar.
+ fastify.get('/jobs/:id/source.tar.gz', async (req, reply) => {
+ const job = await heldJob(req, reply);
+ if (!job) return reply;
+
+ const project = await projects.get(job.project_id);
+ if (!project) return reply.code(404).send({ error: 'unknown project' });
+
+ reply.header('content-type', 'application/gzip');
+ reply.header('content-disposition', `attachment; filename="${job.head_sha.slice(0, 12)}.tar.gz"`);
+ return reply.send(git.archiveStream(project, job.head_sha));
+ });
+
+ // Append to the live log. X-Log-Offset makes a retry after a dropped
+ // connection safe.
+ fastify.post('/jobs/:id/log', { bodyLimit: LOG_CHUNK_LIMIT }, async (req, reply) => {
+ const job = await heldJob(req, reply);
+ if (!job) return reply;
+
+ const chunks = [];
+ let total = 0;
+ for await (const chunk of req.body) {
+ total += chunk.length;
+ if (total > LOG_CHUNK_LIMIT) return reply.code(413).send({ error: 'log chunk too large' });
+ chunks.push(chunk);
+ }
+
+ const rawOffset = req.headers['x-log-offset'];
+ const offset = rawOffset === undefined ? undefined : Number(rawOffset);
+ if (offset !== undefined && !Number.isInteger(offset)) {
+ return reply.code(400).send({ error: 'x-log-offset must be an integer' });
+ }
+
+ try {
+ const result = await logs.append(job.run_id, job.id, Buffer.concat(chunks), offset);
+ await db.run('UPDATE jobs SET log_size = {size}, heartbeat_at = {now} WHERE id = {id}',
+ { id: job.id, size: result.size, now: Date.now() });
+ return reply.send(result);
+ } catch (e) {
+ if (e instanceof LogOffsetError) {
+ // Tell the worker where to resume from.
+ return reply.code(409).send({ error: e.message, expected_offset: e.expected });
+ }
+ throw e;
+ }
+ });
+
+ // Upload one artifact. The path is worker supplied, so it is sanitized
+ // before it can influence a storage key.
+ fastify.post('/jobs/:id/artifact', async (req, reply) => {
+ const job = await heldJob(req, reply);
+ if (!job) return reply;
+
+ const relPath = sanitizeRelativePath(req.headers['x-artifact-path']);
+ if (!relPath) return reply.code(400).send({ error: 'x-artifact-path is missing or unusable' });
+
+ const declared = Number(req.headers['content-length']);
+ if (!Number.isInteger(declared) || declared < 0) {
+ return reply.code(411).send({ error: 'content-length is required for artifact uploads' });
+ }
+
+ // Hash on the way past, so the digest costs nothing extra even when the
+ // backend cannot report one.
+ const hasher = hashingTransform();
+ const key = storageKeys.artifact(job.run_id, job.id, relPath);
+
+ let stored;
+ try {
+ [, stored] = await Promise.all([
+ streamPipeline(req.body, hasher),
+ storage.put(key, hasher, { size: declared }),
+ ]);
+ } catch (e) {
+ // A body shorter or longer than content-length is the worker's
+ // mistake, not a server fault.
+ if (/size mismatch/.test(e.message)) {
+ return reply.code(400).send({ error: e.message });
+ }
+ throw e;
+ }
+
+ const digest = hasher.digest();
+ await db.run(
+ `INSERT INTO artifacts (id, job_id, run_id, path, storage_key, size, sha256, created_at)
+ VALUES ({id}, {job}, {run}, {path}, {key}, {size}, {sha}, {now})`,
+ {
+ id: newArtifactId(),
+ job: job.id,
+ run: job.run_id,
+ path: relPath,
+ key,
+ size: stored.size ?? declared,
+ sha: digest,
+ now: Date.now(),
+ }
+ );
+
+ return reply.send({ path: relPath, size: stored.size ?? declared, sha256: digest });
+ });
+
+ // Keeps the job alive, and is how a worker learns it should stop.
+ fastify.post('/jobs/:id/heartbeat', async (req, reply) => {
+ const result = await scheduler.heartbeat(req.params.id, req.worker.id);
+ if (!result.known) return reply.code(404).send({ error: 'unknown job for this worker' });
+ return reply.send({ cancelled: result.cancelled, state: result.state });
+ });
+
+ // Report the outcome. The log is copied to storage here, whatever the
+ // result, so a failed job keeps its output.
+ fastify.post('/jobs/:id/done', async (req, reply) => {
+ const job = await heldJob(req, reply);
+ if (!job) return reply;
+
+ const body = req.body ?? {};
+ const success = body.success === true;
+ const exitCode = Number.isInteger(body.exit_code) ? body.exit_code : null;
+ const error = typeof body.error === 'string' ? body.error.slice(0, 4096) : null;
+
+ const outcome = await scheduler.complete(job.id, {
+ success,
+ exitCode,
+ error,
+ tokenId: req.worker.id,
+ });
+ if (!outcome.ok) return reply.code(409).send({ error: outcome.reason });
+
+ // A retry keeps writing to the same log, so only archive it once the
+ // job has actually stopped.
+ if (outcome.state !== 'queued') {
+ try {
+ await scheduler.finalizeLog(job.run_id, job.id);
+ } catch (e) {
+ req.log.error({ err: e }, `failed to archive log for ${job.id}`);
+ }
+ }
+
+ return reply.send({
+ state: outcome.state,
+ retry: outcome.retry === true,
+ skipped: outcome.skipped ?? [],
+ run_state: outcome.runState ?? null,
+ });
+ });
+}
+
+function splitList(value) {
+ if (typeof value !== 'string') return [];
+ return value.split(',').map((s) => s.trim()).filter(Boolean).slice(0, 64);
+}
diff --git a/src/conductor/scheduler.js b/src/conductor/scheduler.js
@@ -0,0 +1,418 @@
+// src/conductor/scheduler.js - run creation, dispatch and completion
+//
+// Scheduling reads the dependency edges in job_deps directly. Two defects in
+// the prototype came from not doing that:
+//
+// Dispatch was serialized. A single integer level per job was compared
+// against the lowest queued level, so only one job could run at a time
+// even when the graph allowed the whole width of it. Here any job whose
+// dependencies are all satisfied is eligible, and several workers can
+// claim different jobs concurrently.
+//
+// Failure skipped only direct dependents. Transitive dependents stayed
+// queued and were later dispatched with their inputs missing. Here the
+// whole reachable set is skipped.
+//
+// Claiming is a conditional UPDATE guarded on the previous state, so two
+// workers polling at the same instant cannot both win the same job.
+
+import { compilePipeline, PipelineError, transitiveDependents } from '../lib/pipeline/index.js';
+import { inClause } from '../lib/db/query.js';
+import { newRunId, jobId as makeJobId } from '../lib/ids.js';
+import { keys as storageKeys } from '../lib/storage/index.js';
+
+export const RUN_STATES = ['pending', 'running', 'success', 'failed', 'cancelled'];
+export const JOB_STATES = ['queued', 'running', 'success', 'failed', 'skipped', 'cancelled'];
+
+// States that let a dependent proceed.
+const TERMINAL = ['success', 'failed', 'skipped', 'cancelled'];
+
+export function createScheduler({ cfg, db, git, logs, storage, projects, logger = console }) {
+ // A job is eligible when it is queued, its run is live, and no dependency
+ // is outstanding. A failed dependency marked allow_failure still counts as
+ // 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
+ FROM jobs j
+ JOIN runs r ON r.id = j.run_id
+ WHERE j.state = 'queued'
+ AND r.state = 'running'
+ AND NOT EXISTS (
+ SELECT 1
+ FROM job_deps d
+ JOIN jobs dj ON dj.id = d.depends_on_id
+ WHERE d.job_id = j.id
+ AND dj.state <> 'success'
+ AND NOT (dj.state = 'failed' AND dj.allow_failure = 1)
+ )
+ `;
+
+ async function loadGraph(tx, runId) {
+ const jobs = await tx.all(
+ 'SELECT id, name, state, allow_failure FROM jobs WHERE run_id = {run}',
+ { run: runId }
+ );
+ const deps = await tx.all(
+ `SELECT d.job_id, d.depends_on_id
+ FROM job_deps d
+ JOIN jobs j ON j.id = d.job_id
+ WHERE j.run_id = {run}`,
+ { run: runId }
+ );
+ const needs = new Map(jobs.map((j) => [j.id, []]));
+ for (const d of deps) needs.get(d.job_id)?.push(d.depends_on_id);
+ return jobs.map((j) => ({ ...j, needs: needs.get(j.id) ?? [] }));
+ }
+
+ // Marks the run finished once nothing is left to do.
+ async function settleRun(tx, runId) {
+ const outstanding = await tx.get(
+ "SELECT COUNT(*) AS c FROM jobs WHERE run_id = {run} AND state IN ('queued', 'running')",
+ { run: runId }
+ );
+ if (outstanding.c > 0) return null;
+
+ const bad = await tx.get(
+ `SELECT COUNT(*) AS c FROM jobs
+ WHERE run_id = {run}
+ AND (state IN ('skipped', 'cancelled')
+ OR (state = 'failed' AND allow_failure = 0))`,
+ { run: runId }
+ );
+ const state = bad.c > 0 ? 'failed' : 'success';
+ await tx.run(
+ "UPDATE runs SET state = {state}, finished_at = {now} WHERE id = {run} AND state = 'running'",
+ { state, run: runId, now: Date.now() }
+ );
+ return state;
+ }
+
+ return {
+ // Reads the pipeline at the pushed commit and records the run. The
+ // repository is the source of truth, so nothing the trigger claims about
+ // the contents is trusted.
+ async createRun(project, { ref, baseSha, headSha, trigger = 'push', actor = null }) {
+ await git.sync(project, { force: true });
+
+ if (!(await git.hasCommit(project, headSha))) {
+ throw new Error(`commit ${headSha} is not present in the mirror of ${project.id}`);
+ }
+
+ const text = await git.readFile(project, headSha, project.config_path);
+ if (text === null) {
+ throw new PipelineError(
+ [{ path: '', message: `${project.config_path} not found at ${headSha.slice(0, 12)}` }],
+ project.config_path
+ );
+ }
+
+ const pipeline = compilePipeline(text, {
+ source: project.config_path,
+ defaultTimeout: cfg.scheduler.default_job_timeout,
+ defaultAttempts: 1,
+ });
+
+ const info = await git.commitInfo(project, headSha).catch(() => ({ subject: null }));
+ const runId = newRunId();
+ const now = Date.now();
+
+ await db.transaction(async (tx) => {
+ const number = await projects.nextRunNumber(tx, project.id);
+ const empty = pipeline.jobs.length === 0;
+
+ await tx.run(
+ `INSERT INTO runs
+ (id, project_id, number, ref, base_sha, head_sha, trigger_type, actor,
+ title, state, pipeline, created_at, started_at, finished_at)
+ VALUES
+ ({id}, {project}, {number}, {ref}, {base}, {head}, {trigger}, {actor},
+ {title}, {state}, {pipeline}, {now}, {now}, {finished})`,
+ {
+ id: runId,
+ project: project.id,
+ number,
+ ref: ref ?? null,
+ base: baseSha ?? null,
+ head: headSha,
+ trigger,
+ actor,
+ title: info.subject ?? null,
+ state: empty ? 'success' : 'running',
+ pipeline: JSON.stringify({ version: pipeline.version, jobs: pipeline.jobs }),
+ now,
+ finished: empty ? now : null,
+ }
+ );
+
+ const idByName = new Map(pipeline.jobs.map((j) => [j.name, makeJobId(runId, j.name)]));
+
+ for (const job of pipeline.jobs) {
+ await tx.run(
+ `INSERT INTO jobs
+ (id, run_id, name, base_name, arch, image, requires, spec, state,
+ allow_failure, attempt, max_attempts, timeout, log_size, created_at)
+ VALUES
+ ({id}, {run}, {name}, {base}, {arch}, {image}, {requires}, {spec}, 'queued',
+ {allow}, 0, {attempts}, {timeout}, 0, {now})`,
+ {
+ id: idByName.get(job.name),
+ run: runId,
+ name: job.name,
+ base: job.baseName,
+ arch: job.arch,
+ image: job.image,
+ requires: JSON.stringify(job.requires),
+ spec: JSON.stringify({
+ script: job.script,
+ env: job.env,
+ services: job.services,
+ artifacts: job.artifacts,
+ cache: job.cache,
+ matrix: job.matrix,
+ needs: job.needs,
+ depth: job.depth,
+ }),
+ allow: job.allow_failure ? 1 : 0,
+ attempts: job.max_attempts,
+ timeout: job.timeout,
+ now,
+ }
+ );
+ }
+
+ for (const job of pipeline.jobs) {
+ for (const need of job.needs) {
+ await tx.run(
+ 'INSERT INTO job_deps (job_id, depends_on_id) VALUES ({job}, {dep})',
+ { job: idByName.get(job.name), dep: idByName.get(need) }
+ );
+ }
+ }
+ });
+
+ logger.info?.(`run ${runId} created for ${project.id} with ${pipeline.jobs.length} job(s)`);
+ return { runId, jobCount: pipeline.jobs.length };
+ },
+
+ // Hands one job to a worker. The worker advertises what it can run;
+ // jobs asking for anything it lacks are passed over.
+ async claim({ arches = [], features = [], tokenId, workerName }) {
+ const archFilter = inClause('arch', arches);
+ const featureSet = new Set(features);
+
+ // Oldest run first, so a queue drains in order rather than starving
+ // whichever run happens to sort last.
+ const candidates = await db.all(
+ `${ELIGIBLE}
+ AND (j.arch IS NULL${archFilter.empty ? '' : ` OR j.arch IN (${archFilter.sql})`})
+ ORDER BY r.created_at ASC, j.created_at ASC
+ LIMIT 100`,
+ { ...archFilter.params }
+ );
+
+ for (const candidate of candidates) {
+ let requires;
+ try {
+ requires = JSON.parse(candidate.requires);
+ } catch {
+ requires = [];
+ }
+ if (!requires.every((f) => featureSet.has(f))) continue;
+
+ const now = Date.now();
+ const claimed = await db.run(
+ `UPDATE jobs
+ SET state = 'running', attempt = attempt + 1, worker_token_id = {token},
+ worker_name = {worker}, claimed_at = {now}, heartbeat_at = {now},
+ started_at = {now}
+ WHERE id = {id} AND state = 'queued'`,
+ { id: candidate.id, token: tokenId, worker: workerName ?? null, now }
+ );
+ // Lost the race to another worker; try the next candidate.
+ if (claimed.changes !== 1) continue;
+
+ const spec = JSON.parse(candidate.spec);
+ return {
+ id: candidate.id,
+ run_id: candidate.run_id,
+ project_id: candidate.project_id,
+ name: candidate.name,
+ arch: candidate.arch,
+ image: candidate.image,
+ sha: candidate.head_sha,
+ requires,
+ timeout: candidate.timeout,
+ attempt: candidate.attempt + 1,
+ max_attempts: candidate.max_attempts,
+ script: spec.script,
+ env: spec.env,
+ services: spec.services,
+ artifacts: spec.artifacts,
+ cache: spec.cache,
+ };
+ }
+
+ return null;
+ },
+
+ async heartbeat(jobId, tokenId) {
+ const job = await db.get(
+ 'SELECT id, state, worker_token_id FROM jobs WHERE id = {id}',
+ { id: jobId }
+ );
+ if (!job) return { known: false };
+ if (job.worker_token_id !== tokenId) return { known: false };
+
+ if (job.state === 'running') {
+ await db.run('UPDATE jobs SET heartbeat_at = {now} WHERE id = {id}', { id: jobId, now: Date.now() });
+ return { known: true, cancelled: false };
+ }
+ // The job was cancelled or reaped out from under the worker, which is
+ // how a worker learns to stop.
+ return { known: true, cancelled: true, state: job.state };
+ },
+
+ // Records the outcome, retries if attempts remain, and otherwise skips
+ // everything downstream.
+ async complete(jobId, { success, exitCode = null, error = null, tokenId = null }) {
+ return db.transaction(async (tx) => {
+ const job = await tx.get(
+ `SELECT id, run_id, name, state, attempt, max_attempts, allow_failure, worker_token_id
+ FROM jobs WHERE id = {id}`,
+ { id: jobId }
+ );
+ if (!job) return { ok: false, reason: 'unknown job' };
+ if (tokenId !== null && job.worker_token_id !== tokenId) {
+ return { ok: false, reason: 'job belongs to another worker' };
+ }
+ if (job.state !== 'running') {
+ return { ok: false, reason: `job is ${job.state}, not running` };
+ }
+
+ const now = Date.now();
+
+ if (success) {
+ await tx.run(
+ "UPDATE jobs SET state = 'success', exit_code = {code}, finished_at = {now} WHERE id = {id}",
+ { id: jobId, code: exitCode, now }
+ );
+ const runState = await settleRun(tx, job.run_id);
+ return { ok: true, state: 'success', runState };
+ }
+
+ // Transient failure with attempts left: back to the queue rather
+ // than failing the run.
+ if (job.attempt < job.max_attempts) {
+ await tx.run(
+ `UPDATE jobs
+ SET state = 'queued', exit_code = {code}, error = {error},
+ worker_token_id = NULL, worker_name = NULL,
+ claimed_at = NULL, heartbeat_at = NULL, started_at = NULL
+ WHERE id = {id}`,
+ { id: jobId, code: exitCode, error }
+ );
+ return { ok: true, state: 'queued', retry: true, attempt: job.attempt };
+ }
+
+ await tx.run(
+ "UPDATE jobs SET state = 'failed', exit_code = {code}, error = {error}, finished_at = {now} WHERE id = {id}",
+ { id: jobId, code: exitCode, error, now }
+ );
+
+ let skipped = [];
+ if (!job.allow_failure) {
+ skipped = await skipDependents(tx, job.run_id, [jobId], now);
+ }
+
+ const runState = await settleRun(tx, job.run_id);
+ return { ok: true, state: 'failed', skipped, runState };
+ });
+ },
+
+ // Every job reachable from the failed ones, not just their immediate
+ // dependents.
+ async skipDependents(runId, fromJobIds) {
+ return db.transaction((tx) => skipDependents(tx, runId, fromJobIds, Date.now()));
+ },
+
+ // Requeues or fails jobs whose worker stopped reporting.
+ async reap() {
+ const cutoff = Date.now() - cfg.scheduler.heartbeat_timeout * 1000;
+ const stale = await db.all(
+ `SELECT id, run_id, name, attempt, max_attempts, worker_name
+ FROM jobs
+ WHERE state = 'running' AND heartbeat_at IS NOT NULL AND heartbeat_at < {cutoff}`,
+ { cutoff }
+ );
+
+ const results = [];
+ for (const job of stale) {
+ const outcome = await this.complete(job.id, {
+ success: false,
+ error: `worker ${job.worker_name ?? 'unknown'} stopped reporting for more than ` +
+ `${cfg.scheduler.heartbeat_timeout}s`,
+ });
+ if (outcome.ok) {
+ logger.warn?.(`reaped ${job.name} (${job.id}): ${outcome.retry ? 'requeued' : 'failed'}`);
+ results.push({ id: job.id, ...outcome });
+ }
+ }
+ return results;
+ },
+
+ async cancelRun(runId, reason = 'cancelled by request') {
+ return db.transaction(async (tx) => {
+ const run = await tx.get('SELECT id, state FROM runs WHERE id = {id}', { id: runId });
+ if (!run) return { ok: false, reason: 'unknown run' };
+ if (run.state !== 'running') return { ok: false, reason: `run is ${run.state}` };
+
+ const now = Date.now();
+ await tx.run(
+ `UPDATE jobs SET state = 'cancelled', error = {reason}, finished_at = {now}
+ WHERE run_id = {run} AND state IN ('queued', 'running')`,
+ { run: runId, reason, now }
+ );
+ await tx.run(
+ "UPDATE runs SET state = 'cancelled', finished_at = {now} WHERE id = {run}",
+ { run: runId, now }
+ );
+ return { ok: true };
+ });
+ },
+
+ // Called once a job finishes so the live log stops being the spool copy.
+ async finalizeLog(runId, jobId) {
+ const key = storageKeys.log(runId, jobId);
+ const { size } = await logs.finalize(runId, jobId, storage, key);
+ if (size > 0) {
+ await db.run('UPDATE jobs SET log_key = {key}, log_size = {size} WHERE id = {id}',
+ { id: jobId, key, size });
+ }
+ return { key, size };
+ },
+
+ TERMINAL,
+ };
+
+ async function skipDependents(tx, runId, fromJobIds, now) {
+ const graph = await loadGraph(tx, runId);
+ const reachable = transitiveDependents(
+ graph.map((j) => ({ name: j.id, needs: j.needs })),
+ fromJobIds
+ );
+ if (reachable.size === 0) return [];
+
+ const queued = new Set(graph.filter((j) => j.state === 'queued').map((j) => j.id));
+ const toSkip = [...reachable].filter((id) => queued.has(id));
+
+ for (const id of toSkip) {
+ await tx.run(
+ `UPDATE jobs SET state = 'skipped', error = {error}, finished_at = {now}
+ WHERE id = {id} AND state = 'queued'`,
+ { id, error: 'skipped because a dependency did not succeed', now }
+ );
+ }
+ return toSkip;
+ }
+}
diff --git a/src/lib/config.js b/src/lib/config.js
@@ -185,11 +185,13 @@ export function parseKey(value, label) {
function validate(cfg) {
const errors = [];
- if (!Number.isInteger(cfg.server.port) || cfg.server.port < 1 || cfg.server.port > 65535) {
- errors.push(`server.port must be between 1 and 65535, got ${cfg.server.port}`);
- }
- if (!Number.isInteger(cfg.read_api.port) || cfg.read_api.port < 1 || cfg.read_api.port > 65535) {
- errors.push(`read_api.port must be between 1 and 65535, got ${cfg.read_api.port}`);
+ // Port 0 asks the operating system for a free port, which is useful in
+ // tests and for ephemeral instances.
+ for (const key of ['server.port', 'read_api.port']) {
+ const port = getPath(cfg, key);
+ if (!Number.isInteger(port) || port < 0 || port > 65535) {
+ errors.push(`${key} must be between 0 and 65535, got ${port}`);
+ }
}
if (cfg.database.url && dialectFromUrl(cfg.database.url) === null) {
errors.push(
diff --git a/src/lib/db/migrate-cli.js b/src/lib/db/migrate-cli.js
@@ -0,0 +1,25 @@
+// src/lib/db/migrate-cli.js - apply pending migrations
+//
+// Usage: npm run migrate
+//
+// The conductor also migrates on startup. This exists for deployments that
+// want the schema change to be a separate, reviewable step.
+
+import { loadConfig, ensureStateDirs } from '../config.js';
+import { openDatabase } from './index.js';
+import { runMigrations } from './migrate.js';
+
+const cfg = loadConfig();
+ensureStateDirs(cfg);
+
+const db = await openDatabase(cfg);
+try {
+ console.log(`[migrate] dialect ${db.dialect}`);
+ const result = await runMigrations(db);
+ if (result.applied.length === 0) console.log('[migrate] nothing to do');
+} catch (e) {
+ console.error(`[migrate] ${e.message}`);
+ process.exitCode = 1;
+} finally {
+ await db.close();
+}
diff --git a/src/lib/db/query.js b/src/lib/db/query.js
@@ -156,6 +156,26 @@ export function compileCached(sql, dialect) {
return entry;
}
+// Builds the body of an IN (...) clause with generated marker names, since
+// the number of values is only known at runtime.
+//
+// const arch = inClause('arch', ['x86_64', 'aarch64']);
+// db.all(`... WHERE arch IN (${arch.sql})`, { ...arch.params });
+//
+// Returns a clause that matches nothing when the list is empty, which is
+// the safe reading of "none of these".
+export function inClause(prefix, values) {
+ if (!Array.isArray(values) || values.length === 0) {
+ return { sql: 'NULL', params: {}, empty: true };
+ }
+ const params = {};
+ const markers = values.map((value, i) => {
+ params[`${prefix}${i}`] = value;
+ return `{${prefix}${i}}`;
+ });
+ return { sql: markers.join(', '), params, empty: false };
+}
+
function resolve(params, key) {
// An exact property wins, so a pre-flattened object works unchanged.
if (params != null && Object.hasOwn(params, key)) {
diff --git a/src/lib/git.js b/src/lib/git.js
@@ -0,0 +1,212 @@
+// src/lib/git.js - repository mirrors
+//
+// The conductor keeps one bare mirror per project. Everything it needs from
+// a push is read out of that mirror: the pipeline file at the pushed commit,
+// the commit subject, and the source tarball handed to workers.
+//
+// Serving the source from here rather than having workers clone means an
+// untrusted worker needs no repository credentials and cannot reach any ref
+// other than the commit it was given work for. It also means the worker host
+// does not need git installed.
+//
+// Every value interpolated into a git invocation is validated first. Refs
+// and object ids arrive from webhooks, and a value beginning with a hyphen
+// would otherwise be read as an option.
+
+import fs from 'node:fs/promises';
+import path from 'node:path';
+import { execFile, spawn } from 'node:child_process';
+import { promisify } from 'node:util';
+
+const execFileAsync = promisify(execFile);
+
+export const SHA_PATTERN = /^[0-9a-f]{40}$/;
+export const SHORT_SHA_PATTERN = /^[0-9a-f]{7,40}$/;
+
+// Refuses credential prompts, so a misconfigured private repository fails
+// promptly instead of hanging until the request times out.
+const GIT_ENV = {
+ ...process.env,
+ GIT_TERMINAL_PROMPT: '0',
+ GIT_ASKPASS: '',
+ GCM_INTERACTIVE: 'never',
+ LC_ALL: 'C',
+};
+
+export class GitError extends Error {
+ constructor(message, { stderr, code } = {}) {
+ super(stderr ? `${message}: ${String(stderr).trim().split('\n').slice(0, 3).join('; ')}` : message);
+ this.name = 'GitError';
+ this.stderr = stderr;
+ this.exitCode = code;
+ }
+}
+
+export function assertSha(sha, label = 'commit') {
+ if (typeof sha !== 'string' || !SHA_PATTERN.test(sha)) {
+ throw new GitError(`${label} must be a full 40 character hex object id, got ${JSON.stringify(sha)}`);
+ }
+ return sha;
+}
+
+// Refs come straight from a push hook. Reject anything git itself would
+// consider malformed, along with leading hyphens.
+export function assertRef(ref) {
+ if (typeof ref !== 'string' || ref.length === 0 || ref.length > 255) {
+ throw new GitError(`invalid ref: ${JSON.stringify(ref)}`);
+ }
+ if (ref.startsWith('-') || ref.includes('..') || /[\s~^:?*[\\]/.test(ref) || ref.endsWith('.lock')) {
+ throw new GitError(`invalid ref: ${JSON.stringify(ref)}`);
+ }
+ return ref;
+}
+
+// A path inside the repository, used for the pipeline file.
+export function assertRepoPath(p) {
+ if (typeof p !== 'string' || p.length === 0 || p.length > 512) {
+ throw new GitError(`invalid repository path: ${JSON.stringify(p)}`);
+ }
+ if (p.startsWith('/') || p.startsWith('-') || p.split('/').some((s) => s === '' || s === '.' || s === '..')) {
+ throw new GitError(`invalid repository path: ${JSON.stringify(p)}`);
+ }
+ return p;
+}
+
+// Serializes work per mirror, so two pushes to one project cannot run
+// concurrent fetches into the same directory.
+const locks = new Map();
+
+async function withLock(key, fn) {
+ const previous = locks.get(key) ?? Promise.resolve();
+ let release;
+ const current = new Promise((resolve) => { release = resolve; });
+ locks.set(key, previous.then(() => current));
+ await previous;
+ try {
+ return await fn();
+ } finally {
+ release();
+ if (locks.get(key) === current) locks.delete(key);
+ }
+}
+
+export function createGit(cfg) {
+ const timeout = cfg.git.timeout * 1000;
+ const fetchInterval = cfg.git.fetch_interval * 1000;
+ const lastFetch = new Map();
+
+ function mirrorPath(projectId) {
+ return path.join(cfg.git.mirror_path, `${projectId}.git`);
+ }
+
+ async function run(args, options = {}) {
+ try {
+ const { stdout, stderr } = await execFileAsync('git', args, {
+ env: GIT_ENV,
+ timeout,
+ maxBuffer: options.maxBuffer ?? 16 * 1024 * 1024,
+ encoding: options.encoding ?? 'utf8',
+ });
+ return { stdout, stderr };
+ } catch (e) {
+ throw new GitError(`git ${args[0]} failed`, { stderr: e.stderr || e.message, code: e.code });
+ }
+ }
+
+ async function exists(dir) {
+ try {
+ await fs.stat(path.join(dir, 'HEAD'));
+ return true;
+ } catch {
+ return false;
+ }
+ }
+
+ return {
+ mirrorPath,
+
+ // Clones on first use, then fetches at most once per fetch_interval
+ // unless force is set. A trigger always forces, since it needs the
+ // commit that was just pushed.
+ async sync(project, { force = false } = {}) {
+ const dir = mirrorPath(project.id);
+ return withLock(dir, async () => {
+ if (!(await exists(dir))) {
+ await fs.mkdir(path.dirname(dir), { recursive: true });
+ await run(['clone', '--mirror', '--quiet', '--', project.repo_url, dir]);
+ lastFetch.set(dir, Date.now());
+ return dir;
+ }
+
+ const since = Date.now() - (lastFetch.get(dir) ?? 0);
+ if (!force && since < fetchInterval) return dir;
+
+ await run(['--git-dir', dir, 'fetch', '--prune', '--quiet', 'origin']);
+ lastFetch.set(dir, Date.now());
+ return dir;
+ });
+ },
+
+ async hasCommit(project, sha) {
+ assertSha(sha);
+ try {
+ const { stdout } = await run(['--git-dir', mirrorPath(project.id), 'cat-file', '-t', sha]);
+ return stdout.trim() === 'commit';
+ } catch {
+ return false;
+ }
+ },
+
+ async resolve(project, rev) {
+ assertRef(rev);
+ const { stdout } = await run(['--git-dir', mirrorPath(project.id), 'rev-parse', '--verify', `${rev}^{commit}`]);
+ return stdout.trim();
+ },
+
+ // Reads one file at a commit. Returns null when the path is absent,
+ // which is how a repository without a pipeline is detected.
+ async readFile(project, sha, filePath) {
+ assertSha(sha);
+ assertRepoPath(filePath);
+ try {
+ const { stdout } = await run(
+ ['--git-dir', mirrorPath(project.id), 'cat-file', 'blob', `${sha}:${filePath}`],
+ { maxBuffer: 4 * 1024 * 1024 }
+ );
+ return stdout;
+ } catch (e) {
+ if (/does not exist|not a valid object name|Not a valid object/i.test(e.stderr ?? '')) return null;
+ throw e;
+ }
+ },
+
+ async commitInfo(project, sha) {
+ assertSha(sha);
+ // A unit separator keeps the fields unambiguous when a subject
+ // contains anything at all.
+ const { stdout } = await run([
+ '--git-dir', mirrorPath(project.id),
+ 'show', '--no-patch', '--format=%s%x1f%an%x1f%ae%x1f%aI', sha,
+ ]);
+ const [subject = '', authorName = '', authorEmail = '', authoredAt = ''] = stdout.trim().split('\x1f');
+ return { subject, authorName, authorEmail, authoredAt };
+ },
+
+ // A gzipped tar of the tree at a commit, streamed rather than buffered.
+ // The caller pipes this straight to the worker.
+ archiveStream(project, sha, { prefix = '' } = {}) {
+ assertSha(sha);
+ const args = ['--git-dir', mirrorPath(project.id), 'archive', '--format=tar.gz'];
+ if (prefix) args.push(`--prefix=${prefix}`);
+ args.push(sha);
+
+ const child = spawn('git', args, { env: GIT_ENV, stdio: ['ignore', 'pipe', 'pipe'] });
+ let stderr = '';
+ child.stderr.on('data', (chunk) => { stderr += chunk.toString().slice(0, 4096); });
+ child.on('close', (code) => {
+ if (code !== 0) child.stdout.destroy(new GitError('git archive failed', { stderr, code }));
+ });
+ return child.stdout;
+ },
+ };
+}
diff --git a/src/lib/log.js b/src/lib/log.js
@@ -0,0 +1,163 @@
+// src/lib/log.js - job log spool
+//
+// Logs are written to local disk while a job runs, then copied to object
+// storage when it finishes. The spool is always local even when S3 is
+// configured, because a running job produces many small appends and a live
+// tail wants cheap random reads; neither suits an object store.
+//
+// Appends carry the offset the worker believes it is writing at, which makes
+// a retry after a dropped connection safe: an overlapping chunk is trimmed
+// and a gap is refused so the worker can resend from the right place.
+
+import fs from 'node:fs/promises';
+import path from 'node:path';
+
+export class LogOffsetError extends Error {
+ constructor(message, expected) {
+ super(message);
+ this.name = 'LogOffsetError';
+ this.expected = expected;
+ }
+}
+
+const TRUNCATION_NOTICE = '\n[conductor] log truncated: size limit reached\n';
+
+export function createLogStore(cfg) {
+ const root = cfg.log.spool_path;
+ const maxSize = cfg.log.max_size;
+
+ // One writer per job, so a retried append cannot interleave with the
+ // original.
+ const locks = new Map();
+
+ async function withLock(key, fn) {
+ const previous = locks.get(key) ?? Promise.resolve();
+ let release;
+ const current = new Promise((resolve) => { release = resolve; });
+ locks.set(key, previous.then(() => current));
+ await previous;
+ try {
+ return await fn();
+ } finally {
+ release();
+ if (locks.get(key) === current) locks.delete(key);
+ }
+ }
+
+ // Job ids contain a colon, which is fine on every filesystem we target,
+ // but the run id still gives a directory per run to keep listings sane.
+ function spoolPath(runId, jobId) {
+ return path.join(root, encodeURIComponent(runId), `${encodeURIComponent(jobId)}.log`);
+ }
+
+ async function currentSize(file) {
+ try {
+ return (await fs.stat(file)).size;
+ } catch (e) {
+ if (e.code === 'ENOENT') return 0;
+ throw e;
+ }
+ }
+
+ return {
+ spoolPath,
+
+ async size(runId, jobId) {
+ return currentSize(spoolPath(runId, jobId));
+ },
+
+ // Returns the new total size. When offset is omitted the chunk is simply
+ // appended, which is what a worker that never retries will do.
+ async append(runId, jobId, chunk, offset) {
+ const file = spoolPath(runId, jobId);
+ const key = `${runId}/${jobId}`;
+ const data = Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk), 'utf8');
+
+ return withLock(key, async () => {
+ const size = await currentSize(file);
+
+ let payload = data;
+ if (offset !== undefined && offset !== null) {
+ if (offset > size) {
+ throw new LogOffsetError(
+ `log append starts past the end of the log: got offset ${offset}, have ${size}`,
+ size
+ );
+ }
+ if (offset < size) {
+ // The worker is resending something already stored. Keep only
+ // the part beyond what we have.
+ const overlap = size - offset;
+ if (overlap >= payload.length) return { size, written: 0 };
+ payload = payload.subarray(overlap);
+ }
+ }
+
+ if (size >= maxSize) return { size, written: 0, truncated: true };
+
+ let truncated = false;
+ if (size + payload.length > maxSize) {
+ payload = Buffer.concat([
+ payload.subarray(0, Math.max(0, maxSize - size)),
+ Buffer.from(TRUNCATION_NOTICE, 'utf8'),
+ ]);
+ truncated = true;
+ }
+
+ await fs.mkdir(path.dirname(file), { recursive: true });
+ await fs.appendFile(file, payload);
+ return { size: size + payload.length, written: payload.length, truncated };
+ });
+ },
+
+ // Reads a window for the live tail. Returns the total size too, so the
+ // caller knows whether more is already available.
+ async read(runId, jobId, { offset = 0, limit = 256 * 1024 } = {}) {
+ const file = spoolPath(runId, jobId);
+ const size = await currentSize(file);
+ if (size === 0 || offset >= size) {
+ return { data: Buffer.alloc(0), offset: Math.min(offset, size), size };
+ }
+
+ const start = Math.max(0, offset);
+ const length = Math.min(limit, size - start);
+ const handle = await fs.open(file, 'r');
+ try {
+ const buffer = Buffer.alloc(length);
+ const { bytesRead } = await handle.read(buffer, 0, length, start);
+ return { data: buffer.subarray(0, bytesRead), offset: start, size };
+ } finally {
+ await handle.close();
+ }
+ },
+
+ // Copies the finished log into object storage and drops the spool copy.
+ // Storage is the durable home; the spool only exists to serve a tail.
+ async finalize(runId, jobId, storage, storageKey) {
+ const file = spoolPath(runId, jobId);
+ const size = await currentSize(file);
+ if (size === 0) return { key: null, size: 0 };
+
+ const handle = await fs.open(file, 'r');
+ try {
+ await storage.put(storageKey, handle.createReadStream(), {
+ size,
+ contentType: 'text/plain; charset=utf-8',
+ });
+ } finally {
+ await handle.close();
+ }
+
+ await fs.rm(file, { force: true });
+ return { key: storageKey, size };
+ },
+
+ async remove(runId, jobId) {
+ await fs.rm(spoolPath(runId, jobId), { force: true });
+ },
+
+ async removeRun(runId) {
+ await fs.rm(path.join(root, encodeURIComponent(runId)), { recursive: true, force: true });
+ },
+ };
+}
diff --git a/src/lib/pipeline/schema.js b/src/lib/pipeline/schema.js
@@ -139,7 +139,8 @@ export function asEnvMap(problems, path, value) {
export const MAX_DURATION = 7 * 24 * 60 * 60;
-// Accepts 90, '90s', '30m', '1h', or a composite such as '1h30m'.
+// Accepts 90, '90s', '30m', '1h', '30d', '2w', or a composite such as
+// '1h30m'. A bare number is seconds.
export function asDuration(problems, path, value, { max = MAX_DURATION } = {}) {
let seconds;
@@ -151,12 +152,15 @@ export function asDuration(problems, path, value, { max = MAX_DURATION } = {}) {
if (/^\d+$/.test(text)) {
seconds = parseInt(text, 10);
} else {
- const matches = [...text.matchAll(/(\d+)([smh])/g)];
+ const matches = [...text.matchAll(/(\d+)([smhdw])/g)];
const consumed = matches.reduce((n, m) => n + m[0].length, 0);
if (matches.length === 0 || consumed !== text.length) {
- return problems.add(path, `expected a duration such as 90s, 30m or 1h30m, got ${JSON.stringify(value)}`);
+ return problems.add(
+ path,
+ `expected a duration such as 90s, 30m, 1h30m or 30d, got ${JSON.stringify(value)}`
+ );
}
- const unit = { s: 1, m: 60, h: 3600 };
+ const unit = { s: 1, m: 60, h: 3600, d: 86400, w: 604800 };
seconds = matches.reduce((n, m) => n + parseInt(m[1], 10) * unit[m[2]], 0);
}
} else {
diff --git a/src/lib/projects.js b/src/lib/projects.js
@@ -0,0 +1,100 @@
+// src/lib/projects.js - project records
+//
+// A project pairs a repository with the settings needed to turn a push into
+// a run. The trigger secret is the only sensitive field, and it has to be
+// recoverable rather than hashed because HMAC verification needs the
+// original value, so it is sealed with the secret box.
+
+import { newProjectId, slugify } from './ids.js';
+
+export const SOURCE_MODES = ['archive', 'clone'];
+
+const COLUMNS = `
+ id, name, repo_url, default_branch, config_path, source_mode,
+ trigger_secret, enabled, run_counter, created_at, updated_at
+`;
+
+export function createProjects({ db, secrets }) {
+ function aad(projectId) {
+ return `project:${projectId}/trigger_secret`;
+ }
+
+ return {
+ async get(id) {
+ return db.get(`SELECT ${COLUMNS} FROM projects WHERE id = {id}`, { id });
+ },
+
+ async list() {
+ return db.all(`SELECT ${COLUMNS} FROM projects ORDER BY name`);
+ },
+
+ async create(input) {
+ const id = input.id ? slugify(input.id) : (slugify(input.name) || newProjectId());
+ const now = Date.now();
+
+ if (input.source_mode && !SOURCE_MODES.includes(input.source_mode)) {
+ throw new Error(`source_mode must be one of ${SOURCE_MODES.join(', ')}`);
+ }
+ if (!input.repo_url) throw new Error('repo_url is required');
+
+ const existing = await this.get(id);
+ if (existing) throw new Error(`project ${id} already exists`);
+
+ await db.run(
+ `INSERT INTO projects
+ (id, name, repo_url, default_branch, config_path, source_mode,
+ trigger_secret, enabled, run_counter, created_at, updated_at)
+ VALUES
+ ({id}, {name}, {repo_url}, {branch}, {config_path}, {source_mode},
+ {secret}, {enabled}, 0, {now}, {now})`,
+ {
+ id,
+ name: input.name || id,
+ repo_url: input.repo_url,
+ branch: input.default_branch || 'main',
+ config_path: input.config_path || '.conductor.yml',
+ source_mode: input.source_mode || 'archive',
+ secret: input.trigger_secret ? secrets.seal(input.trigger_secret, aad(id)) : null,
+ enabled: input.enabled === undefined ? 1 : (input.enabled ? 1 : 0),
+ now,
+ }
+ );
+
+ return this.get(id);
+ },
+
+ // Returns the plaintext trigger secret, or null when the project does
+ // not require signed triggers.
+ triggerSecret(project) {
+ if (!project.trigger_secret) return null;
+ return secrets.open(project.trigger_secret, aad(project.id));
+ },
+
+ async setTriggerSecret(id, value) {
+ await db.run(
+ 'UPDATE projects SET trigger_secret = {secret}, updated_at = {now} WHERE id = {id}',
+ { id, secret: value === null ? null : secrets.seal(value, aad(id)), now: Date.now() }
+ );
+ },
+
+ async setEnabled(id, enabled) {
+ await db.run(
+ 'UPDATE projects SET enabled = {enabled}, updated_at = {now} WHERE id = {id}',
+ { id, enabled: enabled ? 1 : 0, now: Date.now() }
+ );
+ },
+
+ async remove(id) {
+ await db.run('DELETE FROM projects WHERE id = {id}', { id });
+ },
+
+ // Allocates the next run number. Must run inside the same transaction
+ // as the run insert, or two concurrent pushes can collide on the
+ // unique (project_id, number) index.
+ async nextRunNumber(tx, projectId) {
+ await tx.run('UPDATE projects SET run_counter = run_counter + 1 WHERE id = {id}', { id: projectId });
+ const row = await tx.get('SELECT run_counter FROM projects WHERE id = {id}', { id: projectId });
+ return row.run_counter;
+ },
+ };
+}
diff --git a/src/lib/stream.js b/src/lib/stream.js
@@ -0,0 +1,27 @@
+// src/lib/stream.js - stream helpers
+//
+// Note for anyone tempted to hash with a 'data' listener on a PassThrough:
+// attaching one switches the stream into flowing mode immediately, so the
+// bytes are gone before the real consumer attaches and the upload silently
+// stores nothing. A Transform stays paused until something reads it, and
+// keeps backpressure intact.
+
+import crypto from 'node:crypto';
+import { Transform } from 'node:stream';
+
+export function hashingTransform(algorithm = 'sha256') {
+ const hash = crypto.createHash(algorithm);
+ let bytes = 0;
+
+ const transform = new Transform({
+ transform(chunk, encoding, callback) {
+ hash.update(chunk);
+ bytes += chunk.length;
+ callback(null, chunk);
+ },
+ });
+
+ transform.digest = () => hash.digest('hex');
+ transform.bytes = () => bytes;
+ return transform;
+}
diff --git a/src/lib/workers.js b/src/lib/workers.js
@@ -0,0 +1,62 @@
+// src/lib/workers.js - worker registration tokens
+//
+// A worker authenticates with a bearer token. Only the SHA-256 of the token
+// is stored, so the database never holds a usable credential; the plaintext
+// is shown once at creation and cannot be recovered afterwards.
+
+import { newToken, hashToken, newWorkerTokenId } from './ids.js';
+
+export function createWorkerTokens({ db }) {
+ return {
+ async create(name) {
+ if (!name || typeof name !== 'string') throw new Error('worker token needs a name');
+ const token = newToken();
+ const id = newWorkerTokenId();
+ await db.run(
+ `INSERT INTO worker_tokens (id, name, token_hash, enabled, created_at)
+ VALUES ({id}, {name}, {hash}, 1, {now})`,
+ { id, name, hash: hashToken(token), now: Date.now() }
+ );
+ // The only time the plaintext exists outside the worker.
+ return { id, name, token };
+ },
+
+ // Returns the token record, or null. Lookup is by hash, so a timing
+ // difference cannot reveal anything beyond whether a hash exists.
+ async verify(presented, { ip = null } = {}) {
+ if (typeof presented !== 'string' || presented.length === 0) return null;
+ const hash = hashToken(presented);
+ const row = await db.get(
+ 'SELECT id, name, enabled FROM worker_tokens WHERE token_hash = {hash}',
+ { hash }
+ );
+ if (!row || row.enabled !== 1) return null;
+
+ await db.run(
+ 'UPDATE worker_tokens SET last_seen_at = {now}, last_ip = {ip} WHERE id = {id}',
+ { id: row.id, now: Date.now(), ip }
+ );
+ return { id: row.id, name: row.name };
+ },
+
+ async list() {
+ return db.all(
+ `SELECT id, name, enabled, created_at, last_seen_at, last_ip
+ FROM worker_tokens ORDER BY created_at DESC`
+ );
+ },
+
+ async setEnabled(id, enabled) {
+ const res = await db.run(
+ 'UPDATE worker_tokens SET enabled = {enabled} WHERE id = {id}',
+ { id, enabled: enabled ? 1 : 0 }
+ );
+ return res.changes > 0;
+ },
+
+ async remove(id) {
+ const res = await db.run('DELETE FROM worker_tokens WHERE id = {id}', { id });
+ return res.changes > 0;
+ },
+ };
+}
diff --git a/test/conductor.test.js b/test/conductor.test.js
@@ -0,0 +1,652 @@
+// test/conductor.test.js - end to end behaviour of the conductor service
+//
+// Drives a real server against a real git repository: trigger, dispatch,
+// logs, artifacts, completion and failure handling.
+
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { startHarness } from './helpers/harness.js';
+
+// Each test gets its own harness so that ordering never matters.
+async function withHarness(options, fn) {
+ const h = await startHarness(options);
+ try {
+ return await fn(h);
+ } finally {
+ await h.stop();
+ }
+}
+
+function jobsByName(payload) {
+ return Object.fromEntries(payload.jobs.map((j) => [j.name, j]));
+}
+
+async function runState(h, runId) {
+ const res = await h.app.inject({ method: 'GET', url: `/api/runs/${runId}` });
+ return res.json();
+}
+
+// Claims everything currently eligible for a worker with the given
+// capabilities, keyed by job name. Tests assert on which jobs appear rather
+// than on the order they arrive in, which is not part of the contract.
+async function drain(h, query = {}) {
+ const claimed = {};
+ for (let i = 0; i < 50; i += 1) {
+ const res = await h.poll(query);
+ if (res.statusCode === 204) break;
+ const job = res.json().job;
+ claimed[job.name] = job;
+ }
+ return claimed;
+}
+
+async function claimNamed(h, name, query) {
+ return (await drain(h, query))[name] ?? null;
+}
+
+async function finish(h, jobId, { success = true, exitCode = 0, error = null } = {}) {
+ const res = await h.app.inject({
+ method: 'POST',
+ url: `/api/workers/jobs/${encodeURIComponent(jobId)}/done`,
+ headers: { ...h.auth, 'content-type': 'application/json' },
+ payload: JSON.stringify({ success, exit_code: exitCode, error }),
+ });
+ return res.json();
+}
+
+test('health reports the selected drivers', async () => {
+ await withHarness({}, async (h) => {
+ const res = await h.app.inject({ method: 'GET', url: '/health' });
+ assert.equal(res.statusCode, 200);
+ assert.deepEqual(res.json(), {
+ ok: true, service: 'conductor', database: 'sqlite', storage: 'local', auth: 'local',
+ });
+ });
+});
+
+test('a signed trigger creates a run and expands the pipeline', async () => {
+ await withHarness({}, async (h) => {
+ const res = await h.trigger({ sha: h.sha, ref: 'refs/heads/main', actor: 'alice' });
+ assert.equal(res.statusCode, 200);
+
+ const body = res.json();
+ assert.equal(body.status, 'created');
+ // lint + 2 build + 2 package + publish
+ assert.equal(body.jobs, 6);
+
+ const detail = await runState(h, body.run_id);
+ assert.equal(detail.run.state, 'running');
+ assert.equal(detail.run.actor, 'alice');
+ assert.equal(detail.run.title, 'initial commit');
+ assert.equal(detail.run.number, 1);
+
+ const jobs = jobsByName(detail);
+ assert.deepEqual(jobs['package:arch=x86_64,pkg=musl'].needs, [`${body.run_id}:build:arch=x86_64`]);
+ assert.equal(jobs.publish.needs.length, 3);
+ });
+});
+
+test('an unsigned or wrongly signed trigger is refused', async () => {
+ await withHarness({}, async (h) => {
+ const unsigned = await h.app.inject({
+ method: 'POST',
+ url: '/api/trigger/demo',
+ headers: { 'content-type': 'application/json' },
+ payload: JSON.stringify({ sha: h.sha }),
+ });
+ assert.equal(unsigned.statusCode, 401);
+
+ const wrong = await h.trigger({ sha: h.sha }, { secret: 'not-the-secret' });
+ assert.equal(wrong.statusCode, 401);
+ });
+});
+
+test('github and gitlab push payloads are understood', async () => {
+ await withHarness({}, async (h) => {
+ const github = await h.trigger({
+ after: h.sha,
+ before: '0'.repeat(40),
+ ref: 'refs/heads/main',
+ pusher: { name: 'octocat' },
+ });
+ assert.equal(github.statusCode, 200);
+ assert.equal(github.json().status, 'created');
+
+ const gitlab = await h.trigger({
+ checkout_sha: h.sha,
+ ref: 'refs/heads/main',
+ user_username: 'gl-user',
+ });
+ assert.equal(gitlab.statusCode, 200);
+ });
+});
+
+test('a branch deletion is ignored rather than failing', async () => {
+ await withHarness({}, async (h) => {
+ const res = await h.trigger({ sha: '0'.repeat(40), ref: 'refs/heads/gone' });
+ assert.equal(res.statusCode, 200);
+ assert.equal(res.json().status, 'ignored');
+ });
+});
+
+test('a broken pipeline is reported as a client error with detail', async () => {
+ await withHarness({ pipeline: 'version: 1\njobs:\n a:\n script: [x]\n' }, async (h) => {
+ const res = await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ assert.equal(res.statusCode, 422);
+ const body = res.json();
+ assert.equal(body.error, 'invalid pipeline');
+ assert.ok(body.problems.some((p) => p.path === 'jobs.a.image'));
+ });
+});
+
+test('a missing pipeline file is reported clearly', async () => {
+ await withHarness({ pipeline: null }, async (h) => {
+ const res = await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ assert.equal(res.statusCode, 422);
+ assert.match(res.json().detail, /\.conductor\.yml not found/);
+ });
+});
+
+test('the worker api rejects missing and invalid tokens', async () => {
+ await withHarness({}, async (h) => {
+ const none = await h.app.inject({ method: 'GET', url: '/api/workers/poll' });
+ assert.equal(none.statusCode, 401);
+
+ const bad = await h.app.inject({
+ method: 'GET', url: '/api/workers/poll', headers: { authorization: 'Bearer nope' },
+ });
+ assert.equal(bad.statusCode, 401);
+ });
+});
+
+test('independent jobs dispatch concurrently', async () => {
+ await withHarness({}, async (h) => {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+
+ // The prototype could only ever hand out one job at a time.
+ const [a, b, c] = await Promise.all([
+ h.poll({ arches: 'x86_64,aarch64' }),
+ h.poll({ arches: 'x86_64,aarch64' }),
+ h.poll({ arches: 'x86_64,aarch64' }),
+ ]);
+
+ const names = [a, b, c].map((r) => r.json().job.name);
+ assert.equal(new Set(names).size, 3, `expected three distinct jobs, got ${names.join(', ')}`);
+ assert.deepEqual(names.slice().sort(), ['build:arch=aarch64', 'build:arch=x86_64', 'lint']);
+ });
+});
+
+test('a job is only offered to a worker that satisfies its requires', async () => {
+ await withHarness({}, async (h) => {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+
+ // Clear the three root jobs so only package and publish remain.
+ const roots = await drain(h, { arches: 'x86_64,aarch64' });
+ assert.deepEqual(Object.keys(roots).sort(), ['build:arch=aarch64', 'build:arch=x86_64', 'lint']);
+ for (const job of Object.values(roots)) await finish(h, job.id, { success: true });
+
+ // package requires sign-key, which this worker does not advertise.
+ const without = await h.poll({ arches: 'x86_64,aarch64' });
+ assert.equal(without.statusCode, 204);
+
+ const with_ = await h.poll({ arches: 'x86_64,aarch64', features: 'sign-key' });
+ assert.equal(with_.statusCode, 200);
+ assert.match(with_.json().job.name, /^package:/);
+ });
+});
+
+test('a worker is not offered work for an architecture it cannot build', async () => {
+ await withHarness({}, async (h) => {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+
+ const claimed = [];
+ for (let i = 0; i < 3; i += 1) {
+ const res = await h.poll({ arches: 'aarch64' });
+ if (res.statusCode === 204) break;
+ claimed.push(res.json().job);
+ }
+
+ // lint has no arch so it is eligible; the x86_64 build is not.
+ assert.ok(!claimed.some((j) => j.arch === 'x86_64'));
+ assert.ok(claimed.some((j) => j.name === 'build:arch=aarch64'));
+ });
+});
+
+test('a claimed job carries everything the worker needs', async () => {
+ await withHarness({}, async (h) => {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const job = await claimNamed(h, 'build:arch=x86_64', { arches: 'x86_64' });
+
+ assert.equal(job.image, 'debian:bookworm-slim');
+ assert.deepEqual(job.script, ['./build.sh $ARCH']);
+ assert.equal(job.env.ARCH, 'x86_64');
+ assert.equal(job.sha, h.sha);
+ assert.equal(job.source.mode, 'archive');
+ assert.match(job.source.url, /^http:\/\/conductor\.test\/api\/workers\/jobs\//);
+ assert.ok(job.endpoints.log.endsWith('/log'));
+ });
+});
+
+test('the source tarball is served from the mirror', async () => {
+ await withHarness({}, async (h) => {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const job = await claimNamed(h, 'lint', {});
+
+ const res = await h.app.inject({
+ method: 'GET',
+ url: `/api/workers/jobs/${encodeURIComponent(job.id)}/source.tar.gz`,
+ headers: h.auth,
+ });
+ assert.equal(res.statusCode, 200);
+ assert.equal(res.headers['content-type'], 'application/gzip');
+ // gzip magic number
+ assert.equal(res.rawPayload[0], 0x1f);
+ assert.equal(res.rawPayload[1], 0x8b);
+ });
+});
+
+test('a worker cannot touch a job it does not hold', async () => {
+ await withHarness({}, async (h) => {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const job = await claimNamed(h, 'lint', {});
+
+ const other = await h.services.workerTokens.create('intruder');
+ const res = await h.app.inject({
+ method: 'POST',
+ url: `/api/workers/jobs/${encodeURIComponent(job.id)}/log`,
+ headers: { authorization: `Bearer ${other.token}`, 'content-type': 'application/octet-stream' },
+ payload: Buffer.from('malicious'),
+ });
+ assert.equal(res.statusCode, 403);
+ });
+});
+
+test('log appends are resumable and deduplicated by offset', async () => {
+ await withHarness({}, async (h) => {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const job = await claimNamed(h, 'lint', {});
+ const url = `/api/workers/jobs/${encodeURIComponent(job.id)}/log`;
+ const headers = { ...h.auth, 'content-type': 'application/octet-stream' };
+
+ const first = await h.app.inject({
+ method: 'POST', url, headers: { ...headers, 'x-log-offset': '0' }, payload: Buffer.from('hello\n'),
+ });
+ assert.deepEqual(first.json(), { size: 6, written: 6, truncated: false });
+
+ // A retry resends what was already stored plus new data.
+ const retry = await h.app.inject({
+ method: 'POST', url, headers: { ...headers, 'x-log-offset': '0' }, payload: Buffer.from('hello\nworld\n'),
+ });
+ assert.deepEqual(retry.json(), { size: 12, written: 6, truncated: false });
+
+ // A gap is refused, and says where to resume.
+ const gap = await h.app.inject({
+ method: 'POST', url, headers: { ...headers, 'x-log-offset': '9999' }, payload: Buffer.from('x'),
+ });
+ assert.equal(gap.statusCode, 409);
+ assert.equal(gap.json().expected_offset, 12);
+
+ const tail = await h.app.inject({ method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}/log` });
+ assert.equal(tail.body, 'hello\nworld\n');
+ assert.equal(tail.headers['x-log-size'], '12');
+ assert.equal(tail.headers['x-log-complete'], 'false');
+
+ const partial = await h.app.inject({
+ method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}/log?offset=6`,
+ });
+ assert.equal(partial.body, 'world\n');
+ });
+});
+
+test('a finished log moves to storage and is still readable', async () => {
+ await withHarness({}, async (h) => {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const job = await claimNamed(h, 'lint', {});
+
+ await h.app.inject({
+ method: 'POST',
+ url: `/api/workers/jobs/${encodeURIComponent(job.id)}/log`,
+ headers: { ...h.auth, 'content-type': 'application/octet-stream' },
+ payload: Buffer.from('compiling\ndone\n'),
+ });
+ await finish(h, job.id, { success: true });
+
+ const tail = await h.app.inject({ method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}/log` });
+ assert.equal(tail.statusCode, 200);
+ assert.equal(tail.body, 'compiling\ndone\n');
+ assert.equal(tail.headers['x-log-complete'], 'true');
+ });
+});
+
+test('artifacts are stored, hashed, and stripped of traversal', async () => {
+ await withHarness({}, async (h) => {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const job = await claimNamed(h, 'build:arch=x86_64', { arches: 'x86_64' });
+
+ const upload = await h.app.inject({
+ method: 'POST',
+ url: `/api/workers/jobs/${encodeURIComponent(job.id)}/artifact`,
+ headers: {
+ ...h.auth,
+ 'content-type': 'application/octet-stream',
+ 'x-artifact-path': '../../../etc/passwd',
+ },
+ payload: Buffer.from('artifact bytes'),
+ });
+ assert.equal(upload.statusCode, 200);
+
+ const body = upload.json();
+ assert.equal(body.path, 'etc/passwd', 'traversal should be stripped, not honoured');
+ assert.equal(body.size, 14);
+ assert.match(body.sha256, /^[0-9a-f]{64}$/);
+
+ const detail = await h.app.inject({ method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}` });
+ const artifacts = detail.json().artifacts;
+ assert.equal(artifacts.length, 1);
+
+ const download = await h.app.inject({ method: 'GET', url: `/api/artifacts/${artifacts[0].id}` });
+ assert.equal(download.statusCode, 200);
+ assert.equal(download.body, 'artifact bytes');
+ });
+});
+
+test('an artifact shorter than its content-length is rejected', async () => {
+ await withHarness({}, async (h) => {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const job = await claimNamed(h, 'lint', {});
+
+ const res = await h.app.inject({
+ method: 'POST',
+ url: `/api/workers/jobs/${encodeURIComponent(job.id)}/artifact`,
+ headers: {
+ ...h.auth,
+ 'content-type': 'application/octet-stream',
+ 'content-length': '500',
+ 'x-artifact-path': 'short.bin',
+ },
+ payload: Buffer.from('tiny'),
+ });
+ assert.equal(res.statusCode, 400);
+ });
+});
+
+test('success releases dependents, matched on shared dimensions', async () => {
+ await withHarness({}, async (h) => {
+ const run = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().run_id;
+
+ const roots = await drain(h, { arches: 'x86_64,aarch64', features: 'sign-key' });
+ await finish(h, roots['build:arch=x86_64'].id, { success: true });
+
+ // Only the x86_64 package becomes eligible; the aarch64 one still waits
+ // on its own architecture's build.
+ const next = await drain(h, { arches: 'x86_64,aarch64', features: 'sign-key' });
+ assert.ok(next['package:arch=x86_64,pkg=musl'], 'x86_64 package should be released');
+ assert.ok(!next['package:arch=aarch64,pkg=musl'], 'aarch64 package should still be waiting');
+
+ const state = jobsByName(await runState(h, run));
+ assert.equal(state['package:arch=aarch64,pkg=musl'].state, 'queued');
+ });
+});
+
+test('failure skips transitive dependents, not just direct ones', async () => {
+ await withHarness({}, async (h) => {
+ const run = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().run_id;
+
+ const build = await claimNamed(h, 'build:arch=x86_64', { arches: 'x86_64' });
+ const outcome = await finish(h, build.id, { success: false, exitCode: 2, error: 'compile failed' });
+ assert.equal(outcome.state, 'failed');
+
+ const jobs = jobsByName(await runState(h, run));
+ // Direct dependent.
+ assert.equal(jobs['package:arch=x86_64,pkg=musl'].state, 'skipped');
+ // Transitive dependent: the prototype left this queued and dispatchable.
+ assert.equal(jobs.publish.state, 'skipped');
+ // Unrelated branches are untouched.
+ assert.equal(jobs['build:arch=aarch64'].state, 'queued');
+ assert.equal(jobs['package:arch=aarch64,pkg=musl'].state, 'queued');
+ });
+});
+
+test('a run settles as failed once every job is terminal', async () => {
+ await withHarness({}, async (h) => {
+ const run = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().run_id;
+
+ const roots = await drain(h, { arches: 'x86_64,aarch64', features: 'sign-key' });
+
+ // Failing the x86_64 build skips its package and, transitively, publish.
+ await finish(h, roots['build:arch=x86_64'].id, { success: false, exitCode: 1 });
+ await finish(h, roots['build:arch=aarch64'].id, { success: true });
+ await finish(h, roots.lint.id, { success: true });
+
+ const released = await drain(h, { arches: 'aarch64', features: 'sign-key' });
+ const last = await finish(h, released['package:arch=aarch64,pkg=musl'].id, { success: true });
+
+ assert.equal(last.run_state, 'failed');
+ assert.equal((await runState(h, run)).run.state, 'failed');
+ });
+});
+
+test('a run settles as success when everything passes', async () => {
+ // Quoted, because an unquoted true in YAML is a boolean, not a command.
+ const pipeline = `
+version: 1
+jobs:
+ a:
+ image: alpine
+ script: ['true']
+ b:
+ image: alpine
+ script: ['true']
+ needs: [a]
+`;
+ await withHarness({ pipeline }, async (h) => {
+ const run = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().run_id;
+
+ const a = await claimNamed(h, 'a', {});
+ await finish(h, a.id, { success: true });
+ const b = await claimNamed(h, 'b', {});
+ const last = await finish(h, b.id, { success: true });
+
+ assert.equal(last.run_state, 'success');
+ assert.equal((await runState(h, run)).run.state, 'success');
+ });
+});
+
+test('an allowed failure does not fail the run or block dependents', async () => {
+ const pipeline = `
+version: 1
+jobs:
+ flaky:
+ image: alpine
+ script: ['false']
+ allow_failure: true
+ after:
+ image: alpine
+ script: ['true']
+ needs: [flaky]
+`;
+ await withHarness({ pipeline }, async (h) => {
+ const run = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().run_id;
+
+ const flaky = await claimNamed(h, 'flaky', {});
+ const outcome = await finish(h, flaky.id, { success: false, exitCode: 1 });
+ assert.deepEqual(outcome.skipped, []);
+
+ const after = await claimNamed(h, 'after', {});
+ assert.ok(after, 'dependent should still run after an allowed failure');
+ const last = await finish(h, after.id, { success: true });
+ assert.equal(last.run_state, 'success');
+ assert.equal((await runState(h, run)).run.state, 'success');
+ });
+});
+
+test('a job with retries left is requeued instead of failing the run', async () => {
+ const pipeline = `
+version: 1
+jobs:
+ retried:
+ image: alpine
+ script: [maybe]
+ max_attempts: 2
+`;
+ await withHarness({ pipeline }, async (h) => {
+ const run = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().run_id;
+
+ const first = await claimNamed(h, 'retried', {});
+ const outcome = await finish(h, first.id, { success: false, exitCode: 1 });
+ assert.equal(outcome.state, 'queued');
+ assert.equal(outcome.retry, true);
+
+ const second = await claimNamed(h, 'retried', {});
+ assert.equal(second.attempt, 2);
+ const last = await finish(h, second.id, { success: false, exitCode: 1 });
+ assert.equal(last.state, 'failed');
+ assert.equal((await runState(h, run)).run.state, 'failed');
+ });
+});
+
+test('heartbeats keep a job alive and report cancellation', 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', {});
+
+ const beat = await h.app.inject({
+ method: 'POST',
+ url: `/api/workers/jobs/${encodeURIComponent(job.id)}/heartbeat`,
+ headers: { ...h.auth, 'content-type': 'application/json' },
+ payload: '{}',
+ });
+ assert.deepEqual(beat.json(), { cancelled: false });
+
+ await h.services.scheduler.cancelRun(run);
+
+ const after = await h.app.inject({
+ method: 'POST',
+ url: `/api/workers/jobs/${encodeURIComponent(job.id)}/heartbeat`,
+ headers: { ...h.auth, 'content-type': 'application/json' },
+ payload: '{}',
+ });
+ assert.equal(after.json().cancelled, true);
+ assert.equal((await runState(h, run)).run.state, 'cancelled');
+ });
+});
+
+test('the reaper fails a job whose worker stopped reporting', 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(jobsByName(await runState(h, run)).lint.state, 'running');
+
+ // Backdate the heartbeat rather than waiting out the real timeout.
+ await h.services.db.run(
+ 'UPDATE jobs SET heartbeat_at = {then} WHERE id = {id}',
+ { then: Date.now() - 10 * 60 * 1000, id: job.id }
+ );
+
+ const reaped = await h.services.scheduler.reap();
+ assert.ok(reaped.length > 0);
+
+ const after = jobsByName(await runState(h, run)).lint;
+ assert.equal(after.state, 'failed');
+ assert.match(after.error, /stopped reporting/);
+
+ // The worker that lost the job can no longer complete it.
+ const late = await finish(h, job.id, { success: true });
+ assert.match(String(late.error ?? ''), /not running|another worker/);
+ });
+});
+
+test('a second push produces an independent, numbered run', async () => {
+ await withHarness({}, async (h) => {
+ const first = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json();
+ const sha2 = await h.commit({ 'README.md': 'changed\n' }, 'second commit');
+ const second = (await h.trigger({ sha: sha2, ref: 'refs/heads/main' })).json();
+
+ assert.notEqual(first.run_id, second.run_id);
+ const detail = await runState(h, second.run_id);
+ assert.equal(detail.run.number, 2);
+ assert.equal(detail.run.title, 'second commit');
+ assert.equal(detail.run.head_sha, sha2);
+ });
+});
+
+test('runs can be listed and filtered by project', async () => {
+ await withHarness({}, async (h) => {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+
+ const all = await h.app.inject({ method: 'GET', url: '/api/runs' });
+ assert.equal(all.json().runs.length, 1);
+
+ const mine = await h.app.inject({ method: 'GET', url: '/api/runs?project=demo' });
+ assert.equal(mine.json().runs.length, 1);
+
+ const other = await h.app.inject({ method: 'GET', url: '/api/runs?project=absent' });
+ assert.equal(other.json().runs.length, 0);
+ });
+});
+
+test('unknown runs, jobs and artifacts return 404', async () => {
+ await withHarness({}, async (h) => {
+ for (const url of ['/api/runs/nope', '/api/jobs/nope', '/api/jobs/nope/log', '/api/artifacts/nope']) {
+ const res = await h.app.inject({ method: 'GET', url });
+ assert.equal(res.statusCode, 404, `${url} should be 404`);
+ }
+ });
+});
+
+test('a disabled project refuses triggers', async () => {
+ await withHarness({}, async (h) => {
+ await h.services.projects.setEnabled('demo', false);
+ const res = await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ assert.equal(res.statusCode, 409);
+ });
+});
+
+test('a trigger for an unknown project is a 404', async () => {
+ await withHarness({}, async (h) => {
+ const res = await h.app.inject({
+ method: 'POST',
+ url: '/api/trigger/absent',
+ headers: { 'content-type': 'application/json' },
+ payload: JSON.stringify({ sha: h.sha }),
+ });
+ assert.equal(res.statusCode, 404);
+ });
+});
+
+test('a commit that is not in the repository is refused', async () => {
+ await withHarness({}, async (h) => {
+ const res = await h.trigger({ sha: 'b'.repeat(40), ref: 'refs/heads/main' });
+ assert.equal(res.statusCode, 500);
+ assert.match(res.json().detail, /not present in the mirror/);
+ });
+});
+
+test('clone mode hands the repository url to the worker instead', async () => {
+ await withHarness({ sourceMode: 'clone' }, async (h) => {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const job = await claimNamed(h, 'lint', {});
+ assert.equal(job.source.mode, 'clone');
+ assert.equal(job.source.repo_url, h.repoDir);
+ });
+});
+
+test('the trigger secret is not stored in the clear', async () => {
+ await withHarness({}, async (h) => {
+ const row = await h.services.db.get('SELECT trigger_secret FROM projects WHERE id = {id}', { id: 'demo' });
+ assert.ok(row.trigger_secret.startsWith('v1.'), 'expected an encrypted value');
+ assert.ok(!row.trigger_secret.includes('test-secret'));
+ assert.equal(h.services.projects.triggerSecret(await h.services.projects.get('demo')), 'test-secret');
+ });
+});
+
+test('worker tokens are stored only as hashes', async () => {
+ await withHarness({}, async (h) => {
+ const rows = await h.services.db.all('SELECT token_hash FROM worker_tokens');
+ assert.ok(rows.length > 0);
+ for (const row of rows) {
+ assert.match(row.token_hash, /^[0-9a-f]{64}$/);
+ assert.notEqual(row.token_hash, h.worker.token);
+ }
+ });
+});
diff --git a/test/examples.test.js b/test/examples.test.js
@@ -0,0 +1,57 @@
+// test/examples.test.js - the shipped examples must stay valid
+//
+// Documentation that no longer parses is worse than no documentation, so
+// every example is compiled as part of the suite.
+
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import fs from 'node:fs/promises';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { compilePipeline } from '../src/lib/pipeline/index.js';
+
+const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..');
+const EXAMPLES = path.join(ROOT, 'examples');
+
+test('every example pipeline compiles', async () => {
+ const files = (await fs.readdir(EXAMPLES)).filter((f) => f.endsWith('.yml') || f.endsWith('.yaml'));
+ assert.ok(files.length > 0, 'expected at least one example');
+
+ for (const file of files) {
+ const text = await fs.readFile(path.join(EXAMPLES, file), 'utf8');
+ const pipeline = compilePipeline(text, { source: file });
+ assert.ok(pipeline.jobs.length > 0, `${file} produced no jobs`);
+ }
+});
+
+test('the distribution example fans in per architecture', async () => {
+ const text = await fs.readFile(path.join(EXAMPLES, 'distro.conductor.yml'), 'utf8');
+ const pipeline = compilePipeline(text, { source: 'distro.conductor.yml' });
+ const jobs = Object.fromEntries(pipeline.jobs.map((j) => [j.name, j]));
+
+ // check, two toolchains, twelve packages, two images, one publish.
+ assert.equal(pipeline.jobs.length, 18);
+
+ // A package waits only on its own architecture's toolchain.
+ assert.deepEqual(jobs['package:arch=aarch64,pkg=musl'].needs, ['toolchain:arch=aarch64']);
+ assert.deepEqual(jobs['package:arch=x86_64,pkg=grub'].needs, ['toolchain:arch=x86_64']);
+
+ // An image waits on every package for its architecture, and no others.
+ const imageNeeds = jobs['image:arch=x86_64'].needs;
+ assert.equal(imageNeeds.length, 6);
+ assert.ok(imageNeeds.every((n) => n.includes('arch=x86_64')));
+
+ // publish has no architecture, so it waits for both, plus check.
+ assert.equal(jobs.publish.needs.length, 3);
+ assert.equal(jobs.publish.depth, 3);
+});
+
+test('the node example wires services and features', async () => {
+ const text = await fs.readFile(path.join(EXAMPLES, 'node-app.conductor.yml'), 'utf8');
+ const pipeline = compilePipeline(text, { source: 'node-app.conductor.yml' });
+ const jobs = Object.fromEntries(pipeline.jobs.map((j) => [j.name, j]));
+
+ assert.deepEqual(jobs['test:suite=unit'].services.map((s) => s.alias), ['postgres']);
+ assert.deepEqual(jobs.image.requires, ['docker']);
+ assert.equal(jobs['test:suite=integration'].artifacts.when, 'always');
+});
diff --git a/test/helpers/harness.js b/test/helpers/harness.js
@@ -0,0 +1,164 @@
+// test/helpers/harness.js - a conductor on a temporary database
+//
+// Builds a real server against a real git repository and a real sqlite file,
+// then drives it through fastify inject. No ports are bound and no
+// background processes are started, so the tests stay deterministic.
+
+import fs from 'node:fs/promises';
+import os from 'node:os';
+import path from 'node:path';
+import crypto from 'node:crypto';
+import { execFile } from 'node:child_process';
+import { promisify } from 'node:util';
+import { loadConfig } from '../../src/lib/config.js';
+import { createServices, buildServer } from '../../src/conductor/app.js';
+
+const execFileAsync = promisify(execFile);
+
+export const DEFAULT_PIPELINE = `
+version: 1
+defaults:
+ image: debian:bookworm-slim
+jobs:
+ lint:
+ script: [make lint]
+ build:
+ arch: [x86_64, aarch64]
+ script: ['./build.sh $ARCH']
+ artifacts:
+ paths: [dist/**]
+ package:
+ needs: [build]
+ arch: [x86_64, aarch64]
+ matrix:
+ pkg: [musl]
+ requires: [sign-key]
+ script: ['./pkg.sh $MATRIX_PKG $ARCH']
+ publish:
+ needs: [package, lint]
+ script: [./publish.sh]
+`;
+
+export async function createRepo(dir, { pipeline = DEFAULT_PIPELINE, configPath = '.conductor.yml' } = {}) {
+ await fs.mkdir(dir, { recursive: true });
+ const git = (...args) => execFileAsync('git', ['-C', dir, ...args]);
+
+ await execFileAsync('git', ['init', '-q', '-b', 'main', dir]);
+ await git('config', 'user.email', 'test@example.invalid');
+ await git('config', 'user.name', 'Test');
+
+ await fs.mkdir(path.dirname(path.join(dir, configPath)), { recursive: true });
+ if (pipeline !== null) await fs.writeFile(path.join(dir, configPath), pipeline);
+ await fs.writeFile(path.join(dir, 'README.md'), 'test repository\n');
+
+ await git('add', '-A');
+ await git('commit', '-q', '-m', 'initial commit');
+ const { stdout } = await git('rev-parse', 'HEAD');
+ return stdout.trim();
+}
+
+export async function startHarness(options = {}) {
+ const root = await fs.mkdtemp(path.join(os.tmpdir(), 'conductor-it-'));
+ const repoDir = path.join(root, 'repo');
+ const sha = await createRepo(repoDir, options);
+
+ const configFile = path.join(root, 'conductor.yaml');
+ await fs.writeFile(configFile, [
+ 'server:',
+ ' port: 0',
+ ' public_url: http://conductor.test',
+ 'database:',
+ ' path: ./state/conductor.db',
+ 'storage:',
+ ' path: ./state/storage',
+ 'git:',
+ ' mirror_path: ./state/mirrors',
+ ' fetch_interval: 0',
+ 'log:',
+ ' spool_path: ./state/logs',
+ 'secrets:',
+ ` encryption_key: ${crypto.randomBytes(32).toString('hex')}`,
+ 'scheduler:',
+ ' heartbeat_timeout: 120',
+ ' reap_interval: 30',
+ '',
+ ].join('\n'));
+
+ const savedConfig = process.env.CONDUCTOR_CONFIG;
+ process.env.CONDUCTOR_CONFIG = configFile;
+ const cfg = loadConfig();
+ if (savedConfig === undefined) delete process.env.CONDUCTOR_CONFIG;
+ else process.env.CONDUCTOR_CONFIG = savedConfig;
+
+ const services = await createServices(cfg, { migrationLogger: () => {}, logger: silentLogger() });
+ const app = await buildServer(services, { logger: false });
+
+ const project = await services.projects.create({
+ id: 'demo',
+ name: 'Demo',
+ repo_url: repoDir,
+ trigger_secret: 'test-secret',
+ config_path: options.configPath ?? '.conductor.yml',
+ source_mode: options.sourceMode ?? 'archive',
+ });
+
+ const worker = await services.workerTokens.create('test-worker');
+
+ return {
+ root,
+ repoDir,
+ sha,
+ cfg,
+ app,
+ services,
+ project,
+ worker,
+ auth: { authorization: `Bearer ${worker.token}` },
+
+ // Commits a change and returns the new commit id.
+ async commit(files, message = 'update') {
+ for (const [name, content] of Object.entries(files)) {
+ await fs.mkdir(path.dirname(path.join(repoDir, name)), { recursive: true });
+ await fs.writeFile(path.join(repoDir, name), content);
+ }
+ await execFileAsync('git', ['-C', repoDir, 'add', '-A']);
+ await execFileAsync('git', ['-C', repoDir, 'commit', '-q', '-m', message]);
+ const { stdout } = await execFileAsync('git', ['-C', repoDir, 'rev-parse', 'HEAD']);
+ return stdout.trim();
+ },
+
+ sign(body) {
+ return `sha256=${crypto.createHmac('sha256', 'test-secret').update(body).digest('hex')}`;
+ },
+
+ async trigger(payload, { secret = 'test-secret' } = {}) {
+ const body = JSON.stringify(payload);
+ const signature = `sha256=${crypto.createHmac('sha256', secret).update(body).digest('hex')}`;
+ return app.inject({
+ method: 'POST',
+ url: '/api/trigger/demo',
+ headers: { 'content-type': 'application/json', 'x-hub-signature-256': signature },
+ payload: body,
+ });
+ },
+
+ async poll(query = {}) {
+ const search = new URLSearchParams(query).toString();
+ return app.inject({
+ method: 'GET',
+ url: `/api/workers/poll${search ? `?${search}` : ''}`,
+ headers: this.auth,
+ });
+ },
+
+ async stop() {
+ await app.close();
+ await services.db.close();
+ await fs.rm(root, { recursive: true, force: true });
+ },
+ };
+}
+
+function silentLogger() {
+ return { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} };
+}
diff --git a/test/pipeline.test.js b/test/pipeline.test.js
@@ -91,6 +91,28 @@ jobs:
assert.equal(jobs.c.timeout, 5400);
});
+test('durations accept day and week units', () => {
+ const p = compilePipeline(`
+version: 1
+jobs:
+ a:
+ image: x
+ script: [y]
+ artifacts:
+ paths: [d]
+ expire: 30d
+ b:
+ image: x
+ script: [y]
+ artifacts:
+ paths: [d]
+ expire: 2w
+`);
+ const jobs = byName(p);
+ assert.equal(jobs.a.artifacts.expire, 30 * 86400);
+ assert.equal(jobs.b.artifacts.expire, 14 * 86400);
+});
+
test('a malformed duration is rejected', () => {
assert.deepEqual(
errorPaths(() => compilePipeline('version: 1\njobs:\n a: { image: x, script: [y], timeout: soon }\n')),
