commit d628031cfedf667d7af02bccfde0141906ef4bd0
parent cd31d529e6f55d00f5736b5fe3c96294a23d9544
Author: finwo <finwo@pm.me>
Date: Fri, 25 Sep 2026 17:16:01 +0200
Let a script hand values to later tasks and name its own outcome
Diffstat:
29 files changed, 1455 insertions(+), 119 deletions(-)
diff --git a/README.md b/README.md
@@ -176,11 +176,12 @@ tasks:
needs: [package, test]
script:
- ./publish.sh
+ - echo "::set-output url=https://dl.example.com/$VERSION::public"
report:
after: [publish]
script:
- - ./report.sh
+ - ./report.sh ${{ tasks.publish.outputs.url }}
```
`arch` and `matrix` fan a task out into one task per combination. A dependency
@@ -192,6 +193,14 @@ architecture of its own, waits for all of them.
`report` runs whether the run went well or badly, which is usually when a
report is worth having.
+A script hands a value to a later task by writing `::set-output` on its own
+output, and reads one back with `${{ tasks.<task>.outputs.<name> }}`, which
+is itself a reason to wait for that task. Outputs are masked unless marked
+`::public`, so a token passed between tasks stays out of both logs. A script
+can also write `::set-state success|failed|skipped` to decide its own
+outcome regardless of what it exited with; a skip is not a failure, and what
+came after it still runs.
+
`requires` names features a worker must offer. A worker holding a signing key
advertises `sign-key`, and only tasks asking for it are sent there. The key
stays on the worker and is never known to the conductor.
diff --git a/assets/style.css b/assets/style.css
@@ -113,10 +113,10 @@ tbody tr:hover { background: #1b1f23; }
tbody.stage:first-of-type .stage-label th { padding-top: .45rem; }
tbody.stage tr.stage-label:hover { background: none; }
-/* Durations and exit codes are read down the column, not across the row,
- so they are right aligned and set in figures of one width. */
+/* Durations are read down the column, not across the row, so they are
+ right aligned and set in figures of one width. */
td.num { text-align: right; font-variant-numeric: tabular-nums; }
-table.tasks thead th:nth-last-child(-n+2) { text-align: right; }
+table.tasks thead th:last-child { text-align: right; }
.logpane { max-height: 32rem; overflow: auto; border: 1px solid var(--border); border-radius: 6px; background: #0b0d0f; }
pre.log {
diff --git a/docs/api.md b/docs/api.md
@@ -198,7 +198,12 @@ empty body.
"allow_failure": false, "exit_code": 0, "error": null,
"worker_name": "builder-1", "timeout": 3600, "workdir": "/work",
"needs": ["build:arch=x86_64"], "after": [], "matrix": { "pkg": "musl" },
- "artifact_paths": ["dist/**"], "log_size": 4096,
+ "artifact_paths": ["dist/**"],
+ "outputs": [
+ { "name": "tag", "masked": false, "value": "v1.2.3" },
+ { "name": "token", "masked": true, "value": null }
+ ],
+ "log_size": 4096,
"created_at": 1800000000000, "started_at": null, "finished_at": null,
"project_id": "demo", "job_id": "m2y5rj7ykdm9sfbr54hf", "job_number": 12,
"ref": "refs/heads/main", "sha": "<40 hex>", "visibility": "public"
@@ -218,6 +223,9 @@ told the task to collect; `artifacts` is what it actually stored. Each carries a
convenience rather than the permission: that route checks visibility again
for itself.
+`outputs` is what the task wrote with `::set-output`. A masked one reports
+its name so the shape of a run is visible, and never its value.
+
There is no `env` and no `services` here, deliberately. Both can hold
project variables, and a service carries an environment of its own. The
`script` is left out for the same reason: a public job can be built from a
@@ -305,13 +313,26 @@ that does not match its declared length is rejected.
Requests are not enveloped; only responses are:
```json
-{ "success": true, "exit_code": 0, "error": null }
+{
+ "success": true, "state": "success", "exit_code": 0, "error": null,
+ "outputs": [{ "name": "tag", "value": "v1.2.3", "masked": false }]
+}
```
-A failure with attempts remaining requeues the task; otherwise everything
-that `needs` it, directly or transitively, is skipped. Tasks that only name
-it under `after` are released instead, since they asked to follow it rather
-than to be fed by it.
+`state` is one of `success`, `failed` or `skipped`, and is what the task
+wrote with `::set-state` if it wrote one. It takes precedence over
+`success`, which is kept for workers that do not send a state. Names and
+sizes in `outputs` are validated again here rather than trusted from the
+worker, and values are encrypted before they are stored.
+
+A failure with attempts remaining requeues the task, discarding any outputs
+that attempt produced; otherwise everything that `needs` it, directly or
+transitively, is skipped. Tasks that only name it under `after` are released
+instead, since they asked to follow it rather than to be fed by it.
+
+A task that reports `skipped` is not a failure: the job can still end
+`success`, and tasks that `needs` it still run. One of them reading an output
+it never produced fails on that, before it is handed to a worker.
Artifacts
---------
diff --git a/docs/pipeline.md b/docs/pipeline.md
@@ -258,6 +258,96 @@ there is nothing left to wait for, so the edge is dropped rather than the
dependent. This is the opposite of `needs`, which takes its dependents out
of the run with it.
+Talking back from a script
+--------------------------
+
+A script can write directives on its own output to hand a value to a later
+task, or to say how it should be counted. Both take the same shape, a line
+starting at column 0:
+
+```
+::<command> <arguments>[::<data>]
+```
+
+The line is removed from the log before it is shipped, so a value written
+this way does not appear in it.
+
+### Outputs
+
+```yaml
+build:
+ script:
+ - 'make'
+ - 'echo "::set-output tag=$(git describe)::public"'
+
+deploy:
+ script: ['./deploy.sh ${{ tasks.build.outputs.tag }}']
+```
+
+`deploy` was never declared to need `build`. Reading its output says so, and
+the dependency is recorded exactly as if `needs: [build]` had been written.
+
+Outputs are masked unless stated otherwise, matching project variables:
+
+```
+::set-output token=hunter2 # masked
+::set-output token=hunter2::masked # the same, said out loud
+::set-output tag=v1.2.3::public # shown in the API and the UI
+```
+
+A masked value is redacted from the rest of that task's log, and from the log
+of every task that goes on to read it. It is never returned by the API and
+shows as `[masked]` in the UI. Both kinds are encrypted at rest, exactly as
+project variables are; `public` decides what may be shown again, not how it
+is stored.
+
+Values are stored per task, and a value is only reachable through a reference
+to the task that produced it, so two tasks may use the same output name
+without colliding. A reference may be read anywhere `${{ }}` works except
+`requires`, which chooses the worker before any task has run.
+
+Naming a matrix task waits for every instance of it and looks the output up
+across all of them, which is what a step collecting the whole matrix wants:
+
+```yaml
+package:
+ arch: [x86_64, aarch64]
+ script: ['./build.sh', 'echo "::set-output path-${{ arch }}=/out/pkg.tar::public"']
+
+index:
+ script: ['./reindex.sh ${{ tasks.package.outputs.path-x86_64 }}']
+```
+
+If the referencing task shares a dimension with the task it reads, the
+reference narrows to its own instance, the same way `needs` does.
+
+A reference to a value that was never produced fails the task before it is
+handed to a worker, naming what is missing. It does not run with a hole where
+the value should have been.
+
+### Choosing the outcome
+
+```yaml
+package:
+ script:
+ - './build.sh'
+ - '[ -f out/pkg.tar ] || echo "::set-state skipped"'
+```
+
+`::set-state` takes `success`, `failed` or `skipped`, and overrides what the
+exit code would have meant. It does not stop the script: the run continues,
+and the last directive written is the one that counts.
+
+A timeout, a cancellation, or a failure to start the container still fails
+the task. Those are not outcomes a script gets to talk its way out of.
+
+**A skip is not a failure.** A skipped task leaves the job green, and
+everything that needed it still runs: the indexer at the end of a package
+repository does its work whether or not this particular package was rebuilt.
+What it does not do is invent values: a task reading an output that a skipped
+task never produced fails on that missing output, which is the difference
+between not needing the value and needing one that is not there.
+
Worker features
---------------
@@ -368,9 +458,14 @@ Injected automatically:
Interpolation
-------------
-`${{ ... }}` is substituted before the task is stored, in `image`, `script`,
-`env` values, `requires` and `services`. Two expressions are available:
-`arch` and `matrix.<name>`.
+`${{ ... }}` is substituted in `image`, `script`, `env` values, `requires`
+and `services`. Three expressions are available:
+
+| Expression | Substituted |
+| --------------------------------- | ---------------------- |
+| `arch` | when the task is stored |
+| `matrix.<name>` | when the task is stored |
+| `tasks.<task>.outputs.<name>` | when the task is dispatched |
```yaml
build:
@@ -386,6 +481,12 @@ Referring to a dimension the task does not declare is an error. Inside
shell expands them; interpolation exists for the fields the shell never sees,
such as `image`.
+The third is different in kind: its value does not exist until another task
+has run, so it survives compilation and is filled in when a worker claims the
+task. That is why it cannot appear in `requires`, which is read to decide
+which worker to offer the task to. See [talking back from a
+script](#talking-back-from-a-script).
+
Full example
------------
diff --git a/migrations/mysql/004_task_outputs.sql b/migrations/mysql/004_task_outputs.sql
@@ -0,0 +1,15 @@
+-- 004_task_outputs.sql - values a task produced while it ran
+--
+-- Written by ::set-output, read by ${{ tasks.<name>.outputs.<key> }} when a
+-- later task is dispatched. Sealed like project variables whether masked or
+-- not; masked only decides whether the value may be shown again.
+
+CREATE TABLE task_outputs (
+ task_id VARCHAR(64) NOT NULL,
+ name VARCHAR(191) NOT NULL,
+ value TEXT NOT NULL,
+ masked TINYINT NOT NULL DEFAULT 1,
+ created_at BIGINT NOT NULL,
+ PRIMARY KEY (task_id, name),
+ CONSTRAINT fk_task_outputs_task FOREIGN KEY (task_id) REFERENCES tasks(id) ON DELETE CASCADE
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
diff --git a/migrations/postgres/004_task_outputs.sql b/migrations/postgres/004_task_outputs.sql
@@ -0,0 +1,14 @@
+-- 004_task_outputs.sql - values a task produced while it ran
+--
+-- Written by ::set-output, read by ${{ tasks.<name>.outputs.<key> }} when a
+-- later task is dispatched. Sealed like project variables whether masked or
+-- not; masked only decides whether the value may be shown again.
+
+CREATE TABLE task_outputs (
+ task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
+ name TEXT NOT NULL,
+ value TEXT NOT NULL,
+ masked SMALLINT NOT NULL DEFAULT 1,
+ created_at BIGINT NOT NULL,
+ PRIMARY KEY (task_id, name)
+);
diff --git a/migrations/sqlite/004_task_outputs.sql b/migrations/sqlite/004_task_outputs.sql
@@ -0,0 +1,14 @@
+-- 004_task_outputs.sql - values a task produced while it ran
+--
+-- Written by ::set-output, read by ${{ tasks.<name>.outputs.<key> }} when a
+-- later task is dispatched. Sealed like project variables whether masked or
+-- not; masked only decides whether the value may be shown again.
+
+CREATE TABLE task_outputs (
+ task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
+ name TEXT NOT NULL,
+ value TEXT NOT NULL,
+ masked INTEGER NOT NULL DEFAULT 1,
+ created_at INTEGER NOT NULL,
+ PRIMARY KEY (task_id, name)
+);
diff --git a/src/conductor/app.js b/src/conductor/app.js
@@ -45,7 +45,7 @@ export async function createServices(cfg, options = {}) {
const users = createUsers({ db, logger });
const auth = createAuth({ cfg, users, logger });
- const scheduler = createScheduler({ cfg, db, git, logs, storage, projects, variables, logger });
+ const scheduler = createScheduler({ cfg, db, git, logs, storage, projects, secrets, variables, logger });
const retention = createRetention({ cfg, db, storage, logs, logger });
// Without a first administrator there is no way into the admin surface.
diff --git a/src/conductor/routes/tasks.js b/src/conductor/routes/tasks.js
@@ -33,9 +33,10 @@ import { canManageProject } from '../../lib/projects.js';
import { verifySignature } from '../../lib/signature.js';
import { ok } from './envelope.js';
import { UnknownTask } from '../../lib/errors.js';
+import { outputAad } from '../../lib/outputs.js';
import { limitOf, offsetOfQuery, pageBlock, paginateOffset } from '../../lib/page.js';
-export default async function taskRoutes(fastify, { cfg, db, projects, auth = null }) {
+export default async function taskRoutes(fastify, { cfg, db, projects, secrets, auth = null }) {
const base = cfg.server.public_url.replace(/\/+$/, '');
// Explicit columns rather than t.*, which would pull spec and
@@ -60,11 +61,33 @@ export default async function taskRoutes(fastify, { cfg, db, projects, auth = nu
}
}
+ // Names always, values only where the task said they could be shown.
+ async function readOutputs(taskId) {
+ const rows = await db.all(
+ 'SELECT task_id, name, value, masked FROM task_outputs WHERE task_id = {task} ORDER BY name',
+ { task: taskId }
+ );
+ return rows.map((row) => {
+ const masked = row.masked === 1;
+ let value = null;
+ if (!masked) {
+ try {
+ value = secrets.open(row.value, outputAad(row.task_id, row.name));
+ } catch {
+ value = null;
+ }
+ }
+ return { name: row.name, masked, value };
+ });
+ }
+
// Everything a caller is allowed to see, assembled once so the two routes
// cannot drift into answering differently.
async function represent(req, reply, row) {
const spec = readSpec(row.spec);
+ const outputs = await readOutputs(row.id);
+
const artifacts = await paginateOffset(db, {
select: 'id, path, size, sha256, created_at, expires_at',
from: 'artifacts',
@@ -99,6 +122,7 @@ export default async function taskRoutes(fastify, { cfg, db, projects, auth = nu
// What the task was told to collect, which is not the same as what
// it managed to store. The list below is what exists.
artifact_paths: spec.artifactPaths,
+ outputs,
log_size: row.log_size,
created_at: row.created_at,
started_at: row.started_at,
diff --git a/src/conductor/routes/workers.js b/src/conductor/routes/workers.js
@@ -21,6 +21,7 @@ import { LogOffsetError } from '../../lib/log.js';
import { hashingTransform } from '../../lib/stream.js';
import { sanitizeRelativePath, keys as storageKeys } from '../../lib/storage/index.js';
import { maskBuffer } from '../../lib/variables.js';
+import { readOutputList } from '../../lib/outputs.js';
import { newArtifactId } from '../../lib/ids.js';
import { ok } from './envelope.js';
import {
@@ -39,6 +40,8 @@ const COMPLETE_FAILURES = {
'not-running': TaskNotRunning,
};
+const TASK_OUTCOMES = ['success', 'failed', 'skipped'];
+
export default async function workerRoutes(fastify, { cfg, db, git, logs, storage, projects, scheduler, workerTokens, variables }) {
// Raw bodies: log chunks and artifacts arrive as octet streams and must
// not be parsed.
@@ -275,13 +278,18 @@ export default async function workerRoutes(fastify, { cfg, db, git, logs, storag
const body = req.body ?? {};
const success = body.success === true;
+ // A worker too old to send a state still says whether it succeeded.
+ const state = TASK_OUTCOMES.includes(body.state)
+ ? body.state
+ : (success ? 'success' : 'failed');
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(task.id, {
- success,
+ state,
exitCode,
error,
+ outputs: readOutputList(body.outputs),
tokenId: req.worker.id,
});
// All 409: by this point heldTask has already answered for a task that
diff --git a/src/conductor/scheduler.js b/src/conductor/scheduler.js
@@ -19,8 +19,12 @@
// Claiming is a conditional UPDATE guarded on the previous state, so two
// workers polling at the same instant cannot both win the same task.
-import { compilePipeline, PipelineError, transitiveDependents, DEFAULT_WORKDIR } from '../lib/pipeline/index.js';
+import {
+ compilePipeline, PipelineError, transitiveDependents, DEFAULT_WORKDIR,
+ outputRefsIn, fillOutputs,
+} from '../lib/pipeline/index.js';
import { inClause } from '../lib/db/query.js';
+import { outputAad } from '../lib/outputs.js';
import { newJobId, newTaskId } from '../lib/ids.js';
import { keys as storageKeys } from '../lib/storage/index.js';
@@ -33,12 +37,11 @@ const TERMINAL = ['success', 'failed', 'skipped', 'cancelled'];
// The two spellings a repository may use without being configured for it.
export const CONVENTIONAL_CONFIG_PATHS = ['.conductor.yml', '.conductor.yaml'];
-export function createScheduler({ cfg, db, git, logs, storage, projects, variables = null, logger = console }) {
- // A task is eligible when it is queued, its job is live, and no dependency
- // is outstanding. What outstanding means depends on the edge. A needs
- // dependency has to have succeeded, and a failed one marked allow_failure
- // counts as satisfied, which is the whole point of the flag. An after
- // dependency only has to have stopped running.
+export function createScheduler({
+ cfg, db, git, logs, storage, projects, secrets, variables = null, logger = console,
+}) {
+ // needs is satisfied by success, skipped, or failed under allow_failure.
+ // after only waits for the dependency to stop running.
const ELIGIBLE = `
SELECT t.id, t.job_id, t.name, t.arch, t.image, t.requires, t.spec,
t.timeout, t.attempt, t.max_attempts,
@@ -54,7 +57,7 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
JOIN tasks dt ON dt.id = d.depends_on_id
WHERE d.task_id = t.id
AND d.kind = 'needs'
- AND dt.state <> 'success'
+ AND dt.state NOT IN ('success', 'skipped')
AND NOT (dt.state = 'failed' AND dt.allow_failure = 1)
)
AND NOT EXISTS (
@@ -127,7 +130,7 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
const bad = await tx.get(
`SELECT COUNT(*) AS c FROM tasks
WHERE job_id = {job}
- AND (state IN ('skipped', 'cancelled')
+ AND (state = 'cancelled'
OR (state = 'failed' AND allow_failure = 0))`,
{ job: jobId }
);
@@ -215,6 +218,7 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
matrix: task.matrix,
needs: task.needs,
after: task.after,
+ output_refs: task.output_refs ?? {},
depth: task.depth,
workdir,
}),
@@ -324,6 +328,14 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
}
if (!requires.every((f) => featureSet.has(f))) continue;
+ // Before claiming: a task whose inputs are not there never runs, so
+ // there is no point handing it to a worker first.
+ const resolution = await resolveOutputs(candidate);
+ if (resolution.error) {
+ await failUnclaimable(candidate, resolution.error);
+ continue;
+ }
+
const now = Date.now();
const claimed = await db.run(
`UPDATE tasks
@@ -338,6 +350,7 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
const spec = JSON.parse(candidate.spec);
const attempt = candidate.attempt + 1;
+ const fill = (value) => fillOutputs(value, resolution.values);
// Project variables are resolved at dispatch and handed to the
// worker, never written into the tasks table, so the only place a
@@ -360,16 +373,17 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
project_id: candidate.project_id,
name: candidate.name,
arch: candidate.arch,
- image: candidate.image,
+ image: fill(candidate.image),
sha: candidate.head_sha,
ref: candidate.ref,
requires,
timeout: candidate.timeout,
attempt,
max_attempts: candidate.max_attempts,
- script: spec.script,
- // Values the worker must redact from the log stream.
- masked: injected.masked,
+ script: fill(spec.script),
+ // Values the worker must redact from the log stream. A masked
+ // output stays masked in every task that goes on to read it.
+ masked: [...injected.masked, ...resolution.masked],
// Project variables, then pipeline env, then the facts about this
// task. Ordered so a pipeline cannot quietly redefine the last
// group, and a variable cannot shadow an explicit task setting.
@@ -379,7 +393,7 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
// script, and the worker only copies them into the container.
env: {
...injected.env,
- ...spec.env,
+ ...fill(spec.env),
CONDUCTOR_PROJECT: candidate.project_id,
CONDUCTOR_JOB_ID: candidate.job_id,
CONDUCTOR_JOB_NUMBER: String(candidate.number),
@@ -389,8 +403,8 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
CONDUCTOR_REF: candidate.ref ?? '',
CONDUCTOR_ATTEMPT: String(attempt),
},
- services: spec.services,
- artifacts: spec.artifacts,
+ services: fill(spec.services),
+ artifacts: fill(spec.artifacts),
workdir: spec.workdir ?? DEFAULT_WORKDIR,
};
}
@@ -417,7 +431,11 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
// Records the outcome, retries if attempts remain, and otherwise skips
// everything downstream.
- async complete(taskId, { success, exitCode = null, error = null, tokenId = null }) {
+ async complete(taskId, {
+ state: reported = null, success, exitCode = null, error = null,
+ outputs = [], tokenId = null,
+ }) {
+ const outcome = reported ?? (success ? 'success' : 'failed');
return db.transaction(async (tx) => {
const task = await tx.get(
`SELECT id, job_id, name, state, attempt, max_attempts, allow_failure, worker_token_id
@@ -436,13 +454,14 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
const now = Date.now();
- if (success) {
+ if (outcome === 'success' || outcome === 'skipped') {
await tx.run(
- "UPDATE tasks SET state = 'success', exit_code = {code}, finished_at = {now} WHERE id = {id}",
- { id: taskId, code: exitCode, now }
+ 'UPDATE tasks SET state = {state}, exit_code = {code}, finished_at = {now} WHERE id = {id}',
+ { id: taskId, state: outcome, code: exitCode, now }
);
+ await writeOutputs(tx, taskId, outputs);
const jobState = await settleJob(tx, task.job_id);
- return { ok: true, state: 'success', jobState };
+ return { ok: true, state: outcome, jobState };
}
// Transient failure with attempts left: back to the queue rather
@@ -456,6 +475,7 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
WHERE id = {id}`,
{ id: taskId, code: exitCode, error }
);
+ await tx.run('DELETE FROM task_outputs WHERE task_id = {id}', { id: taskId });
return { ok: true, state: 'queued', retry: true, attempt: task.attempt };
}
@@ -463,6 +483,9 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
"UPDATE tasks SET state = 'failed', exit_code = {code}, error = {error}, finished_at = {now} WHERE id = {id}",
{ id: taskId, code: exitCode, error, now }
);
+ // Kept even here: an allow_failure dependency still runs its
+ // dependents, and they may read what it produced.
+ await writeOutputs(tx, taskId, outputs);
let skipped = [];
if (!task.allow_failure) {
@@ -539,6 +562,102 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
TERMINAL,
};
+ // Looks up every ${{ tasks.x.outputs.y }} a task is about to run with.
+ // Returns an error instead of values when one of them was never produced,
+ // which is a real failure: the script would otherwise run with a hole in
+ // its arguments.
+ async function resolveOutputs(candidate) {
+ const spec = JSON.parse(candidate.spec);
+ const refs = spec.output_refs ?? {};
+ const wanted = [
+ ...outputRefsIn(candidate.spec),
+ ...outputRefsIn(candidate.image ?? ''),
+ ];
+ if (wanted.length === 0) return { values: new Map(), masked: [] };
+
+ const names = [...new Set(Object.values(refs).flat())];
+ const list = inClause('t', names);
+ const rows = list.empty ? [] : await db.all(
+ `SELECT o.task_id, o.name, o.value, o.masked, t.name AS task
+ FROM task_outputs o
+ JOIN tasks t ON t.id = o.task_id
+ WHERE t.job_id = {job} AND t.name IN (${list.sql})`,
+ { job: candidate.job_id, ...list.params }
+ );
+
+ const byTask = new Map();
+ for (const row of rows) {
+ if (!byTask.has(row.task)) byTask.set(row.task, new Map());
+ byTask.get(row.task).set(row.name, row);
+ }
+
+ const values = new Map();
+ const masked = [];
+ for (const { whole, ref, key } of wanted) {
+ if (values.has(whole)) continue;
+ const from = (refs[ref] ?? []).filter((name) => byTask.get(name)?.has(key));
+
+ if (from.length === 0) {
+ const where = (refs[ref] ?? []).join(', ') || ref;
+ return { error: `${where} produced no output ${JSON.stringify(key)}` };
+ }
+ if (from.length > 1) {
+ return {
+ error: `output ${JSON.stringify(key)} was produced by ${from.join(' and ')}; ` +
+ 'name one of them',
+ };
+ }
+
+ const row = byTask.get(from[0]).get(key);
+ let value;
+ try {
+ value = secrets.open(row.value, outputAad(row.task_id, key));
+ } catch {
+ return { error: `output ${JSON.stringify(key)} of ${from[0]} could not be read` };
+ }
+ values.set(whole, value);
+ if (row.masked === 1) masked.push(value);
+ }
+
+ return { values, masked };
+ }
+
+ // Never claimed, so it never ran: failed straight from queued.
+ async function failUnclaimable(candidate, message) {
+ await db.transaction(async (tx) => {
+ const now = Date.now();
+ const marked = await tx.run(
+ `UPDATE tasks SET state = 'failed', error = {error}, finished_at = {now}
+ WHERE id = {id} AND state = 'queued'`,
+ { id: candidate.id, error: message, now }
+ );
+ if (marked.changes !== 1) return;
+ await skipDependents(tx, candidate.job_id, [candidate.id], now);
+ await settleJob(tx, candidate.job_id);
+ });
+ logger.warn?.(`task ${candidate.name} cannot run: ${message}`);
+ }
+
+ // Sealed whether masked or not; masked only decides what may be shown.
+ async function writeOutputs(tx, taskId, outputs) {
+ if (!Array.isArray(outputs) || outputs.length === 0) return;
+ await tx.run('DELETE FROM task_outputs WHERE task_id = {id}', { id: taskId });
+ const now = Date.now();
+ for (const { name, value, masked } of outputs) {
+ await tx.run(
+ `INSERT INTO task_outputs (task_id, name, value, masked, created_at)
+ VALUES ({task}, {name}, {value}, {masked}, {now})`,
+ {
+ task: taskId,
+ name,
+ value: secrets.seal(value, outputAad(taskId, name)),
+ masked: masked === false ? 0 : 1,
+ now,
+ }
+ );
+ }
+ }
+
async function skipDependents(tx, jobId, fromTaskIds, now) {
const graph = await loadGraph(tx, jobId);
// Needs edges only. A task waiting on this one with `after` asked to run
diff --git a/src/conductor/ui/pages.js b/src/conductor/ui/pages.js
@@ -100,17 +100,16 @@ export function tasksTable(job, tasks) {
return html`<div id="tasks" ${live ? poll(`/partials/jobs/${job.id}/tasks`, 3) : ''}>
<table class="tasks">
<thead><tr>
- <th>task</th><th>state</th><th>arch</th><th>worker</th><th>duration</th><th>exit</th>
+ <th>task</th><th>state</th><th>arch</th><th>worker</th><th>duration</th>
</tr></thead>
${[...stages.keys()].sort((a, b) => a - b).map((depth) => html`<tbody class="stage">
- <tr class="stage-label"><th colspan="6">stage ${depth + 1}</th></tr>
+ <tr class="stage-label"><th colspan="5">stage ${depth + 1}</th></tr>
${stages.get(depth).sort((a, b) => a.name.localeCompare(b.name)).map((task) => html`<tr>
<td><a href="/tasks/${task.id}">${task.name}</a></td>
<td>${badge(task.state)}</td>
<td class="muted">${task.arch ?? ''}</td>
<td class="muted">${task.worker_name ?? ''}</td>
<td class="muted num">${duration(task.started_at, task.finished_at)}</td>
- <td class="muted num">${task.exit_code === null ? '' : `exit ${task.exit_code}`}</td>
</tr>`)}
</tbody>`)}
</table>
@@ -119,7 +118,7 @@ export function tasksTable(job, tasks) {
// --- tasks ---
-export function taskPage({ task, artifacts, log, query = {} }) {
+export function taskPage({ task, artifacts, outputs = [], log, query = {} }) {
return html`
<div class="panel">
<h2>${task.name}</h2>
@@ -138,6 +137,15 @@ export function taskPage({ task, artifacts, log, query = {} }) {
<h3>log</h3>
<div class="logpane">${taskLog(task, log)}</div>
+ ${outputs.length === 0 ? '' : html`
+ <h3>outputs</h3>
+ <table id="outputs"><tbody>
+ ${outputs.map((o) => html`<tr>
+ <td class="mono">${o.name}</td>
+ <td class="muted">${o.masked ? '[masked]' : o.value}</td>
+ </tr>`)}
+ </tbody></table>`}
+
${artifacts.rows.length === 0 ? '' : html`
<h3>artifacts</h3>
<table><tbody>
@@ -519,11 +527,11 @@ function workerTasksTable(token, tasks, query) {
return html`<div id="worker-tasks">
<table class="tasks">
<thead><tr>
- <th>task</th><th>state</th><th>machine</th><th>arch</th><th>claimed</th><th>duration</th><th>exit</th>
+ <th>task</th><th>state</th><th>machine</th><th>arch</th><th>claimed</th><th>duration</th>
</tr></thead>
<tbody>
${tasks.rows.length === 0
- ? html`<tr><td colspan="7" class="muted">This worker has not run anything yet.</td></tr>`
+ ? html`<tr><td colspan="6" class="muted">This worker has not run anything yet.</td></tr>`
: tasks.rows.map((t) => html`<tr>
<td><a href="/tasks/${t.id}">${t.name}</a></td>
<td>${badge(t.state)}</td>
@@ -531,7 +539,6 @@ function workerTasksTable(token, tasks, query) {
<td class="muted">${t.arch ?? ''}</td>
<td class="muted">${t.claimed_at ? ago(t.claimed_at) : ''}</td>
<td class="muted num">${duration(t.started_at, t.finished_at)}</td>
- <td class="muted num">${t.exit_code ?? ''}</td>
</tr>`)}
</tbody>
</table>
diff --git a/src/conductor/ui/routes.js b/src/conductor/ui/routes.js
@@ -22,11 +22,12 @@ import { paginate, pageOf, windowOfAll } from '../../lib/page.js';
import { issueToken, sessionCookie, clearedCookie } from '../../lib/auth/index.js';
import { createState, verifyState } from '../../lib/auth/state.js';
import { PipelineError, DEFAULT_WORKDIR } from '../../lib/pipeline/index.js';
+import { outputAad } from '../../lib/outputs.js';
const LOG_TAIL_BYTES = 256 * 1024;
export default async function uiRoutes(fastify, services) {
- const { cfg, db, auth, users, projects, workerTokens, variables, logs, scheduler } = services;
+ const { cfg, db, auth, users, projects, workerTokens, variables, secrets, logs, scheduler } = services;
const secure = cfg.server.public_url.startsWith('https://');
const oidcRedirectUri = `${cfg.server.public_url.replace(/\/+$/, '')}/oidc/callback`;
@@ -187,9 +188,26 @@ export default async function uiRoutes(fastify, services) {
return { user, job, project, canManage: canManageProject(user, project) };
}
+ // Masked values are never unsealed here: the page shows that one exists.
+ async function outputsOf(taskId) {
+ const rows = await db.all(
+ 'SELECT task_id, name, value, masked FROM task_outputs WHERE task_id = {task} ORDER BY name',
+ { task: taskId }
+ );
+ return rows.map((row) => {
+ const masked = row.masked === 1;
+ if (masked) return { name: row.name, masked, value: null };
+ try {
+ return { name: row.name, masked, value: secrets.open(row.value, outputAad(row.task_id, row.name)) };
+ } catch {
+ return { name: row.name, masked, value: null };
+ }
+ });
+ }
+
async function tasksOf(jobId) {
const tasks = await db.all(
- `SELECT id, name, arch, state, worker_name, exit_code, started_at, finished_at
+ `SELECT id, name, arch, state, worker_name, started_at, finished_at
FROM tasks WHERE job_id = {job} ORDER BY name`,
{ job: jobId }
);
@@ -285,7 +303,13 @@ export default async function uiRoutes(fastify, services) {
title: task.name,
user,
active: 'jobs',
- body: taskPage({ task, artifacts, log: await tailOf(task), query: req.query }),
+ body: taskPage({
+ task,
+ artifacts,
+ outputs: await outputsOf(task.id),
+ log: await tailOf(task),
+ query: req.query,
+ }),
});
});
@@ -637,7 +661,7 @@ export default async function uiRoutes(fastify, services) {
}
const tasks = await paginate(db, {
- select: 'id, name, state, arch, worker_name, exit_code, claimed_at, started_at, finished_at',
+ select: 'id, name, state, arch, worker_name, claimed_at, started_at, finished_at',
from: 'tasks',
where: 'worker_token_id = {worker}',
order: 'claimed_at DESC',
diff --git a/src/lib/outputs.js b/src/lib/outputs.js
@@ -0,0 +1,26 @@
+// src/lib/outputs.js - values a task produced, as the conductor sees them
+//
+// The worker parses these out of a log stream and has its own copy of the
+// rules; it shares no code with the conductor. These are the conductor's,
+// applied to what a worker reports rather than trusted from it.
+
+export const OUTPUT_NAME = /^[A-Za-z_][A-Za-z0-9_-]*(?:\.[A-Za-z_][A-Za-z0-9_-]*)*$/;
+export const MAX_OUTPUT_VALUE = 8192;
+export const MAX_OUTPUTS = 64;
+
+// Binds a sealed value to where it is stored, so a row copied onto another
+// task or name fails to open.
+export const outputAad = (taskId, name) => `task:${taskId}/output:${name}`;
+
+export function readOutputList(value) {
+ if (!Array.isArray(value)) return [];
+ const out = [];
+ for (const entry of value.slice(0, MAX_OUTPUTS)) {
+ if (!entry || typeof entry !== 'object') continue;
+ const { name, value: v } = entry;
+ if (typeof name !== 'string' || !OUTPUT_NAME.test(name)) continue;
+ if (typeof v !== 'string' || v.length > MAX_OUTPUT_VALUE) continue;
+ out.push({ name, value: v, masked: entry.masked !== false });
+ }
+ return out;
+}
diff --git a/src/lib/pipeline/dag.js b/src/lib/pipeline/dag.js
@@ -15,6 +15,9 @@ export const ALL_EDGES = ['needs', 'after'];
// Terminal states, which is what an `after` edge waits for.
const FINISHED = ['success', 'failed', 'skipped', 'cancelled'];
+// What lets a `needs` dependent through. A skip is not a failure.
+const SATISFIED = ['success', 'skipped'];
+
function depsOf(task, edges) {
if (edges.length === 1) return task[edges[0]] ?? [];
return edges.flatMap((kind) => task[kind] ?? []);
@@ -100,7 +103,7 @@ export function transitiveDependents(tasks, startNames, { edges = ['needs'] } =
// Tasks whose dependencies have all reached a satisfying state. Passed the
// current state of every task by name.
-export function runnable(tasks, stateByName, { satisfied = ['success'], finished = FINISHED } = {}) {
+export function runnable(tasks, stateByName, { satisfied = SATISFIED, finished = FINISHED } = {}) {
return tasks.filter((task) => {
if (stateByName.get(task.name) !== 'queued') return false;
if (!(task.needs ?? []).every((dep) => satisfied.includes(stateByName.get(dep)))) return false;
diff --git a/src/lib/pipeline/expand.js b/src/lib/pipeline/expand.js
@@ -16,6 +16,9 @@ import { Problems, MATRIX_KEY_PATTERN } from './schema.js';
// Matches ${{ arch }} and ${{ matrix.pkg }}, tolerating inner whitespace.
const INTERPOLATION = /\$\{\{\s*([^}]*?)\s*\}\}/g;
+// Both halves may contain dots, so the split is on the last `.outputs.`.
+export const OUTPUT_REF = /^tasks\.(.+)\.outputs((?:\.[A-Za-z_][A-Za-z0-9_-]*)+)$/;
+
// tasks.name is 191 characters in the mysql schema.
export const MAX_TASK_NAME = 191;
@@ -46,11 +49,54 @@ export function interpolate(problems, path, text, context) {
return context.matrix[key];
}
- problems.add(path, `unknown expression \${{ ${expr} }}; expected arch or matrix.<name>`);
+ // Left in place: the value only exists once the other task has run, so
+ // this is resolved again when the task is dispatched.
+ const outputMatch = OUTPUT_REF.exec(expr);
+ if (outputMatch) {
+ if (!context.outputs) {
+ problems.add(
+ path,
+ `cannot read \${{ ${expr} }} here; a task output is not known until the ` +
+ 'task is dispatched, and this is needed before that'
+ );
+ return whole;
+ }
+ context.outputs.add(outputMatch[1]);
+ return whole;
+ }
+
+ problems.add(
+ path,
+ `unknown expression \${{ ${expr} }}; expected arch, matrix.<name> or ` +
+ 'tasks.<task>.outputs.<name>'
+ );
return whole;
});
}
+// Every unresolved task-output reference in a string.
+export function outputRefsIn(text) {
+ const found = [];
+ for (const [whole, expr] of String(text).matchAll(INTERPOLATION)) {
+ const match = OUTPUT_REF.exec(expr.trim());
+ if (match) found.push({ whole, ref: match[1], key: match[2].slice(1) });
+ }
+ return found;
+}
+
+// Replaces them once their values are known, keyed by the whole expression.
+export function fillOutputs(value, values) {
+ if (typeof value === 'string') {
+ if (values.size === 0) return value;
+ return value.replace(INTERPOLATION, (whole) => (values.has(whole) ? values.get(whole) : whole));
+ }
+ if (Array.isArray(value)) return value.map((v) => fillOutputs(v, values));
+ if (value && typeof value === 'object') {
+ return Object.fromEntries(Object.entries(value).map(([k, v]) => [k, fillOutputs(v, values)]));
+ }
+ return value;
+}
+
function cartesian(dimensions) {
let rows = [{}];
for (const [key, values] of dimensions) {
@@ -103,7 +149,10 @@ export function expandPipeline(pipeline, options = {}) {
continue;
}
- const context = { arch, matrix };
+ const outputs = new Set();
+ const context = { arch, matrix, outputs };
+ // requires picks the worker, which happens before any output exists.
+ const early = { arch, matrix, outputs: null };
const env = {};
for (const [key, value] of Object.entries(task.env)) {
env[key] = interpolate(problems, `${path}.env.${key}`, value, context);
@@ -123,7 +172,7 @@ export function expandPipeline(pipeline, options = {}) {
dims: combo,
image: interpolate(problems, `${path}.image`, task.image, context),
script: task.script.map((line, i) => interpolate(problems, `${path}.script[${i}]`, line, context)),
- requires: task.requires.map((r, i) => interpolate(problems, `${path}.requires[${i}]`, r, context)),
+ requires: task.requires.map((r, i) => interpolate(problems, `${path}.requires[${i}]`, r, early)),
services: task.services.map((service, i) => ({
image: interpolate(problems, `${path}.services[${i}].image`, service.image, context),
alias: service.alias,
@@ -148,6 +197,8 @@ export function expandPipeline(pipeline, options = {}) {
only: task.only ?? null,
needs: [],
after: [],
+ outputRefs: outputs,
+ output_refs: {},
});
}
@@ -163,9 +214,14 @@ export function expandPipeline(pipeline, options = {}) {
instance.needs = resolveDeps(problems, { byBase, byName, baseName, entry, instance, field: 'needs' });
instance.after = resolveDeps(problems, { byBase, byName, baseName, entry, instance, field: 'after' });
+ // Reading a task's output is a reason to wait for it.
+ instance.output_refs = resolveOutputRefs(problems, { byBase, byName, baseName, instance });
+ delete instance.outputRefs;
+ const needs = new Set([...instance.needs, ...Object.values(instance.output_refs).flat()]);
+ instance.needs = [...needs].sort();
+
// Naming the same dependency both ways is held to the stricter
// reading, since needs already implies after.
- const needs = new Set(instance.needs);
instance.after = instance.after.filter((name) => !needs.has(name));
}
}
@@ -182,50 +238,71 @@ function resolveDeps(problems, { byBase, byName, baseName, entry, instance, fiel
const path = `tasks.${baseName}.${field}`;
for (const need of entry.task[field]) {
- // An exact concrete task name wins, which is the escape hatch for
- // depending on one specific combination.
- if (byName.has(need.task)) {
- if (need.task === instance.name) {
- problems.add(path, `task ${JSON.stringify(baseName)} cannot depend on itself`);
- } else {
- resolved.add(need.task);
- }
- continue;
- }
+ const names = matchTask(problems, path, need.task, need.match, { byBase, byName, baseName, instance });
+ for (const name of names ?? []) resolved.add(name);
+ }
- const target = byBase.get(need.task);
- if (!target) {
- const known = [...byBase.keys()].filter((n) => n !== baseName);
- problems.add(
- path,
- `refers to unknown task ${JSON.stringify(need.task)}` +
- (known.length > 0 ? `; defined tasks are ${known.join(', ')}` : '')
- );
- continue;
- }
- if (need.task === baseName) {
+ return [...resolved].sort();
+}
+
+// The concrete instances one written reference means, or null when it could
+// not be resolved and the reason has been recorded.
+function matchTask(problems, path, ref, match, { byBase, byName, baseName, instance }) {
+ // An exact concrete task name wins, which is the escape hatch for
+ // depending on one specific combination.
+ if (byName.has(ref)) {
+ if (ref === instance.name) {
problems.add(path, `task ${JSON.stringify(baseName)} cannot depend on itself`);
- continue;
+ return null;
}
+ return [ref];
+ }
+
+ const target = byBase.get(ref);
+ if (!target) {
+ const known = [...byBase.keys()].filter((n) => n !== baseName);
+ problems.add(
+ path,
+ `refers to unknown task ${JSON.stringify(ref)}` +
+ (known.length > 0 ? `; defined tasks are ${known.join(', ')}` : '')
+ );
+ return null;
+ }
+ if (ref === baseName) {
+ problems.add(path, `task ${JSON.stringify(baseName)} cannot depend on itself`);
+ return null;
+ }
- let candidates = target.instances;
- if (need.match === 'shared') {
- const shared = target.dimensionNames.filter((d) => Object.hasOwn(instance.dims, d));
- if (shared.length > 0) {
- candidates = candidates.filter((c) => shared.every((d) => c.dims[d] === instance.dims[d]));
- if (candidates.length === 0) {
- problems.add(
- path,
- `no instance of ${JSON.stringify(need.task)} matches ${instance.name} on ` +
- `${shared.join(', ')}; add the missing value, or use { task: ${need.task}, match: all }`
- );
- continue;
- }
+ let candidates = target.instances;
+ if (match === 'shared') {
+ const shared = target.dimensionNames.filter((d) => Object.hasOwn(instance.dims, d));
+ if (shared.length > 0) {
+ candidates = candidates.filter((c) => shared.every((d) => c.dims[d] === instance.dims[d]));
+ if (candidates.length === 0) {
+ problems.add(
+ path,
+ `no instance of ${JSON.stringify(ref)} matches ${instance.name} on ` +
+ `${shared.join(', ')}; add the missing value, or use { task: ${ref}, match: all }`
+ );
+ return null;
}
}
+ }
- for (const candidate of candidates) resolved.add(candidate.name);
+ return candidates.map((c) => c.name);
+}
+
+// A reference may name a matrix task, in which case every instance of it is
+// waited for and searched: each one names its outputs differently, so the
+// name picks the instance out at dispatch.
+function resolveOutputRefs(problems, { byBase, byName, baseName, instance }) {
+ const path = `tasks.${baseName}`;
+ const out = {};
+
+ for (const ref of instance.outputRefs) {
+ const names = matchTask(problems, path, ref, 'shared', { byBase, byName, baseName, instance });
+ if (names !== null) out[ref] = [...names].sort();
}
- return [...resolved].sort();
+ return out;
}
diff --git a/src/lib/pipeline/index.js b/src/lib/pipeline/index.js
@@ -12,7 +12,7 @@ import { selectForRef } from './only.js';
import { PipelineError } from './schema.js';
export { parsePipeline } from './parse.js';
-export { expandPipeline, interpolate } from './expand.js';
+export { expandPipeline, interpolate, outputRefsIn, fillOutputs } from './expand.js';
export { topologicalOrder, depths, transitiveDependents, runnable, CycleError } from './dag.js';
export { refMatches, taskRunsOnRef, selectForRef } from './only.js';
export { PipelineError } from './schema.js';
diff --git a/src/lib/pipeline/only.js b/src/lib/pipeline/only.js
@@ -2,10 +2,8 @@
//
// A task carrying an `only` rule runs when the ref that triggered the run
// matches one of its patterns, and is left out of the run entirely when it
-// does not. Left out rather than recorded as skipped, because the
-// scheduler treats a skipped task as a reason to fail the run: that is the
-// right reading when a dependency collapsed, and the wrong one for a
-// publish step that was never meant to run on this branch.
+// does not. Left out rather than recorded as skipped, so a task that was
+// never meant to run on this branch does not show up in the graph at all.
//
// Anything that needs an excluded task is excluded with it. The
// alternative, quietly dropping the dependency, would let a task run
diff --git a/src/worker/client.js b/src/worker/client.js
@@ -112,11 +112,17 @@ export function createClient(cfg) {
return await data(res);
},
- async done(url, { success, exitCode, error }) {
+ async done(url, { success, state = null, exitCode, error, outputs = [] }) {
const res = await fetch(url, {
method: 'POST',
headers: { ...auth, 'content-type': 'application/json' },
- body: JSON.stringify({ success, exit_code: exitCode ?? null, error: error ?? null }),
+ body: JSON.stringify({
+ success,
+ state,
+ exit_code: exitCode ?? null,
+ error: error ?? null,
+ outputs,
+ }),
});
if (!res.ok) throw await readError(res, 'done');
return await data(res);
diff --git a/src/worker/directives.js b/src/worker/directives.js
@@ -0,0 +1,145 @@
+// src/worker/directives.js - magic lines a script writes to talk back
+//
+// ::set-output NAME=VALUE[::masked|public]
+// ::set-state success|failed|skipped
+//
+// Output arrives as arbitrary chunks and stdout interleaves with stderr, so
+// scanning is per stream. A partial line is held back only while it could
+// still become a directive, otherwise ordinary output would stall waiting
+// for a newline that may never come.
+
+export const STATES = ['success', 'failed', 'skipped'];
+export const VISIBILITIES = ['masked', 'public'];
+
+const COMMANDS = ['set-output', 'set-state'];
+const NAME_PATTERN = /^[A-Za-z_][A-Za-z0-9_-]*(?:\.[A-Za-z_][A-Za-z0-9_-]*)*$/;
+const MAX_VALUE = 8192;
+const MAX_OUTPUTS = 64;
+
+// null when the line is not a directive. A recognised command always yields
+// a directive or an error, so a malformed set-output is still stripped
+// rather than printing the value it failed to parse.
+function parse(line) {
+ if (!line.startsWith('::')) return null;
+
+ const body = line.slice(2);
+ const space = body.indexOf(' ');
+ const command = space === -1 ? body : body.slice(0, space);
+ if (!COMMANDS.includes(command)) return null;
+ const rest = space === -1 ? '' : body.slice(space + 1);
+
+ // Only the last :: separates, so a value may contain one.
+ const cut = rest.lastIndexOf('::');
+ const args = cut === -1 ? rest : rest.slice(0, cut);
+ const data = cut === -1 ? null : rest.slice(cut + 2);
+
+ if (command === 'set-state') {
+ if (data !== null) return { error: 'set-state takes no :: value' };
+ if (!STATES.includes(args)) {
+ return { error: `unknown state ${JSON.stringify(args)}; expected ${STATES.join(', ')}` };
+ }
+ return { state: args };
+ }
+
+ const eq = args.indexOf('=');
+ if (eq === -1) return { error: 'set-output wants name=value' };
+ const name = args.slice(0, eq);
+ const value = args.slice(eq + 1);
+ if (!NAME_PATTERN.test(name)) {
+ return { error: `output name ${JSON.stringify(name)} is not a valid name` };
+ }
+ if (value.length > MAX_VALUE) {
+ return { error: `output ${name} is longer than ${MAX_VALUE} characters` };
+ }
+ if (data !== null && !VISIBILITIES.includes(data)) {
+ return { error: `unknown visibility ${JSON.stringify(data)}; expected ${VISIBILITIES.join(', ')}` };
+ }
+ return { output: { name, value, masked: data !== 'public' } };
+}
+
+// Last one wins, for a repeated output name and for the state alike.
+export function createDirectives({ onMask = null, onProblem = null } = {}) {
+ const outputs = new Map();
+ let state = null;
+
+ return {
+ get state() { return state; },
+ get outputs() { return [...outputs.values()]; },
+
+ accept(parsed) {
+ if (parsed.error) return onProblem?.(parsed.error);
+ if (parsed.state) { state = parsed.state; return; }
+
+ const { name, value, masked } = parsed.output;
+ if (!outputs.has(name) && outputs.size >= MAX_OUTPUTS) {
+ return onProblem?.(`more than ${MAX_OUTPUTS} outputs; ${name} was dropped`);
+ }
+ outputs.set(name, { name, value, masked });
+ if (masked) onMask?.(value, name);
+ },
+ };
+}
+
+export function createScanner(directives) {
+ let pending = Buffer.alloc(0);
+ let passing = false;
+
+ // A directive has to begin at column 0, so only a prefix of '::' waits.
+ function mayBeDirective(buf) {
+ if (buf.length >= 2) return buf[0] === 0x3a && buf[1] === 0x3a;
+ if (buf.length === 1) return buf[0] === 0x3a;
+ return true;
+ }
+
+ function keep(line) {
+ const parsed = parse(line.toString('utf8').replace(/[\r\s]+$/, ''));
+ if (parsed === null) return true;
+ directives.accept(parsed);
+ return false;
+ }
+
+ function consume(chunk, final) {
+ const out = [];
+ let buf = pending.length > 0 ? Buffer.concat([pending, chunk]) : chunk;
+ pending = Buffer.alloc(0);
+
+ while (buf.length > 0) {
+ const nl = buf.indexOf(0x0a);
+
+ if (passing) {
+ if (nl === -1) { out.push(buf); break; }
+ out.push(buf.subarray(0, nl + 1));
+ buf = buf.subarray(nl + 1);
+ passing = false;
+ continue;
+ }
+
+ if (nl === -1) {
+ if (final) {
+ if (keep(buf)) out.push(buf);
+ } else if (mayBeDirective(buf)) {
+ pending = buf;
+ } else {
+ out.push(buf);
+ passing = true;
+ }
+ break;
+ }
+
+ if (keep(buf.subarray(0, nl))) out.push(buf.subarray(0, nl + 1));
+ buf = buf.subarray(nl + 1);
+ }
+
+ return out.length === 0 ? Buffer.alloc(0) : Buffer.concat(out);
+ }
+
+ return {
+ push(chunk) { return consume(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk), false); },
+ end() {
+ if (pending.length === 0) return Buffer.alloc(0);
+ const rest = pending;
+ pending = Buffer.alloc(0);
+ return consume(rest, true);
+ },
+ };
+}
diff --git a/src/worker/logstream.js b/src/worker/logstream.js
@@ -35,12 +35,12 @@ const MIN_SECRET_LENGTH = 4;
export function createLogStream(client, url, { masked = [], flushMs = FLUSH_MS, logger = console } = {}) {
// Longest first, so an overlapping shorter secret cannot partially
// reveal a longer one.
- const secrets = [...new Set(masked.filter((s) => typeof s === 'string' && s.length >= MIN_SECRET_LENGTH))]
+ let secrets = [...new Set(masked.filter((s) => typeof s === 'string' && s.length >= MIN_SECRET_LENGTH))]
.sort((a, b) => b.length - a.length);
- const masking = secrets.length > 0;
- const holdBack = masking ? Math.max(...secrets.map((s) => s.length)) - 1 : 0;
+ let masking = secrets.length > 0;
+ let holdBack = masking ? Math.max(...secrets.map((s) => s.length)) - 1 : 0;
- const decoder = masking ? new StringDecoder('utf8') : null;
+ let decoder = masking ? new StringDecoder('utf8') : null;
let text = '';
let bytes = Buffer.alloc(0);
let outbox = Buffer.alloc(0);
@@ -145,6 +145,24 @@ export function createLogStream(client, url, { masked = [], flushMs = FLUSH_MS,
this.write(`${line}\n`);
},
+ // Forward looking: what has already shipped cannot be recalled. False
+ // means too short to redact usefully.
+ addSecret(value) {
+ if (typeof value !== 'string' || value.length < MIN_SECRET_LENGTH) return false;
+ if (secrets.includes(value)) return true;
+
+ secrets = [...secrets, value].sort((a, b) => b.length - a.length);
+ holdBack = Math.max(...secrets.map((s) => s.length)) - 1;
+
+ if (!masking) {
+ masking = true;
+ decoder = new StringDecoder('utf8');
+ text = decoder.write(bytes);
+ bytes = Buffer.alloc(0);
+ }
+ return true;
+ },
+
async close() {
if (ended) return;
await flush(true);
diff --git a/src/worker/task.js b/src/worker/task.js
@@ -18,6 +18,7 @@ import fs from 'node:fs/promises';
import os from 'node:os';
import path from 'node:path';
import { containerName } from './docker.js';
+import { createDirectives, createScanner } from './directives.js';
import { createLogStream } from './logstream.js';
import { sourceIntoContainer } from './source.js';
@@ -89,6 +90,15 @@ export async function runTask({ cfg, client, runtime, task, logger = console })
const log = createLogStream(client, task.endpoints.log, { masked: task.masked ?? [], logger });
+ const directives = createDirectives({
+ onMask: (value, name) => {
+ if (!log.addSecret(value)) {
+ log.note(`[conductor] output ${name} is too short to mask reliably`);
+ }
+ },
+ onProblem: (message) => log.note(`[conductor] ${message}`),
+ });
+
const services = [];
let networkCreated = false;
let containerCreated = false;
@@ -98,6 +108,7 @@ export async function runTask({ cfg, client, runtime, task, logger = console })
let timedOut = false;
let exitCode = null;
let failure = null;
+ let flushScanners = null;
try {
scratch = await fs.mkdtemp(path.join(os.tmpdir(), 'conductor-task-'));
@@ -156,8 +167,13 @@ export async function runTask({ cfg, client, runtime, task, logger = console })
const { child } = runtime.startCreated(taskContainer);
- child.stdout.on('data', (chunk) => log.write(chunk));
- child.stderr.on('data', (chunk) => log.write(chunk));
+ // One per stream; the two interleave.
+ const scanners = [createScanner(directives), createScanner(directives)];
+ child.stdout.on('data', (chunk) => log.write(scanners[0].push(chunk)));
+ child.stderr.on('data', (chunk) => log.write(scanners[1].push(chunk)));
+ flushScanners = () => {
+ for (const scanner of scanners) log.write(scanner.end());
+ };
// Report in, and find out whether the run was cancelled underneath us.
const beatEvery = Math.max(5, Number(task.heartbeat_interval) || 30) * 1000;
@@ -195,7 +211,15 @@ export async function runTask({ cfg, client, runtime, task, logger = console })
if (timeoutTimer) clearTimeout(timeoutTimer);
}
- const success = failure === null && exitCode === 0 && !cancelled && !timedOut;
+ if (flushScanners) flushScanners();
+
+ // A directive only overrides what the exit code implied. Being killed is
+ // not something a script gets to talk its way out of.
+ const hard = failure !== null || cancelled || timedOut;
+ const state = hard
+ ? 'failed'
+ : (directives.state ?? (exitCode === 0 ? 'success' : 'failed'));
+ const success = state === 'success';
// Artifacts are collected before teardown, and even after a failure
// when the pipeline asked for that. They come out of the stopped
@@ -222,11 +246,14 @@ export async function runTask({ cfg, client, runtime, task, logger = console })
? 'task timed out'
: cancelled
? 'task cancelled'
- : exitCode === 0
+ : state !== 'failed'
? null
- : `task exited ${exitCode}`;
+ : exitCode === 0
+ ? 'task set its state to failed'
+ : `task exited ${exitCode}`;
if (error) log.note(`[conductor] ${error}`);
+ else if (state === 'skipped') log.note('[conductor] skipped');
else log.note('[conductor] done');
await log.close();
@@ -235,12 +262,14 @@ export async function runTask({ cfg, client, runtime, task, logger = console })
await cleanup();
try {
- await client.done(task.endpoints.done, { success, exitCode, error });
+ await client.done(task.endpoints.done, {
+ success, state, exitCode, error, outputs: directives.outputs,
+ });
} catch (e) {
logger.error?.(`could not report completion of ${task.name}: ${e.message}`);
}
- return { success, exitCode, error, cancelled, timedOut };
+ return { success, state, exitCode, error, cancelled, timedOut, outputs: directives.outputs };
// Copies what the artifact patterns could possibly match out of the
// stopped container, into a directory the globbing can then walk.
diff --git a/test/conductor.test.js b/test/conductor.test.js
@@ -67,12 +67,14 @@ async function claimNamed(h, name, query) {
return (await drain(h, query))[name] ?? null;
}
-async function finishRaw(h, taskId, { success = true, exitCode = 0, error = null } = {}) {
+async function finishRaw(h, taskId, {
+ success = true, state = null, exitCode = 0, error = null, outputs = [],
+} = {}) {
return h.app.inject({
method: 'POST',
url: `/api/v1/tasks/${taskId}/complete`,
headers: { ...h.auth, 'content-type': 'application/json' },
- payload: JSON.stringify({ success, exit_code: exitCode, error }),
+ payload: JSON.stringify({ success, state, exit_code: exitCode, error, outputs }),
});
}
@@ -932,7 +934,15 @@ test('a job settles as failed once every task is terminal', async () => {
const last = await finish(h, released['package:arch=aarch64,pkg=musl'].id, { success: true });
assert.equal(last.job_state, 'failed');
- assert.equal((await jobState(h, job)).job.state, 'failed');
+ const detail = await jobState(h, job);
+ assert.equal(detail.job.state, 'failed');
+
+ // The failure fails the job, not the skips it caused.
+ assert.ok(
+ detail.tasks.some((t) => t.state === 'skipped'),
+ 'the cascade should have skipped something'
+ );
+ assert.ok(detail.tasks.some((t) => t.state === 'failed' && t.allow_failure === 0));
});
});
@@ -1353,6 +1363,274 @@ tasks:
});
});
+// --- task outputs and set-state ---
+
+test('an output reaches the task that reads it, and implies the dependency', async () => {
+ const pipeline = `
+version: 1
+defaults:
+ image: alpine
+tasks:
+ build:
+ script: ['true']
+ deploy:
+ script: ['ship \${{ tasks.build.outputs.tag }}']
+ env:
+ TAG: '\${{ tasks.build.outputs.tag }}'
+`;
+ await withHarness({ pipeline }, async (h) => {
+ const job = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().data.job_id;
+ await h.waitForJob(job);
+
+ // deploy was never declared to need build; reading its output says so.
+ const tasks = tasksByName(await jobState(h, job));
+ assert.deepEqual(tasks.deploy.needs, [tasks.build.id]);
+
+ const build = await claimNamed(h, 'build', {});
+ assert.ok(build, 'build runs first');
+ assert.equal(await claimNamed(h, 'deploy', {}), null, 'deploy waits for it');
+
+ await finish(h, build.id, {
+ success: true,
+ outputs: [{ name: 'tag', value: 'v1.2.3', masked: false }],
+ });
+
+ const deploy = await claimNamed(h, 'deploy', {});
+ assert.deepEqual(deploy.script, ['ship v1.2.3']);
+ assert.equal(deploy.env.TAG, 'v1.2.3');
+ });
+});
+
+test('a masked output stays masked in the log of whatever reads it', async () => {
+ const pipeline = `
+version: 1
+defaults:
+ image: alpine
+tasks:
+ build:
+ script: ['true']
+ deploy:
+ script: ['ship \${{ tasks.build.outputs.token }}']
+`;
+ await withHarness({ pipeline }, async (h) => {
+ const job = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().data.job_id;
+ await h.waitForJob(job);
+
+ const build = await claimNamed(h, 'build', {});
+ await finish(h, build.id, {
+ success: true,
+ outputs: [{ name: 'token', value: 'sup3rsecret', masked: true }],
+ });
+
+ const deploy = await claimNamed(h, 'deploy', {});
+ assert.deepEqual(deploy.script, ['ship sup3rsecret'], 'the value is still delivered');
+ assert.ok(
+ deploy.masked.includes('sup3rsecret'),
+ 'and the worker is told to redact it from this task too'
+ );
+ });
+});
+
+test('outputs rest sealed rather than readable in the table', async () => {
+ const pipeline = `
+version: 1
+tasks:
+ build:
+ image: alpine
+ script: ['true']
+`;
+ await withHarness({ pipeline }, async (h) => {
+ const job = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().data.job_id;
+ await h.waitForJob(job);
+
+ const build = await claimNamed(h, 'build', {});
+ await finish(h, build.id, {
+ success: true,
+ outputs: [{ name: 'tag', value: 'v1.2.3', masked: false }],
+ });
+
+ const row = await h.services.db.get(
+ 'SELECT value FROM task_outputs WHERE task_id = {id}', { id: build.id }
+ );
+ assert.ok(!row.value.includes('v1.2.3'), `stored in the clear: ${row.value}`);
+ });
+});
+
+test('a task that reads an output that was never produced fails before it runs', async () => {
+ const pipeline = `
+version: 1
+defaults:
+ image: alpine
+tasks:
+ build:
+ script: ['true']
+ deploy:
+ script: ['ship \${{ tasks.build.outputs.tag }}']
+`;
+ await withHarness({ pipeline }, async (h) => {
+ const job = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().data.job_id;
+ await h.waitForJob(job);
+
+ const build = await claimNamed(h, 'build', {});
+ await finish(h, build.id, { success: true });
+
+ assert.equal(await claimNamed(h, 'deploy', {}), null, 'nothing to hand a worker');
+
+ const tasks = tasksByName(await jobState(h, job));
+ assert.equal(tasks.deploy.state, 'failed');
+ assert.match(tasks.deploy.error, /produced no output "tag"/);
+ assert.equal(tasks.deploy.started_at, null, 'it never started');
+ assert.equal((await jobState(h, job)).job.state, 'failed');
+ });
+});
+
+test('set-state success overrides a non-zero exit', async () => {
+ const pipeline = `
+version: 1
+tasks:
+ check:
+ image: alpine
+ script: ['false']
+`;
+ await withHarness({ pipeline }, async (h) => {
+ const job = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().data.job_id;
+ await h.waitForJob(job);
+
+ const check = await claimNamed(h, 'check', {});
+ const outcome = await finish(h, check.id, { success: false, state: 'success', exitCode: 1 });
+
+ assert.equal(outcome.state, 'success');
+ assert.equal(outcome.job_state, 'success');
+ const tasks = tasksByName(await jobState(h, job));
+ assert.equal(tasks.check.exit_code, 1, 'the exit code is still recorded');
+ });
+});
+
+test('a skipped task is not a failure, and what needed it still runs', async () => {
+ const pipeline = `
+version: 1
+defaults:
+ image: alpine
+tasks:
+ package:
+ script: ['true']
+ index:
+ script: ['reindex']
+ needs: [package]
+`;
+ await withHarness({ pipeline }, async (h) => {
+ const job = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().data.job_id;
+ await h.waitForJob(job);
+
+ const pkg = await claimNamed(h, 'package', {});
+ const outcome = await finish(h, pkg.id, { success: false, state: 'skipped', exitCode: 0 });
+ assert.equal(outcome.state, 'skipped');
+ assert.equal(outcome.job_state, null, 'the job is not settled: index still has to run');
+
+ const index = await claimNamed(h, 'index', {});
+ assert.ok(index, 'a skip is not a failure, so the indexer still runs');
+ const last = await finish(h, index.id, { success: true });
+
+ assert.equal(last.job_state, 'success');
+ const tasks = tasksByName(await jobState(h, job));
+ assert.equal(tasks.package.state, 'skipped');
+ });
+});
+
+test('a skipped task that owed a value fails the one that needed it', async () => {
+ const pipeline = `
+version: 1
+defaults:
+ image: alpine
+tasks:
+ package:
+ script: ['true']
+ index:
+ script: ['reindex \${{ tasks.package.outputs.path }}']
+`;
+ await withHarness({ pipeline }, async (h) => {
+ const job = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().data.job_id;
+ await h.waitForJob(job);
+
+ const pkg = await claimNamed(h, 'package', {});
+ await finish(h, pkg.id, { success: false, state: 'skipped', exitCode: 0 });
+
+ assert.equal(await claimNamed(h, 'index', {}), null);
+ const tasks = tasksByName(await jobState(h, job));
+ assert.equal(tasks.index.state, 'failed');
+ assert.match(tasks.index.error, /produced no output "path"/);
+ });
+});
+
+test('set-state failed fails a task that exited zero', async () => {
+ const pipeline = `
+version: 1
+tasks:
+ check:
+ image: alpine
+ script: ['true']
+`;
+ await withHarness({ pipeline }, async (h) => {
+ const job = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().data.job_id;
+ await h.waitForJob(job);
+
+ const check = await claimNamed(h, 'check', {});
+ const outcome = await finish(h, check.id, { success: false, state: 'failed', exitCode: 0 });
+ assert.equal(outcome.state, 'failed');
+ assert.equal(outcome.job_state, 'failed');
+ });
+});
+
+test('a worker that reports no state at all is still understood', async () => {
+ const pipeline = `
+version: 1
+tasks:
+ check:
+ image: alpine
+ script: ['true']
+`;
+ await withHarness({ pipeline }, async (h) => {
+ const job = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().data.job_id;
+ await h.waitForJob(job);
+
+ const check = await claimNamed(h, 'check', {});
+ const res = await h.app.inject({
+ method: 'POST',
+ url: `/api/v1/tasks/${check.id}/complete`,
+ headers: { ...h.auth, 'content-type': 'application/json' },
+ payload: JSON.stringify({ success: true, exit_code: 0 }),
+ });
+ assert.equal(data(res).state, 'success');
+ });
+});
+
+test('a retry drops what the failed attempt produced', async () => {
+ const pipeline = `
+version: 1
+tasks:
+ flaky:
+ image: alpine
+ script: ['true']
+ max_attempts: 2
+`;
+ await withHarness({ pipeline }, async (h) => {
+ const job = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().data.job_id;
+ await h.waitForJob(job);
+
+ const first = await claimNamed(h, 'flaky', {});
+ await finish(h, first.id, {
+ success: false,
+ exitCode: 1,
+ outputs: [{ name: 'tag', value: 'from-the-bad-run', masked: false }],
+ });
+
+ const rows = await h.services.db.all(
+ 'SELECT name FROM task_outputs WHERE task_id = {id}', { id: first.id }
+ );
+ assert.deepEqual(rows, [], 'a requeued task carries nothing forward');
+ });
+});
+
test('after edges are stored as their own kind', async () => {
const pipeline = `
version: 1
diff --git a/test/dag.test.js b/test/dag.test.js
@@ -89,9 +89,12 @@ test('failure skips transitive dependents, not just direct ones', () => {
assert.equal(skipped.has('island'), false);
});
-test('a skipped dependency never releases a dependent', () => {
+test('a failed dependency never releases a dependent', () => {
assert.equal(runnable(GRAPH, states({ a: 'failed' })).some((j) => j.name === 'b'), false);
- assert.equal(runnable(GRAPH, states({ a: 'skipped' })).some((j) => j.name === 'b'), false);
+});
+
+test('a skipped dependency releases a dependent, since a skip is not a failure', () => {
+ assert.equal(runnable(GRAPH, states({ a: 'skipped' })).some((j) => j.name === 'b'), true);
});
test('allowed failures can satisfy a dependency when asked to', () => {
diff --git a/test/db.test.js b/test/db.test.js
@@ -230,3 +230,48 @@ test('an existing database gains the dependency kind, defaulting to needs', asyn
await cleanup();
}
});
+
+test('an existing database gains task outputs, and drops them with the task', async () => {
+ const { db, dir, cleanup } = await tempDb();
+ try {
+ const real = path.join(process.cwd(), 'migrations', 'sqlite');
+ const staged = path.join(dir, 'migrations');
+ await fs.mkdir(staged);
+
+ const all = (await fs.readdir(real)).filter((f) => f.endsWith('.sql')).sort();
+ const outputsMigration = all.find((f) => f.includes('task_outputs'));
+ assert.ok(outputsMigration, 'expected a migration adding task outputs');
+
+ for (const file of all.filter((f) => f !== outputsMigration)) {
+ await fs.copyFile(path.join(real, file), path.join(staged, file));
+ }
+ await runMigrations(db, { dir: staged, logger: quiet });
+
+ const now = Date.now();
+ await db.run(
+ `INSERT INTO projects (id, name, repo_url, created_at, updated_at)
+ VALUES ('p1', 'p', 'git://x', {now}, {now})`, { now });
+ await db.run(
+ `INSERT INTO jobs (id, project_id, number, head_sha, created_at)
+ VALUES ('j1', 'p1', 1, 'abc', {now})`, { now });
+ await db.run(
+ `INSERT INTO tasks (id, job_id, name, base_name, image, requires, spec, timeout, created_at)
+ VALUES ('t1', 'j1', 't1', 't1', 'alpine', '[]', '{}', 60, {now})`, { now });
+
+ await fs.copyFile(path.join(real, outputsMigration), path.join(staged, outputsMigration));
+ const upgrade = await runMigrations(db, { dir: staged, logger: quiet });
+ assert.deepEqual(upgrade.applied, [outputsMigration]);
+
+ await db.run(
+ `INSERT INTO task_outputs (task_id, name, value, masked, created_at)
+ VALUES ('t1', 'tag', 'sealed', 1, {now})`, { now });
+ const stored = await db.get('SELECT masked FROM task_outputs WHERE task_id = {t}', { t: 't1' });
+ assert.equal(stored.masked, 1, 'masked by default, as a variable is');
+
+ await db.run("DELETE FROM tasks WHERE id = 't1'");
+ const left = await db.all('SELECT name FROM task_outputs');
+ assert.deepEqual(left, [], 'outputs go with the task they came from');
+ } finally {
+ await cleanup();
+ }
+});
diff --git a/test/directives.test.js b/test/directives.test.js
@@ -0,0 +1,167 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+
+import { createDirectives, createScanner } from '../src/worker/directives.js';
+
+function scan(chunks, options = {}) {
+ const problems = [];
+ const masked = [];
+ const directives = createDirectives({
+ onProblem: (m) => problems.push(m),
+ onMask: (v) => masked.push(v),
+ ...options,
+ });
+ const scanner = createScanner(directives);
+ const out = [];
+ for (const chunk of chunks) out.push(scanner.push(Buffer.from(chunk)));
+ out.push(scanner.end());
+ return { log: Buffer.concat(out).toString('utf8'), directives, problems, masked };
+}
+
+test('set-output is taken off the log and defaults to masked', () => {
+ const { log, directives, masked } = scan(['hello\n::set-output token=abcdef\nbye\n']);
+ assert.equal(log, 'hello\nbye\n');
+ assert.deepEqual(directives.outputs, [{ name: 'token', value: 'abcdef', masked: true }]);
+ assert.deepEqual(masked, ['abcdef']);
+});
+
+test('public outputs are not masked', () => {
+ const { directives, masked } = scan(['::set-output tag=v1.2.3::public\n']);
+ assert.deepEqual(directives.outputs, [{ name: 'tag', value: 'v1.2.3', masked: false }]);
+ assert.deepEqual(masked, []);
+});
+
+test('masked may be stated explicitly', () => {
+ const { directives } = scan(['::set-output tag=v1::masked\n']);
+ assert.equal(directives.outputs[0].masked, true);
+});
+
+test('a directive split across chunks is still recognised', () => {
+ const { log, directives } = scan(['out\n::set-out', 'put to', 'ken=abcdef\nmore\n']);
+ assert.equal(log, 'out\nmore\n');
+ assert.deepEqual(directives.outputs, [{ name: 'token', value: 'abcdef', masked: true }]);
+});
+
+test('a directive split one byte at a time is still recognised', () => {
+ const { log, directives } = scan([...'::set-output a=bcdef\ntail\n']);
+ assert.equal(log, 'tail\n');
+ assert.equal(directives.outputs[0].value, 'bcdef');
+});
+
+test('ordinary output is not held back waiting for a newline', () => {
+ const problems = [];
+ const directives = createDirectives({ onProblem: (m) => problems.push(m) });
+ const scanner = createScanner(directives);
+ assert.equal(scanner.push(Buffer.from('progress...')).toString(), 'progress...');
+ assert.equal(scanner.push(Buffer.from(' done\n')).toString(), ' done\n');
+});
+
+test('a trailing line with no newline still parses', () => {
+ const { log, directives } = scan(['::set-state failed']);
+ assert.equal(log, '');
+ assert.equal(directives.state, 'failed');
+});
+
+test('CRLF line endings are tolerated', () => {
+ const { log, directives } = scan(['a\r\n::set-state skipped\r\nb\r\n']);
+ assert.equal(log, 'a\r\nb\r\n');
+ assert.equal(directives.state, 'skipped');
+});
+
+test('last one wins for state and for a repeated output', () => {
+ const { directives } = scan([
+ '::set-state failed\n::set-output a=first\n::set-state success\n::set-output a=second\n',
+ ]);
+ assert.equal(directives.state, 'success');
+ assert.deepEqual(directives.outputs, [{ name: 'a', value: 'second', masked: true }]);
+});
+
+test('a line that is not a directive is left alone', () => {
+ const { log, problems } = scan(['::not-a-command whatever\n:: spaced\nplain\n']);
+ assert.equal(log, '::not-a-command whatever\n:: spaced\nplain\n');
+ assert.deepEqual(problems, []);
+});
+
+test('a malformed set-output is reported and still stripped', () => {
+ const { log, problems, directives } = scan(['::set-output token=abcdef::typo\n']);
+ assert.equal(log, '', 'the value must not reach the log');
+ assert.equal(directives.outputs.length, 0);
+ assert.match(problems[0], /unknown visibility/);
+});
+
+test('an output name may carry dots and dashes', () => {
+ const { directives, problems } = scan([
+ '::set-output myname-x86_64=/pkg/a.tar::public\n::set-output pizza.calzone=tasty::public\n',
+ ]);
+ assert.deepEqual(problems, []);
+ assert.deepEqual(directives.outputs, [
+ { name: 'myname-x86_64', value: '/pkg/a.tar', masked: false },
+ { name: 'pizza.calzone', value: 'tasty', masked: false },
+ ]);
+});
+
+test('a bad output name is reported and stripped', () => {
+ for (const name of ['9lives', 'pizza..calzone', 'pizza.', '.pizza', 'a b']) {
+ const { log, problems } = scan([`::set-output ${name}=x\n`]);
+ assert.equal(log, '', `${name} must not reach the log`);
+ assert.match(problems[0], /is not a valid name/, `${name} should be rejected`);
+ }
+});
+
+test('set-output without name=value is reported', () => {
+ const { problems } = scan(['::set-output nope\n']);
+ assert.match(problems[0], /wants name=value/);
+});
+
+test('an unknown state is reported and stripped', () => {
+ const { log, problems, directives } = scan(['::set-state banana\n']);
+ assert.equal(log, '');
+ assert.equal(directives.state, null);
+ assert.match(problems[0], /unknown state/);
+});
+
+test('a value may contain an equals sign', () => {
+ const { directives } = scan(['::set-output data=a=b=c\n']);
+ assert.equal(directives.outputs[0].value, 'a=b=c');
+});
+
+test('a directive must start at column 0', () => {
+ const { log, directives } = scan([' ::set-output a=b\n']);
+ assert.equal(log, ' ::set-output a=b\n');
+ assert.equal(directives.outputs.length, 0);
+});
+
+test('trailing whitespace after a directive is tolerated', () => {
+ const { directives } = scan(['::set-state success \n']);
+ assert.equal(directives.state, 'success');
+});
+
+test('binary output passes through untouched', () => {
+ const bytes = Buffer.from([0x00, 0xff, 0xfe, 0x0a, 0x01, 0x02]);
+ const directives = createDirectives();
+ const scanner = createScanner(directives);
+ const out = Buffer.concat([scanner.push(bytes), scanner.end()]);
+ assert.deepEqual(out, bytes);
+});
+
+test('outputs are capped and the excess is dropped without leaking', () => {
+ const lines = [];
+ for (let i = 0; i < 70; i += 1) lines.push(`::set-output n${i}=value${i}\n`);
+ const { log, directives, problems } = scan([lines.join('')]);
+ assert.equal(log, '');
+ assert.equal(directives.outputs.length, 64);
+ assert.match(problems[0], /more than 64 outputs/);
+});
+
+test('stdout and stderr scanners do not corrupt each other', () => {
+ const directives = createDirectives();
+ const out = createScanner(directives);
+ const err = createScanner(directives);
+ // Interleaved halves of two different directives.
+ out.push(Buffer.from('::set-out'));
+ err.push(Buffer.from('::set-sta'));
+ out.push(Buffer.from('put a=bcdef\n'));
+ err.push(Buffer.from('te success\n'));
+ assert.equal(directives.state, 'success');
+ assert.deepEqual(directives.outputs, [{ name: 'a', value: 'bcdef', masked: true }]);
+});
diff --git a/test/pipeline.test.js b/test/pipeline.test.js
@@ -548,3 +548,150 @@ tasks:
b: { image: x, script: [s], after: [a] }
`), /dependency cycle/);
});
+
+// --- task outputs ---
+
+test('reading an output implies needing the task, and is left for dispatch', () => {
+ const p = compilePipeline(`
+version: 1
+defaults:
+ image: x
+tasks:
+ build:
+ script: [make]
+ deploy:
+ script: ['ship \${{ tasks.build.outputs.tag }}']
+`);
+ const deploy = byName(p).deploy;
+ assert.deepEqual(deploy.needs, ['build'], 'the reference is a reason to wait');
+ assert.deepEqual(deploy.output_refs, { build: ['build'] });
+ assert.deepEqual(
+ deploy.script,
+ ['ship ${{ tasks.build.outputs.tag }}'],
+ 'the value is not known yet, so the reference survives compilation'
+ );
+ assert.equal(deploy.depth, 1);
+});
+
+test('an implied need is not repeated when it is also declared', () => {
+ const p = compilePipeline(`
+version: 1
+defaults:
+ image: x
+tasks:
+ build:
+ script: [make]
+ deploy:
+ script: ['ship \${{ tasks.build.outputs.tag }}']
+ needs: [build]
+`);
+ assert.deepEqual(byName(p).deploy.needs, ['build']);
+});
+
+test('reading an output of a matrix task waits for every instance of it', () => {
+ const p = compilePipeline(`
+version: 1
+defaults:
+ image: x
+tasks:
+ build:
+ arch: [x86_64, aarch64]
+ script: ['echo "::set-output pkg-\${{ arch }}=/out/\${{ arch }}.tar::public"']
+ index:
+ script: ['index \${{ tasks.build.outputs.pkg-x86_64 }} \${{ tasks.build.outputs.pkg-aarch64 }}']
+`);
+ const index = byName(p).index;
+ assert.deepEqual(index.needs, ['build:arch=aarch64', 'build:arch=x86_64']);
+ assert.deepEqual(index.output_refs, { build: ['build:arch=aarch64', 'build:arch=x86_64'] });
+});
+
+test('a shared dimension still narrows an output reference to its own instance', () => {
+ const p = compilePipeline(`
+version: 1
+defaults:
+ image: x
+tasks:
+ build:
+ arch: [x86_64, aarch64]
+ script: [make]
+ sign:
+ arch: [x86_64, aarch64]
+ script: ['sign \${{ tasks.build.outputs.path }}']
+`);
+ const tasks = byName(p);
+ assert.deepEqual(tasks['sign:arch=x86_64'].output_refs, { build: ['build:arch=x86_64'] });
+ assert.deepEqual(tasks['sign:arch=aarch64'].needs, ['build:arch=aarch64']);
+});
+
+test('an output name may carry dots and dashes', () => {
+ const p = compilePipeline(`
+version: 1
+defaults:
+ image: x
+tasks:
+ cook:
+ script: [make]
+ eat:
+ script: ['eat \${{ tasks.cook.outputs.pizza.calzone }} \${{ tasks.cook.outputs.side-dish }}']
+`);
+ assert.deepEqual(byName(p).eat.needs, ['cook']);
+});
+
+test('outputs cannot be read where a worker is chosen', () => {
+ assert.throws(
+ () => compilePipeline(`
+version: 1
+defaults:
+ image: x
+tasks:
+ build:
+ script: [make]
+ deploy:
+ script: [ship]
+ requires: ['\${{ tasks.build.outputs.feature }}']
+`),
+ /tasks\.deploy\.requires\[0\][\s\S]*not known until the task is dispatched/
+ );
+});
+
+test('an output reference is validated like any other dependency', () => {
+ assert.throws(
+ () => compilePipeline(`
+version: 1
+tasks:
+ a: { image: x, script: ['\${{ tasks.ghost.outputs.v }}'] }
+`),
+ /unknown task "ghost"/
+ );
+ assert.throws(
+ () => compilePipeline(`
+version: 1
+tasks:
+ a: { image: x, script: ['\${{ tasks.a.outputs.v }}'] }
+`),
+ /cannot depend on itself/
+ );
+});
+
+test('an output reference can close a cycle', () => {
+ assert.throws(
+ () => compilePipeline(`
+version: 1
+defaults:
+ image: x
+tasks:
+ a:
+ script: ['\${{ tasks.b.outputs.v }}']
+ b:
+ script: ['\${{ tasks.a.outputs.v }}']
+`),
+ /dependency cycle/
+ );
+});
+
+test('an unknown expression names task outputs as an option', () => {
+ assert.throws(
+ () => compilePipeline('version: 1\ntasks:\n a: { image: x, script: ["${{ nonsense }}"] }\n'),
+ /expected arch, matrix\.<name> or tasks\.<task>\.outputs\.<name>/
+ );
+});
diff --git a/test/ui.test.js b/test/ui.test.js
@@ -422,10 +422,10 @@ test('a mutation without the htmx header is refused', async () => {
test('the task graph is one table, so columns line up across stages', () => {
const job = { id: 'j1', state: 'running' };
const tasks = [
- { id: 't1', name: 'style', state: 'success', arch: 'x86-64', worker_name: 'w1', started_at: 1, finished_at: 2, exit_code: 0, needs: [] },
- { id: 't2', name: 'test', state: 'success', arch: null, worker_name: 'w2', started_at: 1, finished_at: 2, exit_code: 0, needs: [] },
- { id: 't3', name: 'images', state: 'queued', arch: null, worker_name: null, started_at: null, finished_at: null, exit_code: null, needs: ['t1', 't2'] },
- { id: 't4', name: 'publish', state: 'queued', arch: null, worker_name: null, started_at: null, finished_at: null, exit_code: null, needs: ['t3'] },
+ { id: 't1', name: 'style', state: 'success', arch: 'x86-64', worker_name: 'w1', started_at: 1, finished_at: 2, needs: [] },
+ { id: 't2', name: 'test', state: 'success', arch: null, worker_name: 'w2', started_at: 1, finished_at: 2, needs: [] },
+ { id: 't3', name: 'images', state: 'queued', arch: null, worker_name: null, started_at: null, finished_at: null, needs: ['t1', 't2'] },
+ { id: 't4', name: 'publish', state: 'queued', arch: null, worker_name: null, started_at: null, finished_at: null, needs: ['t3'] },
];
const out = toHtml(tasksTable(job, tasks));
@@ -438,11 +438,15 @@ test('the task graph is one table, so columns line up across stages', () => {
assert.match(out, /stage 1/);
assert.match(out, /stage 3/);
- // Every task row carries the same six cells, so a column means the same
+ // Every task row carries the same five cells, so a column means the same
// thing all the way down.
for (const row of out.match(/<tr>\s*<td>.*?<\/tr>/gs) ?? []) {
- assert.equal((row.match(/<td/g) ?? []).length, 6, `row has the wrong cell count: ${row}`);
+ assert.equal((row.match(/<td/g) ?? []).length, 5, `row has the wrong cell count: ${row}`);
}
+
+ // The stage label spans the row, so it must track the column count.
+ assert.match(out, /colspan="5"/);
+ assert.doesNotMatch(out, /<th>exit<\/th>/, 'the exit code is no longer a column');
});
test('the sign in form is centred rather than stranded at the left edge', async () => {
diff --git a/test/worker.test.js b/test/worker.test.js
@@ -415,6 +415,40 @@ test('a secret split across two writes is still masked', async () => {
assert.ok(client.text.includes('[masked]'));
});
+test('a secret named while the task runs is masked from then on', async () => {
+ const client = recordingClient();
+ const log = createLogStream(client, 'http://x/log', { flushMs: 5 });
+
+ assert.equal(log.addSecret('abcdef123456'), true);
+ log.write('token is abcdef123456 ok\n');
+ await log.close();
+
+ assert.ok(!client.text.includes('abcdef123456'), `leaked: ${client.text}`);
+ assert.ok(client.text.includes('[masked]'));
+});
+
+test('naming a secret also masks output still buffered', async () => {
+ const client = recordingClient();
+ const log = createLogStream(client, 'http://x/log', { flushMs: 50 });
+
+ log.write('echoed abcdef123456 early\n');
+ log.addSecret('abcdef123456');
+ await log.close();
+
+ assert.ok(!client.text.includes('abcdef123456'), `leaked: ${client.text}`);
+});
+
+test('a value too short to mask is refused rather than made into noise', async () => {
+ const client = recordingClient();
+ const log = createLogStream(client, 'http://x/log', { flushMs: 5 });
+
+ assert.equal(log.addSecret('ab'), false);
+ log.write('ab is everywhere\n');
+ await log.close();
+
+ assert.equal(client.text, 'ab is everywhere\n');
+});
+
test('a failing conductor does not break the task', async () => {
const client = {
async appendLog() {
