conductor

CI task system
git clone git://git.finwo.net/app/conductor
Log | Files | Refs | README | LICENSE

commit 4d1cc2341a159721b2f54bebef03c9f5dcd7719e
parent 2c4cae92a912e4952bcbf6fd2dcf6119b1523320
Author: finwo <finwo@pm.me>
Date:   Fri, 25 Sep 2026 02:47:19 +0200

Let a task wait for another without requiring it to succeed

Diffstat:
MREADME.md | 9+++++++++
Mdocs/api.md | 6++++--
Mdocs/pipeline.md | 38++++++++++++++++++++++++++++++++++++++
Amigrations/mysql/003_task_dep_kind.sql | 9+++++++++
Amigrations/postgres/003_task_dep_kind.sql | 9+++++++++
Amigrations/sqlite/003_task_dep_kind.sql | 9+++++++++
Msrc/conductor/routes/tasks.js | 4+++-
Msrc/conductor/scheduler.js | 39++++++++++++++++++++++++++++-----------
Msrc/conductor/ui/pages.js | 6++++--
Msrc/conductor/ui/routes.js | 8++++----
Msrc/lib/pipeline/dag.js | 36+++++++++++++++++++++++++++---------
Msrc/lib/pipeline/expand.js | 109++++++++++++++++++++++++++++++++++++++++++++-----------------------------------
Msrc/lib/pipeline/only.js | 14++++++++++++--
Msrc/lib/pipeline/parse.js | 12+++++++-----
Mtest/conductor.test.js | 113++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++---
Mtest/dag.test.js | 55+++++++++++++++++++++++++++++++++++++++++++++++++++++++
Mtest/db.test.js | 43+++++++++++++++++++++++++++++++++++++++++++
Mtest/only.test.js | 49+++++++++++++++++++++++++++++++++++++++++++++++++
Mtest/pipeline.test.js | 108+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
19 files changed, 588 insertions(+), 88 deletions(-)

diff --git a/README.md b/README.md @@ -176,6 +176,11 @@ tasks: needs: [package, test] script: - ./publish.sh + + report: + after: [publish] + script: + - ./report.sh ``` `arch` and `matrix` fan a task out into one task per combination. A dependency @@ -183,6 +188,10 @@ 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. +`needs` waits for a task to succeed. `after` waits only for it to finish, so +`report` runs whether the run went well or badly, which is usually when a +report is worth having. + `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/docs/api.md b/docs/api.md @@ -197,7 +197,7 @@ empty body. "image": "debian:bookworm-slim", "attempt": 1, "max_attempts": 1, "allow_failure": false, "exit_code": 0, "error": null, "worker_name": "builder-1", "timeout": 3600, "workdir": "/work", - "needs": ["build:arch=x86_64"], "matrix": { "pkg": "musl" }, + "needs": ["build:arch=x86_64"], "after": [], "matrix": { "pkg": "musl" }, "artifact_paths": ["dist/**"], "log_size": 4096, "created_at": 1800000000000, "started_at": null, "finished_at": null, "project_id": "demo", "job_id": "m2y5rj7ykdm9sfbr54hf", "job_number": 12, @@ -309,7 +309,9 @@ Requests are not enveloped; only responses are: ``` A failure with attempts remaining requeues the task; otherwise everything -downstream of it is skipped. +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. Artifacts --------- diff --git a/docs/pipeline.md b/docs/pipeline.md @@ -66,6 +66,7 @@ Tasks | `image` | string | none, required | yes | | `script` | list of strings | none, required | no | | `needs` | list | `[]` | no | +| `after` | list | `[]` | no | | `arch` | string or list | none | yes | | `matrix` | mapping | none | no | | `requires` | list of strings | `[]` | yes | @@ -220,6 +221,43 @@ 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. +### Running after something, whatever became of it + +`after` waits for a task to finish without requiring it to have succeeded: + +```yaml +test: + script: ['./test.sh'] + +report: + after: [test] + script: ['./report.sh'] +``` + +`report` runs once `test` has stopped, whether it passed, failed, or was +itself skipped because something it needed failed. This is what to reach for +with teardown, notification and result collection, which are worth running +precisely when the run went badly. + +`after` takes the same three forms as `needs` and matches shared dimensions +the same way, because the matching decides which instance is meant and not +what waiting on it implies. Naming one task under both keys is the same as +naming it under `needs` alone. + +Two things `after` deliberately does not change: + + It does not rescue the run. A failed task still fails the job, whether or + not something ran after it. Use `allow_failure` on the task itself to say + that its failure does not count. + + It does not supply anything. Artifacts from a task that failed may be + absent or half written, so a task that reads them wants `needs`. + +Where `only` keeps a task out of a run, its `after` dependents still run: +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. + Worker features --------------- diff --git a/migrations/mysql/003_task_dep_kind.sql b/migrations/mysql/003_task_dep_kind.sql @@ -0,0 +1,9 @@ +-- 003_task_dep_kind.sql - two kinds of dependency edge +-- +-- 'needs' is the original meaning: the dependency must have succeeded. +-- 'after' waits only for the dependency to have finished, whatever it +-- finished as, and is not followed when a failure skips dependents. +-- +-- Every existing edge is a 'needs', which the default records. + +ALTER TABLE task_deps ADD COLUMN kind VARCHAR(16) NOT NULL DEFAULT 'needs'; diff --git a/migrations/postgres/003_task_dep_kind.sql b/migrations/postgres/003_task_dep_kind.sql @@ -0,0 +1,9 @@ +-- 003_task_dep_kind.sql - two kinds of dependency edge +-- +-- 'needs' is the original meaning: the dependency must have succeeded. +-- 'after' waits only for the dependency to have finished, whatever it +-- finished as, and is not followed when a failure skips dependents. +-- +-- Every existing edge is a 'needs', which the default records. + +ALTER TABLE task_deps ADD COLUMN kind TEXT NOT NULL DEFAULT 'needs'; diff --git a/migrations/sqlite/003_task_dep_kind.sql b/migrations/sqlite/003_task_dep_kind.sql @@ -0,0 +1,9 @@ +-- 003_task_dep_kind.sql - two kinds of dependency edge +-- +-- 'needs' is the original meaning: the dependency must have succeeded. +-- 'after' waits only for the dependency to have finished, whatever it +-- finished as, and is not followed when a failure skips dependents. +-- +-- Every existing edge is a 'needs', which the default records. + +ALTER TABLE task_deps ADD COLUMN kind TEXT NOT NULL DEFAULT 'needs'; diff --git a/src/conductor/routes/tasks.js b/src/conductor/routes/tasks.js @@ -94,6 +94,7 @@ export default async function taskRoutes(fastify, { cfg, db, projects, auth = nu // rather than what the pipeline says today. workdir: spec.workdir, needs: spec.needs, + after: spec.after, matrix: spec.matrix, // What the task was told to collect, which is not the same as what // it managed to store. The list below is what exists. @@ -178,13 +179,14 @@ function readSpec(raw) { spec = null; } if (!spec || typeof spec !== 'object') { - return { workdir: null, needs: [], matrix: {}, artifactPaths: [] }; + return { workdir: null, needs: [], after: [], matrix: {}, artifactPaths: [] }; } const paths = spec.artifacts?.paths; return { workdir: typeof spec.workdir === 'string' ? spec.workdir : null, needs: Array.isArray(spec.needs) ? spec.needs : [], + after: Array.isArray(spec.after) ? spec.after : [], matrix: spec.matrix && typeof spec.matrix === 'object' ? spec.matrix : {}, artifactPaths: Array.isArray(paths) ? paths : [], }; diff --git a/src/conductor/scheduler.js b/src/conductor/scheduler.js @@ -35,8 +35,10 @@ 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. A failed dependency marked allow_failure still counts as - // satisfied, which is the whole point of the flag. + // 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. 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, @@ -51,9 +53,18 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl FROM task_deps d 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 NOT (dt.state = 'failed' AND dt.allow_failure = 1) ) + AND NOT EXISTS ( + SELECT 1 + FROM task_deps d + JOIN tasks dt ON dt.id = d.depends_on_id + WHERE d.task_id = t.id + AND d.kind = 'after' + AND dt.state IN ('queued', 'running') + ) `; async function loadGraph(tx, jobId) { @@ -62,15 +73,15 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl { job: jobId } ); const deps = await tx.all( - `SELECT d.task_id, d.depends_on_id + `SELECT d.task_id, d.depends_on_id, d.kind FROM task_deps d JOIN tasks t ON t.id = d.task_id WHERE t.job_id = {job}`, { job: jobId } ); - const needs = new Map(tasks.map((t) => [t.id, []])); - for (const d of deps) needs.get(d.task_id)?.push(d.depends_on_id); - return tasks.map((t) => ({ ...t, needs: needs.get(t.id) ?? [] })); + const edges = new Map(tasks.map((t) => [t.id, { needs: [], after: [] }])); + for (const d of deps) edges.get(d.task_id)?.[d.kind === 'after' ? 'after' : 'needs'].push(d.depends_on_id); + return tasks.map((t) => ({ ...t, ...(edges.get(t.id) ?? { needs: [], after: [] }) })); } // Reads the pipeline file at a commit. @@ -203,6 +214,7 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl artifacts: task.artifacts, matrix: task.matrix, needs: task.needs, + after: task.after, depth: task.depth, workdir, }), @@ -215,11 +227,13 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl } for (const task of pipeline.tasks) { - for (const need of task.needs) { - await tx.run( - 'INSERT INTO task_deps (task_id, depends_on_id) VALUES ({task}, {dep})', - { task: idByName.get(task.name), dep: idByName.get(need) } - ); + for (const [kind, deps] of [['needs', task.needs], ['after', task.after ?? []]]) { + for (const dep of deps) { + await tx.run( + 'INSERT INTO task_deps (task_id, depends_on_id, kind) VALUES ({task}, {dep}, {kind})', + { task: idByName.get(task.name), dep: idByName.get(dep), kind } + ); + } } } }); @@ -527,6 +541,9 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl 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 + // whatever the outcome was, so the failure stops here rather than + // carrying it off with everything else downstream. const reachable = transitiveDependents( graph.map((t) => ({ name: t.id, needs: t.needs })), fromTaskIds diff --git a/src/conductor/ui/pages.js b/src/conductor/ui/pages.js @@ -76,12 +76,14 @@ export function jobPage({ job, tasks, canManage }) { export function tasksTable(job, tasks) { const live = ACTIVE_JOB_STATES.includes(job.state); - // Grouped by graph depth, which is how the pipeline reads. + // Grouped by graph depth, which is how the pipeline reads. Both kinds of + // edge count: an after task runs later even though it waits for less, and + // showing it alongside what it waits for would read as running beside it. const byId = new Map(tasks.map((j) => [j.id, j])); const depthOf = (task, seen = new Set()) => { if (seen.has(task.id)) return 0; seen.add(task.id); - const deps = (task.needs ?? []).map((id) => byId.get(id)).filter(Boolean); + const deps = [...(task.needs ?? []), ...(task.after ?? [])].map((id) => byId.get(id)).filter(Boolean); return deps.length === 0 ? 0 : Math.max(...deps.map((d) => depthOf(d, seen) + 1)); }; diff --git a/src/conductor/ui/routes.js b/src/conductor/ui/routes.js @@ -194,13 +194,13 @@ export default async function uiRoutes(fastify, services) { { job: jobId } ); const deps = await db.all( - `SELECT d.task_id, d.depends_on_id FROM task_deps d + `SELECT d.task_id, d.depends_on_id, d.kind FROM task_deps d JOIN tasks t ON t.id = d.task_id WHERE t.job_id = {job}`, { job: jobId } ); - const needs = new Map(tasks.map((t) => [t.id, []])); - for (const d of deps) needs.get(d.task_id)?.push(d.depends_on_id); - return tasks.map((j) => ({ ...j, needs: needs.get(j.id) ?? [] })); + const edges = new Map(tasks.map((t) => [t.id, { needs: [], after: [] }])); + for (const d of deps) edges.get(d.task_id)?.[d.kind === 'after' ? 'after' : 'needs'].push(d.depends_on_id); + return tasks.map((j) => ({ ...j, ...(edges.get(j.id) ?? { needs: [], after: [] }) })); } fastify.get('/jobs/:id', async (req, reply) => { diff --git a/src/lib/pipeline/dag.js b/src/lib/pipeline/dag.js @@ -6,6 +6,20 @@ // still be dispatched with its dependency missing. Both problems come from // throwing the graph away, so the graph is kept here and used directly. +// An edge is one of two kinds. `needs` waits for the dependency to succeed, +// `after` only waits for it to finish. Ordering and cycle detection care +// about both, since either kind makes one task wait for another. Failure +// propagation cares about `needs` alone, which is the whole point of `after`. +export const ALL_EDGES = ['needs', 'after']; + +// Terminal states, which is what an `after` edge waits for. +const FINISHED = ['success', 'failed', 'skipped', 'cancelled']; + +function depsOf(task, edges) { + if (edges.length === 1) return task[edges[0]] ?? []; + return edges.flatMap((kind) => task[kind] ?? []); +} + export class CycleError extends Error { constructor(cycle) { super(`pipeline contains a dependency cycle: ${cycle.join(' -> ')}`); @@ -16,7 +30,7 @@ export class CycleError extends Error { // Returns the tasks in an order where every dependency precedes its // dependents. Throws CycleError naming the cycle when none exists. -export function topologicalOrder(tasks) { +export function topologicalOrder(tasks, { edges = ALL_EDGES } = {}) { const byName = new Map(tasks.map((j) => [j.name, j])); const state = new Map(); const order = []; @@ -32,7 +46,7 @@ export function topologicalOrder(tasks) { state.set(name, 'active'); stack.push(name); - for (const dep of byName.get(name).needs) { + for (const dep of depsOf(byName.get(name), edges)) { if (byName.has(dep)) visit(dep); } stack.pop(); @@ -47,11 +61,11 @@ export function topologicalOrder(tasks) { // Depth of each task, where a task with no dependencies is 0. Used only for // display; scheduling reads the edges, never the depth. -export function depths(tasks) { +export function depths(tasks, { edges = ALL_EDGES } = {}) { const byName = new Map(tasks.map((j) => [j.name, j])); const out = new Map(); - for (const task of topologicalOrder(tasks)) { - const deps = task.needs.filter((d) => byName.has(d)); + for (const task of topologicalOrder(tasks, { edges })) { + const deps = depsOf(task, edges).filter((d) => byName.has(d)); out.set(task.name, deps.length === 0 ? 0 : Math.max(...deps.map((d) => out.get(d))) + 1); } return out; @@ -60,10 +74,13 @@ export function depths(tasks) { // Every task reachable by following dependents from the given names. This is // what a failure must skip: direct dependents alone leave transitive ones // runnable against a dependency that never produced anything. -export function transitiveDependents(tasks, startNames) { +// +// Needs edges only, by default. A task declaring `after` asked to run +// whatever became of its predecessor, so a failure must not carry it off. +export function transitiveDependents(tasks, startNames, { edges = ['needs'] } = {}) { const dependents = new Map(tasks.map((j) => [j.name, []])); for (const task of tasks) { - for (const dep of task.needs) { + for (const dep of depsOf(task, edges)) { if (dependents.has(dep)) dependents.get(dep).push(task.name); } } @@ -83,9 +100,10 @@ export function transitiveDependents(tasks, startNames) { // 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'] } = {}) { +export function runnable(tasks, stateByName, { satisfied = ['success'], finished = FINISHED } = {}) { return tasks.filter((task) => { if (stateByName.get(task.name) !== 'queued') return false; - return task.needs.every((dep) => satisfied.includes(stateByName.get(dep))); + if (!(task.needs ?? []).every((dep) => satisfied.includes(stateByName.get(dep)))) return false; + return (task.after ?? []).every((dep) => finished.includes(stateByName.get(dep))); }); } diff --git a/src/lib/pipeline/expand.js b/src/lib/pipeline/expand.js @@ -1,7 +1,7 @@ // src/lib/pipeline/expand.js - matrix and architecture expansion // // Turns each task template into one concrete task per combination of its -// dimensions, then resolves needs between the concrete tasks. +// dimensions, then resolves needs and after between the concrete tasks. // // Dimension matching is the part worth understanding. When a task declares // needs on a template that shares a dimension with it, the dependency is @@ -147,6 +147,7 @@ export function expandPipeline(pipeline, options = {}) { // restricted task stays restricted across all of its expansions. only: task.only ?? null, needs: [], + after: [], }); } @@ -156,63 +157,75 @@ export function expandPipeline(pipeline, options = {}) { const byName = new Map(instances.map((i) => [i.name, i])); - // Pass two: resolve needs now that every concrete task exists. + // Pass two: resolve dependencies now that every concrete task exists. for (const [baseName, entry] of byBase) { for (const instance of entry.instances) { - const resolved = new Set(); - - for (const need of entry.task.needs) { - const path = `tasks.${baseName}.needs`; - - // 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; - } + instance.needs = resolveDeps(problems, { byBase, byName, baseName, entry, instance, field: 'needs' }); + instance.after = resolveDeps(problems, { byBase, byName, baseName, entry, instance, field: 'after' }); + + // 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)); + } + } + + problems.throwIfAny(pipeline.source); + return instances; +} + +// One task's declared dependencies, turned into concrete task names. `needs` +// and `after` resolve identically: dimension matching is about which +// instance is meant, not about what waiting on it implies. +function resolveDeps(problems, { byBase, byName, baseName, entry, instance, field }) { + const resolved = new Set(); + 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 target = byBase.get(need.task); - if (!target) { - const known = [...byBase.keys()].filter((n) => n !== baseName); + 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) { + problems.add(path, `task ${JSON.stringify(baseName)} cannot depend on itself`); + continue; + } + + 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, - `refers to unknown task ${JSON.stringify(need.task)}` + - (known.length > 0 ? `; defined tasks are ${known.join(', ')}` : '') + `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; } - if (need.task === baseName) { - problems.add(path, `task ${JSON.stringify(baseName)} cannot depend on itself`); - continue; - } - - 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; - } - } - } - - for (const candidate of candidates) resolved.add(candidate.name); } - - instance.needs = [...resolved].sort(); } + + for (const candidate of candidates) resolved.add(candidate.name); } - problems.throwIfAny(pipeline.source); - return instances; + return [...resolved].sort(); } diff --git a/src/lib/pipeline/only.js b/src/lib/pipeline/only.js @@ -7,10 +7,14 @@ // right reading when a dependency collapsed, and the wrong one for a // publish step that was never meant to run on this branch. // -// Anything depending on an excluded task is excluded with it. The +// Anything that needs an excluded task is excluded with it. The // alternative, quietly dropping the dependency, would let a task run // without something it declared it needed, which is a worse surprise than // the task not running. +// +// An `after` edge is the other way round. It asks for ordering, not for +// output, and a task that was never going to run has nothing to order +// against, so the edge is dropped and the dependent stays in the run. import { transitiveDependents } from './dag.js'; @@ -47,5 +51,11 @@ export function selectForRef(tasks, ref) { for (const name of transitiveDependents(tasks, [...excluded])) excluded.add(name); - return tasks.filter((task) => !excluded.has(task.name)); + // Surviving after edges pointing into the excluded set are pruned rather + // than left dangling, since everything downstream resolves them by name. + return tasks + .filter((task) => !excluded.has(task.name)) + .map((task) => (task.after?.some((name) => excluded.has(name)) + ? { ...task, after: task.after.filter((name) => !excluded.has(name)) } + : task)); } diff --git a/src/lib/pipeline/parse.js b/src/lib/pipeline/parse.js @@ -67,7 +67,7 @@ export function parseWorkdir(problems, path, raw) { } const JOB_KEYS = [ - 'image', 'script', 'needs', 'arch', 'matrix', 'requires', 'services', + 'image', 'script', 'needs', 'after', 'arch', 'matrix', 'requires', 'services', 'env', 'artifacts', 'allow_failure', 'timeout', 'max_attempts', 'only', ]; @@ -238,7 +238,8 @@ function parseTask(problems, path, raw, defaults) { return { image, script, - needs: parseNeeds(problems, `${path}.needs`, raw.needs), + needs: parseDeps(problems, `${path}.needs`, raw.needs), + after: parseDeps(problems, `${path}.after`, raw.after), arch, matrix, requires: raw.requires === undefined ? defaults.requires : asStringList(problems, `${path}.requires`, raw.requires, { max: 64 }), @@ -292,9 +293,10 @@ function parseOnly(problems, path, raw) { return { refs }; } -// A need is either a task name, or a mapping for the cases where the default -// dimension matching is not what is wanted. -function parseNeeds(problems, path, raw) { +// A dependency is either a task name, or a mapping for the cases where the +// default dimension matching is not what is wanted. `needs` and `after` take +// the same forms and differ only in what the scheduler makes of the edge. +function parseDeps(problems, path, raw) { if (raw === undefined) return []; const list = Array.isArray(raw) ? raw : [raw]; const out = []; diff --git a/test/conductor.test.js b/test/conductor.test.js @@ -33,13 +33,13 @@ async function jobState(h, jobId) { 'SELECT * FROM tasks WHERE job_id = {job} ORDER BY name', { job: jobId } ); const deps = await h.services.db.all( - `SELECT d.task_id, d.depends_on_id FROM task_deps d + `SELECT d.task_id, d.depends_on_id, d.kind FROM task_deps d JOIN tasks t ON t.id = d.task_id WHERE t.job_id = {job}`, { job: jobId } ); - const needs = new Map(tasks.map((t) => [t.id, []])); - for (const d of deps) needs.get(d.task_id)?.push(d.depends_on_id); - return { job, tasks: tasks.map((t) => ({ ...t, needs: needs.get(t.id) ?? [] })) }; + const edges = new Map(tasks.map((t) => [t.id, { needs: [], after: [] }])); + for (const d of deps) edges.get(d.task_id)?.[d.kind === 'after' ? 'after' : 'needs'].push(d.depends_on_id); + return { job, tasks: tasks.map((t) => ({ ...t, ...(edges.get(t.id) ?? { needs: [], after: [] }) })) }; } // The job a task belongs to. A worker is never told this, so a test that @@ -1274,3 +1274,108 @@ test('worker tokens are stored only as hashes', async () => { } }); }); + +test('after runs its dependent whether the dependency passed or failed', async () => { + const pipeline = ` +version: 1 +defaults: + image: alpine +tasks: + build: + script: ['true'] + test: + script: ['false'] + needs: [build] + notify: + script: ['true'] + after: [test] +`; + 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 }); + + // Draining yields test alone: notify waits for it to finish, exactly as + // needs would. after asks for less, not for nothing. + const released = await drain(h, {}); + assert.ok(released.test, 'test should be released by build'); + assert.ok(!released.notify, 'notify should wait while test runs'); + + const outcome = await finish(h, released.test.id, { success: false, exitCode: 1 }); + // The failure has nothing downstream of it to skip. + assert.deepEqual(outcome.skipped, []); + + const notify = await claimNamed(h, 'notify', {}); + assert.ok(notify, 'notify should run after a failed dependency'); + const last = await finish(h, notify.id, { success: true }); + + // The failure still fails the job. after decides what runs, not what counts. + assert.equal(last.job_state, 'failed'); + const tasks = tasksByName(await jobState(h, job)); + assert.equal(tasks.notify.state, 'success'); + assert.equal(tasks.test.state, 'failed'); + }); +}); + +test('after runs its dependent even when the dependency was skipped', async () => { + const pipeline = ` +version: 1 +defaults: + image: alpine +tasks: + build: + script: ['false'] + test: + script: ['true'] + needs: [build] + cleanup: + script: ['true'] + after: [test] +`; + 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', {}); + const outcome = await finish(h, build.id, { success: false, exitCode: 1 }); + // test is skipped with its dependency, cleanup is not. + assert.deepEqual(outcome.skipped, [(await jobState(h, job)).tasks.find((t) => t.name === 'test').id]); + + const cleanup = await claimNamed(h, 'cleanup', {}); + assert.ok(cleanup, 'cleanup should run once test settled as skipped'); + await finish(h, cleanup.id, { success: true }); + + const tasks = tasksByName(await jobState(h, job)); + assert.equal(tasks.test.state, 'skipped'); + assert.equal(tasks.cleanup.state, 'success'); + }); +}); + +test('after edges are stored as their own kind', async () => { + const pipeline = ` +version: 1 +defaults: + image: alpine +tasks: + a: + script: ['true'] + b: + script: ['true'] + needs: [a] + c: + script: ['true'] + after: [a] +`; + 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 tasks = tasksByName(await jobState(h, job)); + assert.deepEqual(tasks.b.needs, [tasks.a.id]); + assert.deepEqual(tasks.b.after, []); + assert.deepEqual(tasks.c.needs, []); + assert.deepEqual(tasks.c.after, [tasks.a.id]); + }); +}); diff --git a/test/dag.test.js b/test/dag.test.js @@ -108,3 +108,58 @@ test('graph helpers tolerate dependencies outside the set', () => { assert.deepEqual(topologicalOrder(partial).map((j) => j.name), ['only']); assert.equal(depths(partial).get('only'), 0); }); + +// --- after --- + +// notify waits for report to finish, but not for it to have worked. +const AFTER_GRAPH = [ + { name: 'build', needs: [], after: [] }, + { name: 'report', needs: ['build'], after: [] }, + { name: 'notify', needs: [], after: ['report'] }, +]; + +function afterStates(overrides = {}) { + const map = new Map(AFTER_GRAPH.map((j) => [j.name, 'queued'])); + for (const [k, v] of Object.entries(overrides)) map.set(k, v); + return map; +} + +test('after orders a task without putting it downstream of a failure', () => { + const order = topologicalOrder(AFTER_GRAPH).map((j) => j.name); + assert.ok(order.indexOf('report') < order.indexOf('notify')); + + // The whole point: a build failure carries report off with it, and leaves + // notify to run. + const skipped = transitiveDependents(AFTER_GRAPH, ['build']); + assert.deepEqual([...skipped], ['report']); +}); + +test('a cycle through after is still a cycle', () => { + // after waits just as needs does, so a loop of them deadlocks the same way. + const cyclic = [ + { name: 'x', needs: [], after: ['y'] }, + { name: 'y', needs: ['x'], after: [] }, + ]; + assert.throws(() => topologicalOrder(cyclic), CycleError); +}); + +test('an after dependency releases its dependent whatever it finished as', () => { + for (const state of ['success', 'failed', 'skipped', 'cancelled']) { + const ready = runnable(AFTER_GRAPH, afterStates({ build: 'success', report: state })); + assert.ok(ready.some((j) => j.name === 'notify'), `should run after a ${state} dependency`); + } +}); + +test('an unfinished after dependency still holds its dependent back', () => { + for (const state of ['queued', 'running']) { + const ready = runnable(AFTER_GRAPH, afterStates({ report: state })); + assert.equal(ready.some((j) => j.name === 'notify'), false, `should wait on a ${state} dependency`); + } +}); + +test('depth counts after edges, since the task still runs later', () => { + const d = depths(AFTER_GRAPH); + assert.equal(d.get('build'), 0); + assert.equal(d.get('report'), 1); + assert.equal(d.get('notify'), 2); +}); diff --git a/test/db.test.js b/test/db.test.js @@ -187,3 +187,46 @@ test('a failed transaction rolls back, and a failed savepoint keeps the outer wo await cleanup(); } }); + +test('an existing database gains the dependency kind, defaulting to needs', async () => { + // The edges recorded before there were two kinds all meant needs, and a + // database upgraded in place has to keep meaning that. + 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 kindMigration = all.find((f) => f.includes('task_dep_kind')); + assert.ok(kindMigration, 'expected a migration adding the edge kind'); + + for (const file of all.filter((f) => f !== kindMigration)) { + 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 }); + for (const id of ['t1', 't2']) { + await db.run( + `INSERT INTO tasks (id, job_id, name, base_name, image, requires, spec, timeout, created_at) + VALUES ({id}, 'j1', {id}, {id}, 'alpine', '[]', '{}', 60, {now})`, { id, now }); + } + await db.run("INSERT INTO task_deps (task_id, depends_on_id) VALUES ('t2', 't1')"); + + await fs.copyFile(path.join(real, kindMigration), path.join(staged, kindMigration)); + const upgrade = await runMigrations(db, { dir: staged, logger: quiet }); + assert.deepEqual(upgrade.applied, [kindMigration]); + + const edge = await db.get('SELECT kind FROM task_deps WHERE task_id = {t}', { t: 't2' }); + assert.equal(edge.kind, 'needs'); + } finally { + await cleanup(); + } +}); diff --git a/test/only.test.js b/test/only.test.js @@ -158,3 +158,52 @@ tasks: }, ); }); + +test('a task that only comes after an excluded one still runs', () => { + const text = ` +version: 1 +defaults: + image: alpine:3 +tasks: + build: + script: ['make'] + publish: + needs: [build] + only: + refs: [refs/heads/main] + script: ['make publish'] + notify: + after: [publish] + script: ['make notify'] +`; + // notify asked to run after publish, not to be fed by it. On a branch + // where publish never happens there is nothing to wait for, so the edge + // goes and notify stays. + assert.deepEqual(names(text, 'refs/heads/feature'), ['build', 'notify']); + assert.deepEqual(names(text, 'refs/heads/main'), ['build', 'notify', 'publish']); +}); + +test('an after edge into an excluded task is pruned, not left dangling', () => { + // The scheduler resolves these names to task ids, so one pointing at a + // task that is not in the job would be written as a broken edge. + const { tasks } = compilePipeline(` +version: 1 +defaults: + image: alpine:3 +tasks: + build: + script: ['make'] + publish: + only: + refs: [refs/heads/main] + script: ['make publish'] + notify: + after: [build, publish] + script: ['make notify'] +`, { ref: 'refs/heads/feature' }); + + const present = new Set(tasks.map((t) => t.name)); + const notify = tasks.find((t) => t.name === 'notify'); + assert.deepEqual(notify.after, ['build']); + for (const dep of notify.after) assert.ok(present.has(dep), `${dep} should be in the job`); +}); diff --git a/test/pipeline.test.js b/test/pipeline.test.js @@ -440,3 +440,111 @@ tasks: assert.equal(tasks.publish.depth, 2); assert.deepEqual(tasks['package:arch=aarch64,pkg=musl'].requires, ['sign-key']); }); + +// --- after --- + +test('after resolves to task names in its own list, separate from needs', () => { + const p = compilePipeline(` +version: 1 +defaults: + image: x +tasks: + build: + script: [y] + test: + script: [y] + needs: [build] + notify: + script: [y] + after: [test] +`); + const tasks = byName(p); + assert.deepEqual(tasks.notify.needs, []); + assert.deepEqual(tasks.notify.after, ['test']); + // It still sits downstream, so the graph reads in running order. + assert.equal(tasks.notify.depth, 2); +}); + +test('after matches shared dimensions exactly as needs does', () => { + const p = compilePipeline(` +version: 1 +defaults: + image: x +tasks: + build: + script: [y] + arch: [x86_64, aarch64] + cleanup: + script: [y] + arch: [x86_64, aarch64] + after: [build] + report: + script: [y] + after: [build] +`); + const tasks = byName(p); + assert.deepEqual(tasks['cleanup:arch=x86_64'].after, ['build:arch=x86_64']); + assert.deepEqual(tasks['cleanup:arch=aarch64'].after, ['build:arch=aarch64']); + // No shared dimension, so it waits for every instance. + assert.deepEqual(tasks.report.after, ['build:arch=aarch64', 'build:arch=x86_64']); +}); + +test('after accepts the mapping form too', () => { + const p = compilePipeline(` +version: 1 +defaults: + image: x +tasks: + build: + script: [y] + arch: [x86_64, aarch64] + cleanup: + script: [y] + arch: [x86_64, aarch64] + after: + - task: build + match: all +`); + assert.deepEqual( + byName(p)['cleanup:arch=x86_64'].after, + ['build:arch=aarch64', 'build:arch=x86_64'] + ); +}); + +test('a dependency named both ways keeps only the stricter one', () => { + // needs already implies after, so recording both would be a contradiction + // waiting to be read the wrong way round. + const p = compilePipeline(` +version: 1 +defaults: + image: x +tasks: + build: + script: [y] + deploy: + script: [y] + needs: [build] + after: [build] +`); + const deploy = byName(p).deploy; + assert.deepEqual(deploy.needs, ['build']); + assert.deepEqual(deploy.after, []); +}); + +test('after is validated like needs', () => { + assert.throws( + () => compilePipeline('version: 1\ntasks:\n a:\n image: x\n script: [y]\n after: [a]\n'), + /cannot depend on itself/ + ); + assert.throws( + () => compilePipeline('version: 1\ntasks:\n a:\n image: x\n script: [y]\n after: [ghost]\n'), + /tasks\.a\.after[\s\S]*unknown task "ghost"/ + ); + assert.throws( + () => compilePipeline(` +version: 1 +tasks: + a: { image: x, script: [s], after: [b] } + b: { image: x, script: [s], after: [a] } +`), /dependency cycle/); +});