commit 876f2f34a765135d939972f9dbaac807efffcbf9
parent f997ab1673eacee94e95040b6a698507bd03eb6d
Author: finwo <finwo@pm.me>
Date: Sun, 20 Sep 2026 03:24:29 +0200
Rename a run to a job and a job to a task throughout
Diffstat:
34 files changed, 1325 insertions(+), 719 deletions(-)
diff --git a/src/admin-cli.js b/src/admin-cli.js
@@ -13,12 +13,12 @@
// node src/admin-cli.js var:set <project> <NAME> <value> [--visible]
// node src/admin-cli.js var:list <project>
// node src/admin-cli.js var:remove <project> <NAME>
-// node src/admin-cli.js user:add <username> <password> [--role admin|viewer]
+// node src/admin-cli.js user:add <username> <password> [--role admin|user]
// node src/admin-cli.js user:list
// node src/admin-cli.js user:passwd <username> <password>
// node src/admin-cli.js user:remove <username>
-// node src/admin-cli.js run:trigger <project> <sha> [--ref r] [--base b]
-// node src/admin-cli.js run:cancel <run-id>
+// node src/admin-cli.js job:trigger <project> <sha> [--ref r] [--base b]
+// node src/admin-cli.js job:cancel <job-id>
//
// This runs against the database directly, so it works before the first
// account exists and when the server is down.
@@ -67,12 +67,12 @@ function usage(message) {
' var:set <project> <NAME> <value> [--visible]',
' var:list <project>',
' var:remove <project> <NAME>',
- ' user:add <username> <password> [--role admin|viewer]',
+ ' user:add <username> <password> [--role admin|user]',
' user:list',
' user:passwd <username> <password>',
' user:remove <username>',
- ' run:trigger <project> <sha> [--ref r] [--base b]',
- ' run:cancel <run-id>',
+ ' job:trigger <project> <sha> [--ref r] [--base b]',
+ ' job:cancel <job-id>',
].join('\n')
);
process.exit(message ? 1 : 0);
@@ -137,9 +137,9 @@ try {
branch: p.default_branch,
config: p.config_path,
enabled: p.enabled ? 'yes' : 'no',
- runs: p.run_counter,
+ jobs: p.job_counter,
}));
- table(rows, ['id', 'repo_url', 'branch', 'config', 'source', 'enabled', 'runs']);
+ table(rows, ['id', 'repo_url', 'branch', 'config', 'source', 'enabled', 'jobs']);
break;
}
@@ -230,7 +230,7 @@ try {
const user = await users.create({
username,
password,
- role: flags.role ? String(flags.role) : 'viewer',
+ role: flags.role ? String(flags.role) : 'user',
});
console.log(`created user ${user.username} with role ${user.role}`);
break;
@@ -268,28 +268,28 @@ try {
break;
}
- case 'run:trigger': {
+ case 'job:trigger': {
const [projectId, sha] = positional;
- if (!projectId || !sha) usage('run:trigger needs a project and a commit');
+ if (!projectId || !sha) usage('job:trigger needs a project and a commit');
const project = await projects.get(projectId);
if (!project) usage(`unknown project ${projectId}`);
- const result = await scheduler.createRun(project, {
+ const result = await scheduler.createJob(project, {
headSha: sha,
ref: flags.ref ? String(flags.ref) : null,
baseSha: flags.base ? String(flags.base) : null,
trigger: 'manual',
actor: 'admin-cli',
});
- console.log(`created run ${result.runId} with ${result.jobCount} job(s)`);
+ console.log(`created job ${result.jobId} with ${result.taskCount} task(s)`);
break;
}
- case 'run:cancel': {
- const [runId] = positional;
- if (!runId) usage('run:cancel needs a run id');
- const result = await scheduler.cancelRun(runId);
- console.log(result.ok ? `cancelled ${runId}` : `could not cancel: ${result.reason}`);
+ case 'job:cancel': {
+ const [jobId] = positional;
+ if (!jobId) usage('job:cancel needs a job id');
+ const result = await scheduler.cancelJob(jobId);
+ console.log(result.ok ? `cancelled ${jobId}` : `could not cancel: ${result.reason}`);
break;
}
diff --git a/src/conductor/app.js b/src/conductor/app.js
@@ -20,8 +20,8 @@ import { createAuth } from '../lib/auth/index.js';
import { createScheduler } from './scheduler.js';
import { createRetention } from './retention.js';
-import workerRoutes from './routes/workers.js';
-import triggerRoutes from './routes/trigger.js';
+import taskRoutes from './routes/tasks.js';
+import jobRoutes from './routes/jobs.js';
import artifactRoutes from './routes/artifacts.js';
import uiRoutes from './ui/routes.js';
import staticRoutes from '../lib/static.js';
@@ -81,10 +81,11 @@ export async function buildServer(services, options = {}) {
auth: cfg.auth.mode,
}));
- // The only HTTP surface besides the interface: the trigger edge, the
- // worker edge, and artifact downloads. Everything else is the UI.
- await fastify.register(triggerRoutes, { ...services, prefix: '/api/v1' });
- await fastify.register(workerRoutes, { ...services, prefix: '/api/v1' });
+ // The only HTTP surface besides the interface: creating and reading jobs,
+ // the task edge workers poll, and artifact downloads. Everything else
+ // is the UI.
+ await fastify.register(jobRoutes, { ...services, prefix: '/api/v1' });
+ await fastify.register(taskRoutes, { ...services, prefix: '/api/v1' });
await fastify.register(artifactRoutes, { ...services, prefix: '/api/v1' });
// Signing in, managing projects and registering workers all happen in the
diff --git a/src/conductor/retention.js b/src/conductor/retention.js
@@ -1,20 +1,20 @@
// src/conductor/retention.js - deleting build output that is past its time
//
-// Everything a run produces is kept forever unless something removes it,
-// and logs are the worst of it: written for every job, read for almost
+// Everything a job produces is kept forever unless something removes it,
+// and logs are the worst of it: written for every task, read for almost
// none, and never stopping. This sweeps both.
//
// Artifacts follow two rules per project, and an artifact survives if
-// either wants it. The last artifact_keep_runs runs keep their artifacts
+// either wants it. The last artifact_keep_jobs jobs keep their artifacts
// however old they are, and anything younger than artifact_keep_days is
-// kept however many runs have followed. The most recent successful run is
+// kept however many jobs have followed. The most recent successful job is
// kept regardless, so a project that has gone quiet still has something
// to download.
//
-// A job that set artifacts.expire in its pipeline overrides all of that
+// A task that set artifacts.expire in its pipeline overrides all of that
// with an exact deadline, in either direction. That is deliberate: the
// point of the key is to drop a bulky intermediate early, and a
-// node_modules tree is not worth keeping just because its run was green.
+// node_modules tree is not worth keeping just because its job was green.
//
// Logs go by age alone.
//
@@ -72,49 +72,49 @@ export function createRetention({ cfg, db, storage, logs, logger = {} }) {
}
async function sweepProjectArtifacts(project, now, batch) {
- const keepRuns = setting(project.artifact_keep_runs, defaults.artifact_keep_runs);
+ const keepJobs = setting(project.artifact_keep_jobs, defaults.artifact_keep_jobs);
const keepDays = setting(project.artifact_keep_days, defaults.artifact_keep_days);
// Both rules off means keep everything, and with the rules combined by
// whichever keeps longer there is nothing left to delete.
- if (keepRuns === 0 && keepDays === 0) return { deleted: 0, bytes: 0 };
+ if (keepJobs === 0 && keepDays === 0) return { deleted: 0, bytes: 0 };
- const protectedRuns = new Set();
+ const protectedJobs = new Set();
- if (keepRuns > 0) {
+ if (keepJobs > 0) {
const recent = await db.all(
- `SELECT id FROM runs WHERE project_id = {project}
+ `SELECT id FROM jobs WHERE project_id = {project}
ORDER BY number DESC
LIMIT {limit}`,
- { project: project.id, limit: keepRuns }
+ { project: project.id, limit: keepJobs }
);
- for (const run of recent) protectedRuns.add(run.id);
+ for (const job of recent) protectedJobs.add(job.id);
}
- // The last green run, however old. Looked up separately rather than
+ // The last green job, however old. Looked up separately rather than
// folded into the query above, since it may be far outside the recent
// window on a project that has been failing for a while.
const lastGood = await db.get(
- `SELECT id FROM runs WHERE project_id = {project} AND state = 'success'
+ `SELECT id FROM jobs WHERE project_id = {project} AND state = 'success'
ORDER BY number DESC
LIMIT 1`,
{ project: project.id }
);
- if (lastGood) protectedRuns.add(lastGood.id);
+ if (lastGood) protectedJobs.add(lastGood.id);
- // With no age rule, anything outside the run window goes; otherwise
+ // With no age rule, anything outside the job window goes; otherwise
// only what is also older than the cutoff.
const cutoff = keepDays > 0 ? now - keepDays * DAY : now;
const candidates = await db.all(
- `SELECT a.id, a.storage_key, a.size, a.run_id
+ `SELECT a.id, a.storage_key, a.size, a.job_id
FROM artifacts a
- JOIN runs r ON r.id = a.run_id
- WHERE r.project_id = {project}
+ JOIN jobs j ON j.id = a.job_id
+ WHERE j.project_id = {project}
AND a.expires_at IS NULL
- AND r.created_at < {cutoff}
- AND r.finished_at IS NOT NULL
- ORDER BY r.created_at
+ AND j.created_at < {cutoff}
+ AND j.finished_at IS NOT NULL
+ ORDER BY j.created_at
LIMIT {batch}`,
{ project: project.id, cutoff, batch }
);
@@ -122,7 +122,7 @@ export function createRetention({ cfg, db, storage, logs, logger = {} }) {
let deleted = 0;
let bytes = 0;
for (const row of candidates) {
- if (protectedRuns.has(row.run_id)) continue;
+ if (protectedJobs.has(row.job_id)) continue;
if (!(await deleteObject(row.storage_key, 'artifact'))) continue;
await db.run('DELETE FROM artifacts WHERE id = {id}', { id: row.id });
deleted += 1;
@@ -138,14 +138,14 @@ export function createRetention({ cfg, db, storage, logs, logger = {} }) {
const cutoff = now - keepDays * DAY;
const rows = await db.all(
- `SELECT j.id, j.run_id, j.log_key, j.log_size
- FROM jobs j
- JOIN runs r ON r.id = j.run_id
- WHERE r.project_id = {project}
- AND j.log_expired_at IS NULL
- AND j.finished_at IS NOT NULL
- AND j.finished_at < {cutoff}
- ORDER BY j.finished_at
+ `SELECT t.id, t.job_id, t.log_key, t.log_size
+ FROM tasks t
+ JOIN jobs j ON j.id = t.job_id
+ WHERE j.project_id = {project}
+ AND t.log_expired_at IS NULL
+ AND t.finished_at IS NOT NULL
+ AND t.finished_at < {cutoff}
+ ORDER BY t.finished_at
LIMIT {batch}`,
{ project: project.id, cutoff, batch }
);
@@ -155,14 +155,14 @@ export function createRetention({ cfg, db, storage, logs, logger = {} }) {
for (const row of rows) {
if (!(await deleteObject(row.log_key, 'log'))) continue;
- // A job that never finished archiving still has a spool file, and a
- // job with nothing to say has neither. Both are fine to ask about.
- await logs.remove(row.run_id, row.id).catch((e) => {
+ // A task that never finished archiving still has a spool file, and a
+ // task with nothing to say has neither. Both are fine to ask about.
+ await logs.remove(row.job_id, row.id).catch((e) => {
logger.warn?.(`retention: could not remove spooled log for ${row.id}: ${e.message}`);
});
await db.run(
- 'UPDATE jobs SET log_key = NULL, log_expired_at = {now} WHERE id = {id}',
+ 'UPDATE tasks SET log_key = NULL, log_expired_at = {now} WHERE id = {id}',
{ id: row.id, now }
);
deleted += 1;
@@ -184,7 +184,7 @@ export function createRetention({ cfg, db, storage, logs, logger = {} }) {
total.artifactBytes += expired.bytes;
const projects = await db.all(
- `SELECT id, artifact_keep_runs, artifact_keep_days, log_keep_days
+ `SELECT id, artifact_keep_jobs, artifact_keep_days, log_keep_days
FROM projects`,
{}
);
diff --git a/src/conductor/routes/artifacts.js b/src/conductor/routes/artifacts.js
@@ -1,18 +1,18 @@
// src/conductor/routes/artifacts.js - artifact downloads
//
-// The one read endpoint that is not the interface: the job page links here,
+// The one read endpoint that is not the interface: the task page links here,
// so a browser can download build output without the conductor proxying the
// bytes when the store can presign them.
//
-// Visibility matches the interface: a public run is downloadable by anyone,
-// a private one by its owner or an administrator. The project, run and job
+// Visibility matches the interface: a public job is downloadable by anyone,
+// a private one by its owner or an administrator. The project, job and task
// in the path must all agree with the artifact, so a guessed identifier
// cannot reach anything.
import { StorageNotFound } from '../../lib/storage/index.js';
export default async function artifactRoutes(fastify, { db, storage, auth = null }) {
- async function viewer(req) {
+ async function currentUser(req) {
if (!auth) return null;
try {
return await auth.identify(req);
@@ -25,28 +25,28 @@ export default async function artifactRoutes(fastify, { db, storage, auth = null
if (user && user.role === 'admin') return { sql: '1 = 1', params: {} };
if (user) {
return {
- sql: "(r.visibility = 'public' OR p.owner_id = {viewerId})",
+ sql: "(j.visibility = 'public' OR p.owner_id = {viewerId})",
params: { viewerId: user.id },
};
}
- return { sql: "r.visibility = 'public'", params: {} };
+ return { sql: "j.visibility = 'public'", params: {} };
}
- fastify.get('/projects/:project/runs/:run/jobs/:job/artifacts/:artifact', async (req, reply) => {
- const visible = scope(await viewer(req));
+ fastify.get('/projects/:project/jobs/:job/tasks/:task/artifacts/:artifact', async (req, reply) => {
+ const visible = scope(await currentUser(req));
const artifact = await db.get(
`SELECT a.id, a.path, a.storage_key, a.size
FROM artifacts a
+ JOIN tasks t ON t.id = a.task_id
JOIN jobs j ON j.id = a.job_id
- JOIN runs r ON r.id = a.run_id
- JOIN projects p ON p.id = r.project_id
+ JOIN projects p ON p.id = j.project_id
WHERE a.id = {id}
- AND a.job_id = {job} AND a.run_id = {run} AND r.project_id = {project}
+ AND a.task_id = {task} AND a.job_id = {job} AND j.project_id = {project}
AND ${visible.sql}`,
{
id: req.params.artifact,
+ task: req.params.task,
job: req.params.job,
- run: req.params.run,
project: req.params.project,
...visible.params,
}
diff --git a/src/conductor/routes/jobs.js b/src/conductor/routes/jobs.js
@@ -0,0 +1,247 @@
+// src/conductor/routes/jobs.js - creating jobs, and reading them back
+//
+// A job is a pipeline run: compiling the repository's .conductor.yml at one
+// commit produces the tasks that a worker will later claim. There are two
+// ways in, differing only in the shape of what they accept.
+//
+// POST /projects/:project/trigger takes a push payload. It accepts the
+// generic one produced by hooks/post-receive and the push events of
+// GitHub, Gitea and GitLab, since they all carry the same three facts.
+//
+// POST /projects/:project/jobs takes those facts directly, for anything
+// driving the conductor deliberately rather than forwarding a hook.
+//
+// Nothing in either payload is trusted beyond which commit to look at: the
+// pipeline is always read from the repository at that commit.
+//
+// A project with a trigger secret requires a valid signature. A project
+// without one accepts unsigned requests, which is convenient on a private
+// network and a bad idea anywhere else, so it is logged.
+
+import crypto from 'node:crypto';
+import { SHA_PATTERN } from '../../lib/git.js';
+import { PipelineError } from '../../lib/pipeline/index.js';
+import { canManageProject } from '../../lib/projects.js';
+
+const ZERO_SHA = '0'.repeat(40);
+
+export default async function jobRoutes(fastify, { auth, db, projects, scheduler, logger }) {
+ // HMAC is computed over the exact bytes received, so the raw body has to
+ // survive JSON parsing.
+ fastify.addContentTypeParser('application/json', { parseAs: 'buffer' }, (req, body, done) => {
+ req.rawBody = body;
+ if (body.length === 0) return done(null, {});
+ try {
+ done(null, JSON.parse(body.toString('utf8')));
+ } catch (e) {
+ e.statusCode = 400;
+ done(e);
+ }
+ });
+
+ // Shared by both creation routes: resolve the project, check the secret.
+ // Returns null once it has already answered.
+ async function authorizedProject(req, reply) {
+ const project = await projects.get(req.params.project);
+ if (!project) {
+ reply.code(404).send({ error: 'unknown project' });
+ return null;
+ }
+ if (project.enabled !== 1) {
+ reply.code(409).send({ error: 'project is disabled' });
+ return null;
+ }
+
+ const secret = projects.triggerSecret(project);
+ if (secret) {
+ if (!verifySignature(req, secret)) {
+ reply.code(401).send({ error: 'invalid or missing signature' });
+ return null;
+ }
+ } else {
+ logger.warn?.(`project ${project.id} accepted an unsigned request; set a trigger secret`);
+ }
+ return project;
+ }
+
+ // Compiles the pipeline and records the job, or answers with the reason
+ // it could not.
+ async function create(reply, req, project, { ref, base, sha, actor, trigger }) {
+ try {
+ const { jobId, taskCount } = await scheduler.createJob(project, {
+ ref,
+ baseSha: base,
+ headSha: sha,
+ trigger,
+ actor,
+ });
+ return reply.send({ status: 'created', job_id: jobId, tasks: taskCount });
+ } catch (e) {
+ if (e instanceof PipelineError) {
+ // A broken pipeline is the caller's problem, not a server fault.
+ return reply.code(422).send({ error: 'invalid pipeline', detail: e.message, problems: e.errors });
+ }
+ req.log.error({ err: e }, `could not create a job for ${project.id}`);
+ return reply.code(500).send({ error: 'could not create job', detail: String(e.message ?? e) });
+ }
+ }
+
+ fastify.post('/projects/:project/trigger', async (req, reply) => {
+ const project = await authorizedProject(req, reply);
+ if (!project) return reply;
+
+ const push = normalizePush(req.body);
+ if (!push) {
+ return reply.code(400).send({
+ error: 'could not read a commit from the payload; expected sha, or a GitHub, Gitea or GitLab push event',
+ });
+ }
+ // A branch deletion is not a thing to build, and saying so is more
+ // useful than refusing it.
+ if (push.sha === ZERO_SHA) {
+ return reply.send({ status: 'ignored', reason: 'branch deletion' });
+ }
+ if (!SHA_PATTERN.test(push.sha)) {
+ return reply.code(400).send({ error: `not a full commit id: ${JSON.stringify(push.sha)}` });
+ }
+
+ return create(reply, req, project, { ...push, trigger: 'push' });
+ });
+
+ // The deliberate form. Same authentication, no payload archaeology: a
+ // caller that means to start a build says so, and a missing or malformed
+ // commit is an error rather than something to infer.
+ fastify.post('/projects/:project/jobs', async (req, reply) => {
+ const project = await authorizedProject(req, reply);
+ if (!project) return reply;
+
+ const body = req.body ?? {};
+ const sha = typeof body.sha === 'string' ? body.sha : null;
+ if (!sha) return reply.code(400).send({ error: 'sha is required' });
+ if (!SHA_PATTERN.test(sha)) {
+ return reply.code(400).send({ error: `not a full commit id: ${JSON.stringify(sha)}` });
+ }
+
+ return create(reply, req, project, {
+ sha,
+ base: usableSha(body.base),
+ ref: typeof body.ref === 'string' ? body.ref : null,
+ actor: typeof body.actor === 'string' ? body.actor : null,
+ trigger: typeof body.trigger === 'string' ? body.trigger.slice(0, 32) : 'api',
+ });
+ });
+
+ // Read one back. A caller that can start a job should be able to see how
+ // it went, and scraping the interface for that is no kind of answer.
+ //
+ // Three ways to be allowed: the job is public, the caller holds the
+ // project's trigger secret, or the caller is signed in and may manage the
+ // project. Refusal is a 404 either way, so this cannot be used to probe
+ // which jobs exist.
+ fastify.get('/projects/:project/jobs/:job', async (req, reply) => {
+ const project = await projects.get(req.params.project);
+ if (!project) return reply.code(404).send({ error: 'unknown job' });
+
+ const job = await db.get(
+ `SELECT id, project_id, number, ref, head_sha, trigger_type, actor, title,
+ state, visibility, error, created_at, started_at, finished_at
+ FROM jobs
+ WHERE id = {id} AND project_id = {project}`,
+ { id: req.params.job, project: project.id }
+ );
+ if (!job) return reply.code(404).send({ error: 'unknown job' });
+
+ if (job.visibility !== 'public' && !(await mayRead(req, project))) {
+ return reply.code(404).send({ error: 'unknown job' });
+ }
+
+ const tasks = await db.all(
+ `SELECT id, name, state, arch, attempt, exit_code, error,
+ worker_name, created_at, started_at, finished_at
+ FROM tasks
+ WHERE job_id = {job}
+ ORDER BY created_at, name`,
+ { job: job.id }
+ );
+
+ return reply.send({ job, tasks });
+ });
+
+ async function mayRead(req, project) {
+ const secret = projects.triggerSecret(project);
+ if (secret && verifySignature(req, secret)) return true;
+
+ const user = await auth.identify(req).catch(() => null);
+ return canManageProject(user, project);
+ }
+}
+
+function verifySignature(req, secret) {
+ const raw = req.rawBody ?? Buffer.alloc(0);
+
+ // GitHub and Gitea: sha256=<hex> over the body.
+ const hubSignature = req.headers['x-hub-signature-256'];
+ if (typeof hubSignature === 'string' && hubSignature.length > 0) {
+ const expected = `sha256=${crypto.createHmac('sha256', secret).update(raw).digest('hex')}`;
+ return timingSafeEqual(expected, hubSignature);
+ }
+
+ // GitLab: the secret itself, compared rather than signed.
+ const gitlabToken = req.headers['x-gitlab-token'];
+ if (typeof gitlabToken === 'string' && gitlabToken.length > 0) {
+ return timingSafeEqual(secret, gitlabToken);
+ }
+
+ return false;
+}
+
+function timingSafeEqual(a, b) {
+ const left = Buffer.from(String(a), 'utf8');
+ const right = Buffer.from(String(b), 'utf8');
+ if (left.length !== right.length) return false;
+ return crypto.timingSafeEqual(left, right);
+}
+
+// Reduces the supported payload shapes to { sha, base, ref, actor }.
+function normalizePush(body) {
+ if (!body || typeof body !== 'object') return null;
+
+ // hooks/post-receive, and anything else driving the API directly.
+ if (typeof body.sha === 'string') {
+ return {
+ sha: body.sha,
+ base: usableSha(body.base),
+ ref: typeof body.ref === 'string' ? body.ref : null,
+ actor: typeof body.actor === 'string' ? body.actor : null,
+ };
+ }
+
+ // GitLab sends both checkout_sha and after; checkout_sha is the one that
+ // refers to the tip of the pushed branch.
+ if (typeof body.checkout_sha === 'string') {
+ return {
+ sha: body.checkout_sha,
+ base: usableSha(body.before),
+ ref: typeof body.ref === 'string' ? body.ref : null,
+ actor: body.user_username ?? body.user_name ?? null,
+ };
+ }
+
+ // GitHub and Gitea.
+ if (typeof body.after === 'string') {
+ return {
+ sha: body.after,
+ base: usableSha(body.before),
+ ref: typeof body.ref === 'string' ? body.ref : null,
+ actor: body.pusher?.name ?? body.pusher?.login ?? body.sender?.login ?? null,
+ };
+ }
+
+ return null;
+}
+
+// A new branch reports an all zero parent, which is not a commit.
+function usableSha(value) {
+ if (typeof value !== 'string' || value === ZERO_SHA || !SHA_PATTERN.test(value)) return null;
+ return value;
+}
diff --git a/src/conductor/routes/tasks.js b/src/conductor/routes/tasks.js
@@ -0,0 +1,307 @@
+// src/conductor/routes/tasks.js - the API workers poll and report to
+//
+// Every route here requires a worker token. A worker may only touch a task it
+// currently holds, which is checked against worker_token_id rather than
+// trusting the task id it sends.
+//
+// A worker is handed a context and a script and nothing else. It is never
+// told which project or job a task belongs to, and the task id it receives
+// carries no structure it could read that out of. The identifiers a build
+// legitimately wants are in the task's environment, where the worker treats
+// them as opaque strings on their way into the container.
+
+import { pipeline as streamPipeline } from 'node:stream/promises';
+import { LogOffsetError } from '../../lib/log.js';
+import { hashingTransform } from '../../lib/stream.js';
+import { sanitizeRelativePath, keys as storageKeys } from '../../lib/storage/index.js';
+import { maskBuffer } from '../../lib/variables.js';
+import { newArtifactId } from '../../lib/ids.js';
+
+// Log chunks are small and frequent; artifacts are streamed, so this only
+// bounds a single log append.
+const LOG_CHUNK_LIMIT = 1024 * 1024;
+
+export default async function taskRoutes(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.
+ fastify.addContentTypeParser('application/octet-stream', (req, payload, done) => done(null, payload));
+
+ async function requireWorker(req, reply) {
+ const header = req.headers.authorization || '';
+ const presented = header.startsWith('Bearer ') ? header.slice(7).trim() : '';
+ const token = await workerTokens.verify(presented, { ip: req.ip });
+ if (!token) {
+ reply.code(401).send({ error: 'invalid or missing worker token' });
+ return;
+ }
+ req.worker = token;
+ }
+
+ // Confirms the task exists and is held by the calling worker.
+ async function heldTask(req, reply) {
+ const task = await db.get(
+ `SELECT t.id, t.job_id, t.name, t.state, t.worker_token_id, t.log_size, t.spec,
+ j.project_id, j.head_sha
+ FROM tasks t JOIN jobs j ON j.id = t.job_id
+ WHERE t.id = {id}`,
+ { id: req.params.task }
+ );
+ if (!task) {
+ reply.code(404).send({ error: 'unknown task' });
+ return null;
+ }
+ if (task.worker_token_id !== req.worker.id) {
+ reply.code(403).send({ error: 'task is not held by this worker' });
+ return null;
+ }
+ return task;
+ }
+
+ fastify.addHook('preHandler', requireWorker);
+
+ // Ask for work. The worker describes what it can run; the scheduler
+ // decides. Returns 204 when there is nothing to do.
+ fastify.post('/tasks/claim', async (req, reply) => {
+ const body = req.body ?? {};
+ const arches = splitList(body.arches);
+ const features = splitList(body.features);
+ const name = typeof body.name === 'string' ? body.name.slice(0, 255) : req.worker.name;
+
+ const task = await scheduler.claim({
+ arches,
+ features,
+ tokenId: req.worker.id,
+ workerName: name,
+ // A worker registered by a user only ever sees that user's projects.
+ ownerId: req.worker.owner_id ?? null,
+ });
+ if (!task) return reply.code(204).send();
+
+ const base = cfg.server.public_url.replace(/\/+$/, '');
+ const at = (suffix) => `${base}/api/v1/tasks/${task.id}/${suffix}`;
+
+ return reply.send({
+ task: {
+ id: task.id,
+ name: task.name,
+ arch: task.arch,
+ image: task.image,
+ // The worker maps these onto its own local mounts, environment and
+ // privileges. Without them no feature is applied at all.
+ requires: task.requires,
+ script: task.script,
+ env: task.env,
+ services: task.services,
+ artifacts: task.artifacts,
+ // Where the tree is unpacked and the script runs. Resolved when
+ // the job was created, so it does not shift under a retry.
+ workdir: task.workdir,
+ timeout: task.timeout,
+ attempt: task.attempt,
+ sha: task.sha,
+ ref: task.ref,
+ // Values the worker must redact from the log stream. Populated
+ // once project variables exist; masking also happens on ingest.
+ masked: task.masked ?? [],
+ // Derived from the server's own timeout so a worker never has to
+ // guess how often it is expected to check in.
+ heartbeat_interval: Math.max(5, Math.floor(cfg.scheduler.heartbeat_timeout / 4)),
+ endpoints: {
+ // The tree comes from the conductor, so a worker never needs
+ // credentials for the repository and cannot reach any commit
+ // other than the one it was given work for.
+ source: at('source.tar.gz'),
+ log: at('log'),
+ artifact: at('artifacts'),
+ heartbeat: at('heartbeat'),
+ done: at('complete'),
+ },
+ },
+ });
+ });
+
+ // The tree at the task's commit, as a gzipped tar.
+ fastify.get('/tasks/:task/source.tar.gz', async (req, reply) => {
+ const task = await heldTask(req, reply);
+ if (!task) return reply;
+
+ const project = await projects.get(task.project_id);
+ if (!project) return reply.code(404).send({ error: 'unknown project' });
+
+ // The tree rooted at the archive, with no wrapping directory: a
+ // worker extracts it at whatever path the task is to run in, and
+ // docker creates that path on the way.
+ reply.header('content-type', 'application/gzip');
+ reply.header('content-disposition', `attachment; filename="${task.head_sha.slice(0, 12)}.tar.gz"`);
+ return reply.send(git.archiveStream(project, task.head_sha));
+ });
+
+ // Append to the live log. X-Log-Offset makes a retry after a dropped
+ // connection safe.
+ fastify.post('/tasks/:task/log', { bodyLimit: LOG_CHUNK_LIMIT }, async (req, reply) => {
+ const task = await heldTask(req, reply);
+ if (!task) return reply;
+
+ const chunks = [];
+ let total = 0;
+ for await (const chunk of req.body) {
+ total += chunk.length;
+ if (total > LOG_CHUNK_LIMIT) return reply.code(413).send({ error: 'log chunk too large' });
+ chunks.push(chunk);
+ }
+
+ const rawOffset = req.headers['x-log-offset'];
+ const offset = rawOffset === undefined ? undefined : Number(rawOffset);
+ if (offset !== undefined && !Number.isInteger(offset)) {
+ return reply.code(400).send({ error: 'x-log-offset must be an integer' });
+ }
+
+ // The worker masks before sending; this repeats it on ingest so a
+ // worker that does not still cannot write a secret to disk. It works
+ // within a chunk only, which is why the worker holds back a boundary.
+ let payload = Buffer.concat(chunks);
+ if (variables) {
+ try {
+ payload = maskBuffer(payload, await variables.maskedValues(task.project_id));
+ } catch (e) {
+ req.log.warn({ err: e }, `could not mask log output for ${task.id}`);
+ }
+ }
+
+ try {
+ const result = await logs.append(task.job_id, task.id, payload, offset);
+ await db.run('UPDATE tasks SET log_size = {size}, heartbeat_at = {now} WHERE id = {id}',
+ { id: task.id, size: result.size, now: Date.now() });
+ return reply.send(result);
+ } catch (e) {
+ if (e instanceof LogOffsetError) {
+ // Tell the worker where to resume from.
+ return reply.code(409).send({ error: e.message, expected_offset: e.expected });
+ }
+ throw e;
+ }
+ });
+
+ // artifacts.expire in the pipeline, as an absolute time. Returns null
+ // when the task said nothing, leaving the artifact to the project policy.
+ function artifactExpiry(specJson, now) {
+ if (!specJson) return null;
+ try {
+ const spec = typeof specJson === 'string' ? JSON.parse(specJson) : specJson;
+ const seconds = spec?.artifacts?.expire;
+ if (!Number.isFinite(seconds) || seconds <= 0) return null;
+ return now + Math.round(seconds * 1000);
+ } catch {
+ // A spec that will not parse is a problem, but not this request's
+ // problem: the artifact is already stored.
+ return null;
+ }
+ }
+
+ // Upload one artifact. The path is worker supplied, so it is sanitized
+ // before it can influence a storage key.
+ fastify.post('/tasks/:task/artifacts', async (req, reply) => {
+ const task = await heldTask(req, reply);
+ if (!task) return reply;
+
+ const relPath = sanitizeRelativePath(req.headers['x-artifact-path']);
+ if (!relPath) return reply.code(400).send({ error: 'x-artifact-path is missing or unusable' });
+
+ const declared = Number(req.headers['content-length']);
+ if (!Number.isInteger(declared) || declared < 0) {
+ return reply.code(411).send({ error: 'content-length is required for artifact uploads' });
+ }
+
+ // Hash on the way past, so the digest costs nothing extra even when the
+ // backend cannot report one.
+ const hasher = hashingTransform();
+ const key = storageKeys.artifact(task.job_id, task.id, relPath);
+
+ let stored;
+ try {
+ [, stored] = await Promise.all([
+ streamPipeline(req.body, hasher),
+ storage.put(key, hasher, { size: declared }),
+ ]);
+ } catch (e) {
+ // A body shorter or longer than content-length is the worker's
+ // mistake, not a server fault.
+ if (/size mismatch/.test(e.message)) {
+ return reply.code(400).send({ error: e.message });
+ }
+ throw e;
+ }
+
+ const digest = hasher.digest();
+ const now = Date.now();
+
+ await db.run(
+ `INSERT INTO artifacts (id, task_id, job_id, path, storage_key, size, sha256, created_at, expires_at)
+ VALUES ({id}, {task}, {job}, {path}, {key}, {size}, {sha}, {now}, {expires})`,
+ {
+ id: newArtifactId(),
+ task: task.id,
+ job: task.job_id,
+ path: relPath,
+ key,
+ size: stored.size ?? declared,
+ sha: digest,
+ now,
+ // An explicit deadline from the pipeline. Absent, the project's
+ // retention policy decides, which is the usual case.
+ expires: artifactExpiry(task.spec, now),
+ }
+ );
+
+ return reply.send({ path: relPath, size: stored.size ?? declared, sha256: digest });
+ });
+
+ // Keeps the task alive, and is how a worker learns it should stop.
+ fastify.post('/tasks/:task/heartbeat', async (req, reply) => {
+ const result = await scheduler.heartbeat(req.params.task, req.worker.id);
+ if (!result.known) return reply.code(404).send({ error: 'unknown task for this worker' });
+ return reply.send({ cancelled: result.cancelled, state: result.state });
+ });
+
+ // Report the outcome. The log is copied to storage here, whatever the
+ // result, so a failed task keeps its output.
+ fastify.post('/tasks/:task/complete', async (req, reply) => {
+ const task = await heldTask(req, reply);
+ if (!task) return reply;
+
+ const body = req.body ?? {};
+ const success = body.success === true;
+ const exitCode = Number.isInteger(body.exit_code) ? body.exit_code : null;
+ const error = typeof body.error === 'string' ? body.error.slice(0, 4096) : null;
+
+ const outcome = await scheduler.complete(task.id, {
+ success,
+ exitCode,
+ error,
+ tokenId: req.worker.id,
+ });
+ if (!outcome.ok) return reply.code(409).send({ error: outcome.reason });
+
+ // A retry keeps writing to the same log, so only archive it once the
+ // task has actually stopped.
+ if (outcome.state !== 'queued') {
+ try {
+ await scheduler.finalizeLog(task.job_id, task.id);
+ } catch (e) {
+ req.log.error({ err: e }, `failed to archive log for ${task.id}`);
+ }
+ }
+
+ return reply.send({
+ state: outcome.state,
+ retry: outcome.retry === true,
+ skipped: outcome.skipped ?? [],
+ job_state: outcome.jobState ?? null,
+ });
+ });
+}
+
+function splitList(value) {
+ const items = Array.isArray(value) ? value : typeof value === 'string' ? value.split(',') : [];
+ return items.map((s) => String(s).trim()).filter(Boolean).slice(0, 64);
+}
diff --git a/src/conductor/scheduler.js b/src/conductor/scheduler.js
@@ -1,144 +1,182 @@
-// src/conductor/scheduler.js - run creation, dispatch and completion
+// src/conductor/scheduler.js - job creation, dispatch and completion
//
-// Scheduling reads the dependency edges in job_deps directly. Two defects in
+// A trigger creates a job. Compiling the repository's pipeline turns that job
+// into tasks, and a task is what a worker runs.
+//
+// Scheduling reads the dependency edges in task_deps directly. Two defects in
// the prototype came from not doing that:
//
-// Dispatch was serialized. A single integer level per job was compared
-// against the lowest queued level, so only one job could run at a time
-// even when the graph allowed the whole width of it. Here any job whose
+// Dispatch was serialized. A single integer level per task was compared
+// against the lowest queued level, so only one task could run at a time
+// even when the graph allowed the whole width of it. Here any task whose
// dependencies are all satisfied is eligible, and several workers can
-// claim different jobs concurrently.
+// claim different tasks concurrently.
//
// Failure skipped only direct dependents. Transitive dependents stayed
// queued and were later dispatched with their inputs missing. Here the
// whole reachable set is skipped.
//
// Claiming is a conditional UPDATE guarded on the previous state, so two
-// workers polling at the same instant cannot both win the same job.
+// workers polling at the same instant cannot both win the same task.
import { compilePipeline, PipelineError, transitiveDependents, DEFAULT_WORKDIR } from '../lib/pipeline/index.js';
import { inClause } from '../lib/db/query.js';
-import { newRunId, jobId as makeJobId } from '../lib/ids.js';
+import { newJobId, newTaskId } from '../lib/ids.js';
import { keys as storageKeys } from '../lib/storage/index.js';
-export const RUN_STATES = ['pending', 'running', 'success', 'failed', 'cancelled'];
-export const JOB_STATES = ['queued', 'running', 'success', 'failed', 'skipped', 'cancelled'];
+export const JOB_STATES = ['pending', 'running', 'success', 'failed', 'cancelled'];
+export const TASK_STATES = ['queued', 'running', 'success', 'failed', 'skipped', 'cancelled'];
// States that let a dependent proceed.
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 job is eligible when it is queued, its run is live, and no dependency
+ // 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.
const ELIGIBLE = `
- SELECT j.id, j.run_id, j.name, j.arch, j.image, j.requires, j.spec,
- j.timeout, j.attempt, j.max_attempts,
- r.head_sha, r.project_id, r.ref, r.number
- FROM jobs j
- JOIN runs r ON r.id = j.run_id
- JOIN projects p ON p.id = r.project_id
- WHERE j.state = 'queued'
- AND r.state = 'running'
+ SELECT t.id, t.job_id, t.name, t.arch, t.image, t.requires, t.spec,
+ t.timeout, t.attempt, t.max_attempts,
+ j.head_sha, j.project_id, j.ref, j.number
+ FROM tasks t
+ JOIN jobs j ON j.id = t.job_id
+ JOIN projects p ON p.id = j.project_id
+ WHERE t.state = 'queued'
+ AND j.state = 'running'
AND NOT EXISTS (
SELECT 1
- FROM job_deps d
- JOIN jobs dj ON dj.id = d.depends_on_id
- WHERE d.job_id = j.id
- AND dj.state <> 'success'
- AND NOT (dj.state = 'failed' AND dj.allow_failure = 1)
+ FROM task_deps d
+ JOIN tasks dt ON dt.id = d.depends_on_id
+ WHERE d.task_id = t.id
+ AND dt.state <> 'success'
+ AND NOT (dt.state = 'failed' AND dt.allow_failure = 1)
)
`;
- async function loadGraph(tx, runId) {
- const jobs = await tx.all(
- 'SELECT id, name, state, allow_failure FROM jobs WHERE run_id = {run}',
- { run: runId }
+ async function loadGraph(tx, jobId) {
+ const tasks = await tx.all(
+ 'SELECT id, name, state, allow_failure FROM tasks WHERE job_id = {job}',
+ { job: jobId }
);
const deps = await tx.all(
- `SELECT d.job_id, d.depends_on_id
- FROM job_deps d
- JOIN jobs j ON j.id = d.job_id
- WHERE j.run_id = {run}`,
- { run: runId }
+ `SELECT d.task_id, d.depends_on_id
+ FROM task_deps d
+ JOIN tasks t ON t.id = d.task_id
+ WHERE t.job_id = {job}`,
+ { job: jobId }
);
- const needs = new Map(jobs.map((j) => [j.id, []]));
- for (const d of deps) needs.get(d.job_id)?.push(d.depends_on_id);
- return jobs.map((j) => ({ ...j, needs: needs.get(j.id) ?? [] }));
+ 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) ?? [] }));
+ }
+
+ // Reads the pipeline file at a commit.
+ //
+ // A project that has not been told otherwise accepts either conventional
+ // spelling, so a repository may use .conductor.yml or .conductor.yaml
+ // without anyone configuring it. Only the conventional names get that
+ // fallback: a project pointed at ci/build.yml means that file and nothing
+ // else. Finding both is an error rather than a coin toss, since two
+ // pipelines in one repository will disagree sooner or later.
+ async function readPipelineFile(project, sha) {
+ const wanted = project.config_path;
+ const alternate = CONVENTIONAL_CONFIG_PATHS.find((p) => p !== wanted);
+ const candidates = CONVENTIONAL_CONFIG_PATHS.includes(wanted) ? [wanted, alternate] : [wanted];
+
+ const found = [];
+ for (const path of candidates) {
+ const text = await git.readFile(project, sha, path);
+ if (text !== null) found.push({ path, text });
+ }
+
+ if (found.length > 1) {
+ throw new PipelineError(
+ [{
+ path: '',
+ message: `${found.map((f) => f.path).join(' and ')} both exist at ` +
+ `${sha.slice(0, 12)}; keep one`,
+ }],
+ wanted
+ );
+ }
+ return found[0] ?? null;
}
- // Marks the run finished once nothing is left to do.
- async function settleRun(tx, runId) {
+ // Marks the job finished once nothing is left to do.
+ async function settleJob(tx, jobId) {
const outstanding = await tx.get(
- "SELECT COUNT(*) AS c FROM jobs WHERE run_id = {run} AND state IN ('queued', 'running')",
- { run: runId }
+ "SELECT COUNT(*) AS c FROM tasks WHERE job_id = {job} AND state IN ('queued', 'running')",
+ { job: jobId }
);
if (outstanding.c > 0) return null;
const bad = await tx.get(
- `SELECT COUNT(*) AS c FROM jobs
- WHERE run_id = {run}
+ `SELECT COUNT(*) AS c FROM tasks
+ WHERE job_id = {job}
AND (state IN ('skipped', 'cancelled')
OR (state = 'failed' AND allow_failure = 0))`,
- { run: runId }
+ { job: jobId }
);
const state = bad.c > 0 ? 'failed' : 'success';
await tx.run(
- "UPDATE runs SET state = {state}, finished_at = {now} WHERE id = {run} AND state = 'running'",
- { state, run: runId, now: Date.now() }
+ "UPDATE jobs SET state = {state}, finished_at = {now} WHERE id = {job} AND state = 'running'",
+ { state, job: jobId, now: Date.now() }
);
return state;
}
return {
- // Reads the pipeline at the pushed commit and records the run. The
+ // Reads the pipeline at the pushed commit and records the job. The
// repository is the source of truth, so nothing the trigger claims about
// the contents is trusted.
- async createRun(project, { ref, baseSha, headSha, trigger = 'push', actor = null }) {
+ async createJob(project, { ref, baseSha, headSha, trigger = 'push', actor = null }) {
await git.sync(project, { force: true });
if (!(await git.hasCommit(project, headSha))) {
throw new Error(`commit ${headSha} is not present in the mirror of ${project.id}`);
}
- const text = await git.readFile(project, headSha, project.config_path);
- if (text === null) {
+ const found = await readPipelineFile(project, headSha);
+ if (found === null) {
throw new PipelineError(
[{ path: '', message: `${project.config_path} not found at ${headSha.slice(0, 12)}` }],
project.config_path
);
}
- const pipeline = compilePipeline(text, {
- source: project.config_path,
- defaultTimeout: cfg.scheduler.default_job_timeout,
+ const pipeline = compilePipeline(found.text, {
+ source: found.path,
+ defaultTimeout: cfg.scheduler.default_task_timeout,
defaultAttempts: 1,
- // Decides which jobs this push runs at all, so a publish step
- // restricted to main is absent from a branch run rather than
+ // Decides which tasks this push runs at all, so a publish step
+ // restricted to main is absent from a branch's job rather than
// recorded against it.
ref: ref ?? '',
});
const info = await git.commitInfo(project, headSha).catch(() => ({ subject: null }));
- const runId = newRunId();
+ const jobId = newJobId();
const now = Date.now();
// The repository may declare its own visibility; otherwise the
- // project's setting stands. Stored per run, because the answer can
+ // project's setting stands. Stored per job, because the answer can
// legitimately change from one commit to the next.
const visibility = pipeline.visibility ?? project.visibility ?? 'private';
- // Same shape of answer for where the tree is unpacked inside a job
+ // Same shape of answer for where the tree is unpacked inside a task
// container: the repository knows what its build expects, the
// project can set a house default, and failing both there is one.
const workdir = pipeline.workdir ?? project.workdir ?? DEFAULT_WORKDIR;
await db.transaction(async (tx) => {
- const number = await projects.nextRunNumber(tx, project.id);
- const empty = pipeline.jobs.length === 0;
+ const number = await projects.nextJobNumber(tx, project.id);
+ const empty = pipeline.tasks.length === 0;
await tx.run(
- `INSERT INTO runs
+ `INSERT INTO jobs
(id, project_id, number, ref, base_sha, head_sha, trigger_type, actor,
title, state, pipeline, visibility, created_at, started_at, finished_at)
VALUES
@@ -146,7 +184,7 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
{title}, {state}, {pipeline}, {visibility}, {now}, {now}, {finished})`,
{
visibility,
- id: runId,
+ id: jobId,
project: project.id,
number,
ref: ref ?? null,
@@ -156,67 +194,70 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
actor,
title: info.subject ?? null,
state: empty ? 'success' : 'running',
- pipeline: JSON.stringify({ version: pipeline.version, jobs: pipeline.jobs }),
+ pipeline: JSON.stringify({ version: pipeline.version, tasks: pipeline.tasks }),
now,
finished: empty ? now : null,
}
);
- const idByName = new Map(pipeline.jobs.map((j) => [j.name, makeJobId(runId, j.name)]));
+ // Task ids are minted here rather than derived from the name, so
+ // the dependency edges below are the only thing that maps a name
+ // onto an id.
+ const idByName = new Map(pipeline.tasks.map((t) => [t.name, newTaskId()]));
- for (const job of pipeline.jobs) {
+ for (const task of pipeline.tasks) {
await tx.run(
- `INSERT INTO jobs
- (id, run_id, name, base_name, arch, image, requires, spec, state,
+ `INSERT INTO tasks
+ (id, job_id, name, base_name, arch, image, requires, spec, state,
allow_failure, attempt, max_attempts, timeout, log_size, created_at)
VALUES
- ({id}, {run}, {name}, {base}, {arch}, {image}, {requires}, {spec}, 'queued',
+ ({id}, {job}, {name}, {base}, {arch}, {image}, {requires}, {spec}, 'queued',
{allow}, 0, {attempts}, {timeout}, 0, {now})`,
{
- id: idByName.get(job.name),
- run: runId,
- name: job.name,
- base: job.baseName,
- arch: job.arch,
- image: job.image,
- requires: JSON.stringify(job.requires),
+ id: idByName.get(task.name),
+ job: jobId,
+ name: task.name,
+ base: task.baseName,
+ arch: task.arch,
+ image: task.image,
+ requires: JSON.stringify(task.requires),
spec: JSON.stringify({
- script: job.script,
- env: job.env,
- services: job.services,
- artifacts: job.artifacts,
- matrix: job.matrix,
- needs: job.needs,
- depth: job.depth,
- // Recorded per job rather than looked up when a worker
- // polls, so a run stays reproducible after the project
+ script: task.script,
+ env: task.env,
+ services: task.services,
+ artifacts: task.artifacts,
+ matrix: task.matrix,
+ needs: task.needs,
+ depth: task.depth,
+ // Recorded per task rather than looked up when a worker
+ // polls, so a job stays reproducible after the project
// or the pipeline changes.
workdir,
}),
- allow: job.allow_failure ? 1 : 0,
- attempts: job.max_attempts,
- timeout: job.timeout,
+ allow: task.allow_failure ? 1 : 0,
+ attempts: task.max_attempts,
+ timeout: task.timeout,
now,
}
);
}
- for (const job of pipeline.jobs) {
- for (const need of job.needs) {
+ for (const task of pipeline.tasks) {
+ for (const need of task.needs) {
await tx.run(
- 'INSERT INTO job_deps (job_id, depends_on_id) VALUES ({job}, {dep})',
- { job: idByName.get(job.name), dep: idByName.get(need) }
+ 'INSERT INTO task_deps (task_id, depends_on_id) VALUES ({task}, {dep})',
+ { task: idByName.get(task.name), dep: idByName.get(need) }
);
}
}
});
- logger.info?.(`run ${runId} created for ${project.id} with ${pipeline.jobs.length} job(s)`);
- return { runId, jobCount: pipeline.jobs.length };
+ logger.info?.(`job ${jobId} created for ${project.id} with ${pipeline.tasks.length} task(s)`);
+ return { jobId, taskCount: pipeline.tasks.length };
},
- // Hands one job to a worker. The worker advertises what it can run;
- // jobs asking for anything it lacks are passed over.
+ // Hands one task to a worker. The worker advertises what it can run;
+ // tasks asking for anything it lacks are passed over.
//
// ownerId scopes the worker to one user's projects. That is the whole
// basis of letting people contribute their own hardware: a worker
@@ -226,13 +267,13 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
const archFilter = inClause('arch', arches);
const featureSet = new Set(features);
- // Oldest run first, so a queue drains in order rather than starving
- // whichever run happens to sort last.
+ // Oldest job first, so a queue drains in order rather than starving
+ // whichever job happens to sort last.
const candidates = await db.all(
`${ELIGIBLE}
- AND (j.arch IS NULL${archFilter.empty ? '' : ` OR j.arch IN (${archFilter.sql})`})
+ AND (t.arch IS NULL${archFilter.empty ? '' : ` OR t.arch IN (${archFilter.sql})`})
${ownerId === null ? '' : 'AND p.owner_id = {owner}'}
- ORDER BY r.created_at ASC, j.created_at ASC
+ ORDER BY j.created_at ASC, t.created_at ASC
LIMIT 100`,
{ ...archFilter.params, ...(ownerId === null ? {} : { owner: ownerId }) }
);
@@ -248,7 +289,7 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
const now = Date.now();
const claimed = await db.run(
- `UPDATE jobs
+ `UPDATE tasks
SET state = 'running', attempt = attempt + 1, worker_token_id = {token},
worker_name = {worker}, claimed_at = {now}, heartbeat_at = {now},
started_at = {now}
@@ -262,7 +303,7 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
const attempt = candidate.attempt + 1;
// Project variables are resolved at dispatch and handed to the
- // worker, never written into the jobs table, so the only place a
+ // worker, never written into the tasks table, so the only place a
// secret rests is project_variables.
let injected = { env: {}, masked: [] };
if (variables) {
@@ -273,9 +314,12 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
}
}
+ // project_id and job_id are returned for the route to use, not for
+ // the worker. What the worker is handed is assembled in the route,
+ // and deliberately carries neither.
return {
id: candidate.id,
- run_id: candidate.run_id,
+ job_id: candidate.job_id,
project_id: candidate.project_id,
name: candidate.name,
arch: candidate.arch,
@@ -290,23 +334,26 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
// Values the worker must redact from the log stream.
masked: injected.masked,
// Project variables, then pipeline env, then the facts about this
- // run. Ordered so a pipeline cannot quietly redefine the last
- // group, and a variable cannot shadow an explicit job setting.
+ // task. Ordered so a pipeline cannot quietly redefine the last
+ // group, and a variable cannot shadow an explicit task setting.
+ //
+ // These name the project and job the worker is not told about,
+ // which is deliberate: they are opaque strings bound for the
+ // script, and the worker only copies them into the container.
env: {
...injected.env,
...spec.env,
CONDUCTOR_PROJECT: candidate.project_id,
- CONDUCTOR_RUN_ID: candidate.run_id,
- CONDUCTOR_RUN_NUMBER: String(candidate.number),
- CONDUCTOR_JOB: candidate.name,
- CONDUCTOR_JOB_ID: candidate.id,
+ CONDUCTOR_JOB_ID: candidate.job_id,
+ CONDUCTOR_JOB_NUMBER: String(candidate.number),
+ CONDUCTOR_TASK: candidate.name,
+ CONDUCTOR_TASK_ID: candidate.id,
CONDUCTOR_SHA: candidate.head_sha,
CONDUCTOR_REF: candidate.ref ?? '',
CONDUCTOR_ATTEMPT: String(attempt),
},
services: spec.services,
artifacts: spec.artifacts,
- // Runs created before this existed have no workdir recorded.
workdir: spec.workdir ?? DEFAULT_WORKDIR,
};
}
@@ -314,138 +361,138 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
return null;
},
- async heartbeat(jobId, tokenId) {
- const job = await db.get(
- 'SELECT id, state, worker_token_id FROM jobs WHERE id = {id}',
- { id: jobId }
+ async heartbeat(taskId, tokenId) {
+ const task = await db.get(
+ 'SELECT id, state, worker_token_id FROM tasks WHERE id = {id}',
+ { id: taskId }
);
- if (!job) return { known: false };
- if (job.worker_token_id !== tokenId) return { known: false };
+ if (!task) return { known: false };
+ if (task.worker_token_id !== tokenId) return { known: false };
- if (job.state === 'running') {
- await db.run('UPDATE jobs SET heartbeat_at = {now} WHERE id = {id}', { id: jobId, now: Date.now() });
+ if (task.state === 'running') {
+ await db.run('UPDATE tasks SET heartbeat_at = {now} WHERE id = {id}', { id: taskId, now: Date.now() });
return { known: true, cancelled: false };
}
- // The job was cancelled or reaped out from under the worker, which is
+ // The task was cancelled or reaped out from under the worker, which is
// how a worker learns to stop.
- return { known: true, cancelled: true, state: job.state };
+ return { known: true, cancelled: true, state: task.state };
},
// Records the outcome, retries if attempts remain, and otherwise skips
// everything downstream.
- async complete(jobId, { success, exitCode = null, error = null, tokenId = null }) {
+ async complete(taskId, { success, exitCode = null, error = null, tokenId = null }) {
return db.transaction(async (tx) => {
- const job = await tx.get(
- `SELECT id, run_id, name, state, attempt, max_attempts, allow_failure, worker_token_id
- FROM jobs WHERE id = {id}`,
- { id: jobId }
+ const task = await tx.get(
+ `SELECT id, job_id, name, state, attempt, max_attempts, allow_failure, worker_token_id
+ FROM tasks WHERE id = {id}`,
+ { id: taskId }
);
- if (!job) return { ok: false, reason: 'unknown job' };
- if (tokenId !== null && job.worker_token_id !== tokenId) {
- return { ok: false, reason: 'job belongs to another worker' };
+ if (!task) return { ok: false, reason: 'unknown task' };
+ if (tokenId !== null && task.worker_token_id !== tokenId) {
+ return { ok: false, reason: 'task belongs to another worker' };
}
- if (job.state !== 'running') {
- return { ok: false, reason: `job is ${job.state}, not running` };
+ if (task.state !== 'running') {
+ return { ok: false, reason: `task is ${task.state}, not running` };
}
const now = Date.now();
if (success) {
await tx.run(
- "UPDATE jobs SET state = 'success', exit_code = {code}, finished_at = {now} WHERE id = {id}",
- { id: jobId, code: exitCode, now }
+ "UPDATE tasks SET state = 'success', exit_code = {code}, finished_at = {now} WHERE id = {id}",
+ { id: taskId, code: exitCode, now }
);
- const runState = await settleRun(tx, job.run_id);
- return { ok: true, state: 'success', runState };
+ const jobState = await settleJob(tx, task.job_id);
+ return { ok: true, state: 'success', jobState };
}
// Transient failure with attempts left: back to the queue rather
- // than failing the run.
- if (job.attempt < job.max_attempts) {
+ // than failing the job.
+ if (task.attempt < task.max_attempts) {
await tx.run(
- `UPDATE jobs
+ `UPDATE tasks
SET state = 'queued', exit_code = {code}, error = {error},
worker_token_id = NULL, worker_name = NULL,
claimed_at = NULL, heartbeat_at = NULL, started_at = NULL
WHERE id = {id}`,
- { id: jobId, code: exitCode, error }
+ { id: taskId, code: exitCode, error }
);
- return { ok: true, state: 'queued', retry: true, attempt: job.attempt };
+ return { ok: true, state: 'queued', retry: true, attempt: task.attempt };
}
await tx.run(
- "UPDATE jobs SET state = 'failed', exit_code = {code}, error = {error}, finished_at = {now} WHERE id = {id}",
- { id: jobId, code: exitCode, error, now }
+ "UPDATE tasks SET state = 'failed', exit_code = {code}, error = {error}, finished_at = {now} WHERE id = {id}",
+ { id: taskId, code: exitCode, error, now }
);
let skipped = [];
- if (!job.allow_failure) {
- skipped = await skipDependents(tx, job.run_id, [jobId], now);
+ if (!task.allow_failure) {
+ skipped = await skipDependents(tx, task.job_id, [taskId], now);
}
- const runState = await settleRun(tx, job.run_id);
- return { ok: true, state: 'failed', skipped, runState };
+ const jobState = await settleJob(tx, task.job_id);
+ return { ok: true, state: 'failed', skipped, jobState };
});
},
- // Every job reachable from the failed ones, not just their immediate
+ // Every task reachable from the failed ones, not just their immediate
// dependents.
- async skipDependents(runId, fromJobIds) {
- return db.transaction((tx) => skipDependents(tx, runId, fromJobIds, Date.now()));
+ async skipDependents(jobId, fromTaskIds) {
+ return db.transaction((tx) => skipDependents(tx, jobId, fromTaskIds, Date.now()));
},
- // Requeues or fails jobs whose worker stopped reporting.
+ // Requeues or fails tasks whose worker stopped reporting.
async reap() {
const cutoff = Date.now() - cfg.scheduler.heartbeat_timeout * 1000;
const stale = await db.all(
- `SELECT id, run_id, name, attempt, max_attempts, worker_name
- FROM jobs
+ `SELECT id, job_id, name, attempt, max_attempts, worker_name
+ FROM tasks
WHERE state = 'running' AND heartbeat_at IS NOT NULL AND heartbeat_at < {cutoff}`,
{ cutoff }
);
const results = [];
- for (const job of stale) {
- const outcome = await this.complete(job.id, {
+ for (const task of stale) {
+ const outcome = await this.complete(task.id, {
success: false,
- error: `worker ${job.worker_name ?? 'unknown'} stopped reporting for more than ` +
+ error: `worker ${task.worker_name ?? 'unknown'} stopped reporting for more than ` +
`${cfg.scheduler.heartbeat_timeout}s`,
});
if (outcome.ok) {
- logger.warn?.(`reaped ${job.name} (${job.id}): ${outcome.retry ? 'requeued' : 'failed'}`);
- results.push({ id: job.id, ...outcome });
+ logger.warn?.(`reaped ${task.name} (${task.id}): ${outcome.retry ? 'requeued' : 'failed'}`);
+ results.push({ id: task.id, ...outcome });
}
}
return results;
},
- async cancelRun(runId, reason = 'cancelled by request') {
+ async cancelJob(jobId, reason = 'cancelled by request') {
return db.transaction(async (tx) => {
- const run = await tx.get('SELECT id, state FROM runs WHERE id = {id}', { id: runId });
- if (!run) return { ok: false, reason: 'unknown run' };
- if (run.state !== 'running') return { ok: false, reason: `run is ${run.state}` };
+ const job = await tx.get('SELECT id, state FROM jobs WHERE id = {id}', { id: jobId });
+ if (!job) return { ok: false, reason: 'unknown job' };
+ if (job.state !== 'running') return { ok: false, reason: `job is ${job.state}` };
const now = Date.now();
await tx.run(
- `UPDATE jobs SET state = 'cancelled', error = {reason}, finished_at = {now}
- WHERE run_id = {run} AND state IN ('queued', 'running')`,
- { run: runId, reason, now }
+ `UPDATE tasks SET state = 'cancelled', error = {reason}, finished_at = {now}
+ WHERE job_id = {job} AND state IN ('queued', 'running')`,
+ { job: jobId, reason, now }
);
await tx.run(
- "UPDATE runs SET state = 'cancelled', finished_at = {now} WHERE id = {run}",
- { run: runId, now }
+ "UPDATE jobs SET state = 'cancelled', finished_at = {now} WHERE id = {job}",
+ { job: jobId, now }
);
return { ok: true };
});
},
- // Called once a job finishes so the live log stops being the spool copy.
- async finalizeLog(runId, jobId) {
- const key = storageKeys.log(runId, jobId);
- const { size } = await logs.finalize(runId, jobId, storage, key);
+ // Called once a task finishes so the live log stops being the spool copy.
+ async finalizeLog(jobId, taskId) {
+ const key = storageKeys.log(jobId, taskId);
+ const { size } = await logs.finalize(jobId, taskId, storage, key);
if (size > 0) {
- await db.run('UPDATE jobs SET log_key = {key}, log_size = {size} WHERE id = {id}',
- { id: jobId, key, size });
+ await db.run('UPDATE tasks SET log_key = {key}, log_size = {size} WHERE id = {id}',
+ { id: taskId, key, size });
}
return { key, size };
},
@@ -453,20 +500,20 @@ export function createScheduler({ cfg, db, git, logs, storage, projects, variabl
TERMINAL,
};
- async function skipDependents(tx, runId, fromJobIds, now) {
- const graph = await loadGraph(tx, runId);
+ async function skipDependents(tx, jobId, fromTaskIds, now) {
+ const graph = await loadGraph(tx, jobId);
const reachable = transitiveDependents(
- graph.map((j) => ({ name: j.id, needs: j.needs })),
- fromJobIds
+ graph.map((t) => ({ name: t.id, needs: t.needs })),
+ fromTaskIds
);
if (reachable.size === 0) return [];
- const queued = new Set(graph.filter((j) => j.state === 'queued').map((j) => j.id));
+ const queued = new Set(graph.filter((t) => t.state === 'queued').map((t) => t.id));
const toSkip = [...reachable].filter((id) => queued.has(id));
for (const id of toSkip) {
await tx.run(
- `UPDATE jobs SET state = 'skipped', error = {error}, finished_at = {now}
+ `UPDATE tasks SET state = 'skipped', error = {error}, finished_at = {now}
WHERE id = {id} AND state = 'queued'`,
{ id, error: 'skipped because a dependency did not succeed', now }
);
diff --git a/src/conductor/ui/layout.js b/src/conductor/ui/layout.js
@@ -8,7 +8,7 @@ import { html, raw, esc } from './html.js';
export function layout({ title, user, body, active = '', localLogin = true }) {
const nav = [
- { href: '/', label: 'runs', key: 'runs' },
+ { href: '/', label: 'jobs', key: 'jobs' },
user ? { href: '/projects', label: 'projects', key: 'projects' } : null,
user ? { href: '/workers', label: 'workers', key: 'workers' } : null,
// Accounts cannot sign in while the provider owns identity, so the page
diff --git a/src/conductor/ui/pages.js b/src/conductor/ui/pages.js
@@ -2,7 +2,7 @@
//
// Each live region follows the same htmx pattern: the fragment carries its
// own polling attribute only while there is something left to happen. When
-// the run or job finishes, the replacement fragment has no trigger, so
+// the job or task finishes, the replacement fragment has no trigger, so
// polling stops by itself rather than needing to be cancelled.
import { html, raw } from './html.js';
@@ -10,126 +10,126 @@ import { badge, notice, oneTimeSecret, shortSha, ago, duration, ACTIVE_RUN_STATE
const poll = (url, seconds = 3) => raw(`hx-get="${url}" hx-trigger="every ${seconds}s" hx-swap="outerHTML"`);
-// --- runs ---
+// --- jobs ---
-export function runsPage({ runs, user, anonymous }) {
+export function jobsPage({ jobs, user, anonymous }) {
return html`
- <h2>Runs</h2>
- ${anonymous ? html`<p class="muted">Showing public runs. <a href="/login">Sign in</a> to see your own.</p>` : ''}
- ${runsTable(runs)}
+ <h2>Jobs</h2>
+ ${anonymous ? html`<p class="muted">Showing public jobs. <a href="/login">Sign in</a> to see your own.</p>` : ''}
+ ${jobsTable(jobs)}
`;
}
-export function runsTable(runs) {
- const live = runs.some((r) => ACTIVE_RUN_STATES.includes(r.state));
- return html`<div id="runs" ${live ? poll('/partials/runs', 4) : ''}>
+export function jobsTable(jobs) {
+ const live = jobs.some((r) => ACTIVE_RUN_STATES.includes(r.state));
+ return html`<div id="jobs" ${live ? poll('/partials/jobs', 4) : ''}>
<table>
<thead><tr>
- <th>run</th><th>project</th><th>ref</th><th>commit</th><th>state</th><th>started</th>
+ <th>job</th><th>project</th><th>ref</th><th>commit</th><th>state</th><th>started</th>
</tr></thead>
<tbody>
- ${runs.length === 0
- ? html`<tr><td colspan="6" class="muted">No runs to show.</td></tr>`
- : runs.map((run) => html`<tr>
- <td><a href="/runs/${run.id}">#${run.number}</a></td>
- <td>${run.project_id}</td>
- <td class="muted">${(run.ref ?? '').replace('refs/heads/', '')}</td>
- <td class="mono muted">${shortSha(run.head_sha)}</td>
- <td>${badge(run.state)}</td>
- <td class="muted">${ago(run.created_at)}</td>
+ ${jobs.length === 0
+ ? html`<tr><td colspan="6" class="muted">No jobs to show.</td></tr>`
+ : jobs.map((job) => html`<tr>
+ <td><a href="/jobs/${job.id}">#${job.number}</a></td>
+ <td>${job.project_id}</td>
+ <td class="muted">${(job.ref ?? '').replace('refs/heads/', '')}</td>
+ <td class="mono muted">${shortSha(job.head_sha)}</td>
+ <td>${badge(job.state)}</td>
+ <td class="muted">${ago(job.created_at)}</td>
</tr>`)}
</tbody>
</table>
</div>`;
}
-export function runPage({ run, jobs, canManage }) {
+export function jobPage({ job, tasks, canManage }) {
return html`
<div class="panel">
- <h2>${run.title || `Run #${run.number}`}</h2>
+ <h2>${job.title || `Job #${job.number}`}</h2>
<div class="row">
- ${field('project', html`<a href="/runs?project=${run.project_id}">${run.project_id}</a>`)}
- ${field('state', badge(run.state))}
- ${field('ref', (run.ref ?? '').replace('refs/heads/', '') || '-')}
- ${field('commit', html`<span class="mono">${shortSha(run.head_sha)}</span>`)}
- ${field('trigger', `${run.trigger_type}${run.actor ? ` by ${run.actor}` : ''}`)}
- ${field('visibility', run.visibility)}
- ${field('duration', duration(run.started_at, run.finished_at))}
+ ${field('project', html`<a href="/jobs?project=${job.project_id}">${job.project_id}</a>`)}
+ ${field('state', badge(job.state))}
+ ${field('ref', (job.ref ?? '').replace('refs/heads/', '') || '-')}
+ ${field('commit', html`<span class="mono">${shortSha(job.head_sha)}</span>`)}
+ ${field('trigger', `${job.trigger_type}${job.actor ? ` by ${job.actor}` : ''}`)}
+ ${field('visibility', job.visibility)}
+ ${field('duration', duration(job.started_at, job.finished_at))}
</div>
${canManage ? html`<div class="actions">
- ${run.state === 'running'
- ? html`<button hx-post="/runs/${run.id}/cancel" hx-target="#jobs" hx-swap="outerHTML">cancel</button>`
+ ${job.state === 'running'
+ ? html`<button hx-post="/jobs/${job.id}/cancel" hx-target="#tasks" hx-swap="outerHTML">cancel</button>`
: ''}
- <button hx-post="/runs/${run.id}/retry" hx-swap="none">run again</button>
+ <button hx-post="/jobs/${job.id}/retry" hx-swap="none">run again</button>
</div>` : ''}
</div>
- ${jobsTable(run, jobs)}
+ ${tasksTable(job, tasks)}
`;
}
-export function jobsTable(run, jobs) {
- const live = ACTIVE_RUN_STATES.includes(run.state);
+export function tasksTable(job, tasks) {
+ const live = ACTIVE_RUN_STATES.includes(job.state);
// Grouped by graph depth, which is how the pipeline reads.
- const byId = new Map(jobs.map((j) => [j.id, j]));
- const depthOf = (job, seen = new Set()) => {
- if (seen.has(job.id)) return 0;
- seen.add(job.id);
- const deps = (job.needs ?? []).map((id) => byId.get(id)).filter(Boolean);
+ 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);
return deps.length === 0 ? 0 : Math.max(...deps.map((d) => depthOf(d, seen) + 1));
};
const stages = new Map();
- for (const job of jobs) {
- const depth = depthOf(job);
+ for (const task of tasks) {
+ const depth = depthOf(task);
if (!stages.has(depth)) stages.set(depth, []);
- stages.get(depth).push(job);
+ stages.get(depth).push(task);
}
- return html`<div id="jobs" ${live ? poll(`/partials/runs/${run.id}/jobs`, 3) : ''}>
+ return html`<div id="tasks" ${live ? poll(`/partials/jobs/${job.id}/tasks`, 3) : ''}>
${[...stages.keys()].sort((a, b) => a - b).map((depth) => html`
<div class="stage">
<h3>stage ${depth + 1}</h3>
<table><tbody>
- ${stages.get(depth).sort((a, b) => a.name.localeCompare(b.name)).map((job) => html`<tr>
- <td><a href="/jobs/${job.id}">${job.name}</a></td>
- <td>${badge(job.state)}</td>
- <td class="muted">${job.arch ?? ''}</td>
- <td class="muted">${job.worker_name ?? ''}</td>
- <td class="muted">${duration(job.started_at, job.finished_at)}</td>
- <td class="muted">${job.exit_code === null ? '' : `exit ${job.exit_code}`}</td>
+ ${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">${duration(task.started_at, task.finished_at)}</td>
+ <td class="muted">${task.exit_code === null ? '' : `exit ${task.exit_code}`}</td>
</tr>`)}
</tbody></table>
</div>`)}
</div>`;
}
-// --- jobs ---
+// --- tasks ---
-export function jobPage({ job, artifacts, log }) {
+export function taskPage({ task, artifacts, log }) {
return html`
<div class="panel">
- <h2>${job.name}</h2>
+ <h2>${task.name}</h2>
<div class="row">
- ${field('state', badge(job.state))}
- ${field('run', html`<a href="/runs/${job.run_id}">back to run</a>`)}
- ${field('image', html`<span class="mono">${job.image}</span>`)}
- ${field('arch', job.arch ?? '-')}
- ${field('attempt', `${job.attempt} of ${job.max_attempts}`)}
- ${field('worker', job.worker_name ?? '-')}
- ${field('duration', duration(job.started_at, job.finished_at))}
+ ${field('state', badge(task.state))}
+ ${field('job', html`<a href="/jobs/${task.job_id}">back to job</a>`)}
+ ${field('image', html`<span class="mono">${task.image}</span>`)}
+ ${field('arch', task.arch ?? '-')}
+ ${field('attempt', `${task.attempt} of ${task.max_attempts}`)}
+ ${field('worker', task.worker_name ?? '-')}
+ ${field('duration', duration(task.started_at, task.finished_at))}
</div>
- ${job.error ? notice(job.error) : ''}
+ ${task.error ? notice(task.error) : ''}
</div>
<h3>log</h3>
- <div class="logpane">${jobLog(job, log)}</div>
+ <div class="logpane">${taskLog(task, log)}</div>
${artifacts.length === 0 ? '' : html`
<h3>artifacts</h3>
<table><tbody>
${artifacts.map((a) => html`<tr>
- <td><a href="/api/v1/projects/${job.project_id}/runs/${job.run_id}/jobs/${job.id}/artifacts/${a.id}">${a.path}</a></td>
+ <td><a href="/api/v1/projects/${task.project_id}/jobs/${task.job_id}/tasks/${task.id}/artifacts/${a.id}">${a.path}</a></td>
<td class="muted">${a.size} bytes</td>
<td class="mono muted">${a.sha256.slice(0, 12)}</td>
</tr>`)}
@@ -137,9 +137,9 @@ export function jobPage({ job, artifacts, log }) {
`;
}
-export function jobLog(job, log) {
- const live = ACTIVE_JOB_STATES.includes(job.state);
- return html`<pre id="log" class="log" ${live ? poll(`/partials/jobs/${job.id}/log`, 2) : ''}>${
+export function taskLog(task, log) {
+ const live = ACTIVE_JOB_STATES.includes(task.state);
+ return html`<pre id="log" class="log" ${live ? poll(`/partials/tasks/${task.id}/log`, 2) : ''}>${
log || (live ? 'Waiting for output.' : 'No output.')
}</pre>`;
}
@@ -190,7 +190,7 @@ export function projectsPage({ projects, user }) {
export function projectsTable(projects, user) {
return html`<table id="projects">
- <thead><tr><th>project</th><th>repository</th><th>visibility</th><th>owner</th><th>runs</th><th></th></tr></thead>
+ <thead><tr><th>project</th><th>repository</th><th>visibility</th><th>owner</th><th>jobs</th><th></th></tr></thead>
<tbody>
${projects.length === 0
? html`<tr><td colspan="6" class="muted">No projects yet.</td></tr>`
@@ -221,7 +221,7 @@ export function projectPage({ project, variables, triggerUrl, secret, user, rete
${field('pipeline', html`<span class="mono">${project.config_path}</span>`)}
${field('owner', project.owner_id === null ? 'shared' : (project.owner_id === user.id ? 'you' : project.owner_id))}
- ${field('runs', project.run_counter)}
+ ${field('jobs', project.run_counter)}
</div>
<p class="muted">Trigger endpoint</p>
<input class="wide mono" type="text" readonly value="${triggerUrl}" onclick="this.select()">
@@ -251,7 +251,7 @@ export function projectPage({ project, variables, triggerUrl, secret, user, rete
<p class="muted">
A pipeline may override visibility per commit with a top level
visibility key, and the working directory with a workdir key.
- Left empty, jobs run in ${retention.workdir}.
+ Left empty, tasks run in ${retention.workdir}.
</p>
<div class="actions"><button type="submit">save</button></div>
</form>
@@ -265,10 +265,10 @@ export function projectPage({ project, variables, triggerUrl, secret, user, rete
</p>
<form hx-patch="/projects/${project.id}/retention" hx-target="body" hx-swap="none">
<div class="grid">
- <label>keep artifacts for runs
- <input name="artifact_keep_runs" type="number" min="0" inputmode="numeric"
- placeholder="${retention.artifact_keep_runs}"
- value="${project.artifact_keep_runs ?? ''}">
+ <label>keep artifacts for jobs
+ <input name="artifact_keep_jobs" type="number" min="0" inputmode="numeric"
+ placeholder="${retention.artifact_keep_jobs}"
+ value="${project.artifact_keep_jobs ?? ''}">
</label>
<label>keep artifacts for days
<input name="artifact_keep_days" type="number" min="0" inputmode="numeric"
@@ -283,8 +283,8 @@ export function projectPage({ project, variables, triggerUrl, secret, user, rete
</div>
<p class="muted">
An artifact is kept if either rule wants it, so the last
- ${project.artifact_keep_runs ?? retention.artifact_keep_runs} runs survive
- whatever their age. The most recent successful run is always kept. A job
+ ${project.artifact_keep_jobs ?? retention.artifact_keep_jobs} jobs survive
+ whatever their age. The most recent successful job is always kept. A task
that sets artifacts.expire overrides all of this.
</p>
<div class="actions"><button type="submit">save</button></div>
@@ -293,7 +293,7 @@ export function projectPage({ project, variables, triggerUrl, secret, user, rete
<div class="panel">
<h3>Variables</h3>
- <p class="muted">Injected into every job. Masked values are redacted from logs.</p>
+ <p class="muted">Injected into every task. Masked values are redacted from logs.</p>
${variablesTable(project, variables)}
<form hx-put="/projects/${project.id}/variables" hx-target="#variables" hx-swap="outerHTML">
<div class="grid">
@@ -315,7 +315,7 @@ export function projectPage({ project, variables, triggerUrl, secret, user, rete
</button>
<button class="danger"
hx-delete="/projects/${project.id}"
- hx-confirm="Delete ${project.id} and all of its runs?"
+ hx-confirm="Delete ${project.id} and all of its jobs?"
hx-swap="none">delete project</button>
</div>
</div>
@@ -343,8 +343,8 @@ export function workersPage({ tokens, user, created }) {
return html`
<h2>Workers</h2>
<p class="muted">
- A worker you register only ever receives jobs from your own projects.
- ${user.role === 'admin' ? 'A shared worker receives jobs from any project.' : ''}
+ A worker you register only ever receives tasks from your own projects.
+ ${user.role === 'admin' ? 'A shared worker receives tasks from any project.' : ''}
</p>
${created ? oneTimeSecret(
`Worker token for ${created.name}`,
@@ -408,7 +408,7 @@ export function usersPage({ users, localLogin }) {
<label>username<input name="username" required></label>
<label>password<input name="password" type="password" required></label>
<label>role
- <select name="role"><option value="viewer">viewer</option><option value="admin">admin</option></select>
+ <select name="role"><option value="user">user</option><option value="admin">admin</option></select>
</label>
</div>
<div class="actions"><button type="submit">create</button></div>
@@ -450,7 +450,7 @@ function deleteWarning(user) {
}
const parts = [];
if (user.project_count > 0) {
- parts.push(`${user.project_count} project(s) and all of their run history`);
+ parts.push(`${user.project_count} project(s) and all of their job history`);
}
if (user.worker_count > 0) parts.push(`${user.worker_count} worker token(s)`);
return `Remove ${user.username}? This also deletes ${parts.join(' and ')}. This cannot be undone.`;
diff --git a/src/conductor/ui/routes.js b/src/conductor/ui/routes.js
@@ -12,7 +12,7 @@ import crypto from 'node:crypto';
import { toHtml, html } from './html.js';
import { layout } from './layout.js';
import {
- runsPage, runsTable, runPage, jobsTable, jobPage, jobLog, loginPage,
+ jobsPage, jobsTable, jobPage, tasksTable, taskPage, taskLog, loginPage,
projectsPage, projectPage, variablesTable, workersPage, workersTable,
usersPage, usersTable,
} from './pages.js';
@@ -49,7 +49,7 @@ export default async function uiRoutes(fastify, services) {
return send(reply, layout({ title, user, body, active, localLogin: auth.localLogin }));
}
- const viewer = (req) => auth.identify(req).catch(() => null);
+ const currentUser = (req) => auth.identify(req).catch(() => null);
// Only a path on this conductor, so a crafted link cannot bounce someone
// to another site after signing in.
@@ -80,7 +80,7 @@ export default async function uiRoutes(fastify, services) {
// A signed in user, or a redirect to sign in. With OIDC that means the
// provider; otherwise the built-in sign in page.
async function required(req, reply) {
- const user = await viewer(req);
+ const user = await currentUser(req);
if (user) return user;
if (auth.oidc) {
@@ -122,189 +122,189 @@ export default async function uiRoutes(fastify, services) {
.send(toHtml(html`<p class="error">${message}</p>`));
}
- // --- runs ---
+ // --- jobs ---
- async function visibleRuns(user, projectId = null) {
+ async function visibleJobs(user, projectId = null) {
const scope = user && user.role === 'admin'
? { sql: '1 = 1', params: {} }
: user
- ? { sql: "(r.visibility = 'public' OR p.owner_id = {viewerId})", params: { viewerId: user.id } }
- : { sql: "r.visibility = 'public'", params: {} };
+ ? { sql: "(j.visibility = 'public' OR p.owner_id = {viewerId})", params: { viewerId: user.id } }
+ : { sql: "j.visibility = 'public'", params: {} };
return db.all(
- `SELECT r.id, r.project_id, r.number, r.ref, r.head_sha, r.state, r.visibility,
- r.title, r.created_at, r.started_at, r.finished_at
- FROM runs r JOIN projects p ON p.id = r.project_id
- WHERE ${scope.sql} ${projectId ? 'AND r.project_id = {project}' : ''}
- ORDER BY r.created_at DESC LIMIT 50`,
+ `SELECT j.id, j.project_id, j.number, j.ref, j.head_sha, j.state, j.visibility,
+ j.title, j.created_at, j.started_at, j.finished_at
+ FROM jobs j JOIN projects p ON p.id = j.project_id
+ WHERE ${scope.sql} ${projectId ? 'AND j.project_id = {project}' : ''}
+ ORDER BY j.created_at DESC LIMIT 50`,
{ ...scope.params, ...(projectId ? { project: projectId } : {}) }
);
}
fastify.get('/', async (req, reply) => {
- const user = await viewer(req);
- const runs = await visibleRuns(user, typeof req.query.project === 'string' ? req.query.project : null);
+ const user = await currentUser(req);
+ const jobs = await visibleJobs(user, typeof req.query.project === 'string' ? req.query.project : null);
return page(reply, {
- title: 'runs',
+ title: 'jobs',
user,
- active: 'runs',
- body: runsPage({ runs, user, anonymous: !user }),
+ active: 'jobs',
+ body: jobsPage({ jobs, user, anonymous: !user }),
});
});
- fastify.get('/runs', async (req, reply) => {
- const user = await viewer(req);
- const runs = await visibleRuns(user, typeof req.query.project === 'string' ? req.query.project : null);
- return page(reply, { title: 'runs', user, active: 'runs', body: runsPage({ runs, user, anonymous: !user }) });
+ fastify.get('/jobs', async (req, reply) => {
+ const user = await currentUser(req);
+ const jobs = await visibleJobs(user, typeof req.query.project === 'string' ? req.query.project : null);
+ return page(reply, { title: 'jobs', user, active: 'jobs', body: jobsPage({ jobs, user, anonymous: !user }) });
});
- fastify.get('/partials/runs', async (req, reply) => {
- const user = await viewer(req);
- return send(reply, runsTable(await visibleRuns(user)));
+ fastify.get('/partials/jobs', async (req, reply) => {
+ const user = await currentUser(req);
+ return send(reply, jobsTable(await visibleJobs(user)));
});
- // Loads a run the caller may see, along with whether they may act on it.
- async function runFor(req, reply) {
- const user = await viewer(req);
- const run = await db.get('SELECT * FROM runs WHERE id = {id}', { id: req.params.id });
- if (!run) {
+ // Loads a job the caller may see, along with whether they may act on it.
+ async function jobFor(req, reply) {
+ const user = await currentUser(req);
+ const job = await db.get('SELECT * FROM jobs WHERE id = {id}', { id: req.params.id });
+ if (!job) {
reply.code(404);
- return { user, run: null };
+ return { user, job: null };
}
- const project = await projects.get(run.project_id);
- if (run.visibility !== 'public' && !canManageProject(user, project)) {
+ const project = await projects.get(job.project_id);
+ if (job.visibility !== 'public' && !canManageProject(user, project)) {
reply.code(404);
- return { user, run: null };
+ return { user, job: null };
}
- return { user, run, project, canManage: canManageProject(user, project) };
+ return { user, job, project, canManage: canManageProject(user, project) };
}
- async function jobsOf(runId) {
- const jobs = await db.all(
+ async function tasksOf(jobId) {
+ const tasks = await db.all(
`SELECT id, name, arch, state, worker_name, exit_code, started_at, finished_at
- FROM jobs WHERE run_id = {run} ORDER BY name`,
- { run: runId }
+ FROM tasks WHERE job_id = {job} ORDER BY name`,
+ { job: jobId }
);
const deps = await db.all(
- `SELECT d.job_id, d.depends_on_id FROM job_deps d
- JOIN jobs j ON j.id = d.job_id WHERE j.run_id = {run}`,
- { run: runId }
+ `SELECT d.task_id, d.depends_on_id FROM task_deps d
+ JOIN tasks t ON t.id = d.task_id WHERE t.job_id = {job}`,
+ { job: jobId }
);
- const needs = new Map(jobs.map((j) => [j.id, []]));
- for (const d of deps) needs.get(d.job_id)?.push(d.depends_on_id);
- return jobs.map((j) => ({ ...j, needs: needs.get(j.id) ?? [] }));
+ 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) ?? [] }));
}
- fastify.get('/runs/:id', async (req, reply) => {
- const { user, run, canManage } = await runFor(req, reply);
- if (!run) return page(reply, { title: 'not found', user, body: html`<p class="error">No such run.</p>` });
+ fastify.get('/jobs/:id', async (req, reply) => {
+ const { user, job, canManage } = await jobFor(req, reply);
+ if (!job) return page(reply, { title: 'not found', user, body: html`<p class="error">No such job.</p>` });
return page(reply, {
- title: `run #${run.number}`,
+ title: `job #${job.number}`,
user,
- active: 'runs',
- body: runPage({ run, jobs: await jobsOf(run.id), canManage }),
+ active: 'jobs',
+ body: jobPage({ job, tasks: await tasksOf(job.id), canManage }),
});
});
- fastify.get('/partials/runs/:id/jobs', async (req, reply) => {
- const { run } = await runFor(req, reply);
- if (!run) return reply.send('');
- return send(reply, jobsTable(run, await jobsOf(run.id)));
+ fastify.get('/partials/jobs/:id/tasks', async (req, reply) => {
+ const { job } = await jobFor(req, reply);
+ if (!job) return reply.send('');
+ return send(reply, tasksTable(job, await tasksOf(job.id)));
});
- fastify.post('/runs/:id/cancel', async (req, reply) => {
+ fastify.post('/jobs/:id/cancel', async (req, reply) => {
if (!fromUi(req, reply)) return reply;
const user = await required(req, reply);
if (!user) return reply;
- const { run, canManage } = await runFor(req, reply);
- if (!run || !canManage) return fail(reply, 'Not yours to manage.', 403);
+ const { job, canManage } = await jobFor(req, reply);
+ if (!job || !canManage) return fail(reply, 'Not yours to manage.', 403);
- await scheduler.cancelRun(run.id, `cancelled by ${user.username}`);
- const fresh = await db.get('SELECT * FROM runs WHERE id = {id}', { id: run.id });
- return send(reply, jobsTable(fresh, await jobsOf(run.id)));
+ await scheduler.cancelJob(job.id, `cancelled by ${user.username}`);
+ const fresh = await db.get('SELECT * FROM jobs WHERE id = {id}', { id: job.id });
+ return send(reply, tasksTable(fresh, await tasksOf(job.id)));
});
- fastify.post('/runs/:id/retry', async (req, reply) => {
+ fastify.post('/jobs/:id/retry', async (req, reply) => {
if (!fromUi(req, reply)) return reply;
const user = await required(req, reply);
if (!user) return reply;
- const { run, project, canManage } = await runFor(req, reply);
- if (!run || !canManage) return fail(reply, 'Not yours to manage.', 403);
+ const { job, project, canManage } = await jobFor(req, reply);
+ if (!job || !canManage) return fail(reply, 'Not yours to manage.', 403);
try {
- const created = await scheduler.createRun(project, {
- ref: run.ref,
- baseSha: run.base_sha,
- headSha: run.head_sha,
+ const created = await scheduler.createJob(project, {
+ ref: job.ref,
+ baseSha: job.base_sha,
+ headSha: job.head_sha,
trigger: 'manual',
actor: user.username,
});
- return refresh(reply, `/runs/${created.runId}`);
+ return refresh(reply, `/jobs/${created.jobId}`);
} catch (e) {
return fail(reply, e instanceof PipelineError ? e.message : String(e.message ?? e), 422);
}
});
- // --- jobs ---
+ // --- tasks ---
- fastify.get('/jobs/:id', async (req, reply) => {
- const user = await viewer(req);
- const job = await db.get(
- `SELECT j.*, r.visibility, r.project_id FROM jobs j
- JOIN runs r ON r.id = j.run_id WHERE j.id = {id}`,
+ fastify.get('/tasks/:id', async (req, reply) => {
+ const user = await currentUser(req);
+ const task = await db.get(
+ `SELECT t.*, j.visibility, j.project_id FROM tasks t
+ JOIN jobs j ON j.id = t.job_id WHERE t.id = {id}`,
{ id: req.params.id }
);
- if (!job) return page(reply, { title: 'not found', user, body: html`<p class="error">No such job.</p>` });
+ if (!task) return page(reply, { title: 'not found', user, body: html`<p class="error">No such task.</p>` });
- const project = await projects.get(job.project_id);
- if (job.visibility !== 'public' && !canManageProject(user, project)) {
- return page(reply, { title: 'not found', user, body: html`<p class="error">No such job.</p>` });
+ const project = await projects.get(task.project_id);
+ if (task.visibility !== 'public' && !canManageProject(user, project)) {
+ return page(reply, { title: 'not found', user, body: html`<p class="error">No such task.</p>` });
}
const artifacts = await db.all(
- 'SELECT id, path, size, sha256 FROM artifacts WHERE job_id = {job} ORDER BY path',
- { job: job.id }
+ 'SELECT id, path, size, sha256 FROM artifacts WHERE task_id = {task} ORDER BY path',
+ { task: task.id }
);
return page(reply, {
- title: job.name,
+ title: task.name,
user,
- active: 'runs',
- body: jobPage({ job, artifacts, log: await tailOf(job) }),
+ active: 'jobs',
+ body: taskPage({ task, artifacts, log: await tailOf(task) }),
});
});
- fastify.get('/partials/jobs/:id/log', async (req, reply) => {
- const user = await viewer(req);
- const job = await db.get(
- `SELECT j.*, r.visibility, r.project_id FROM jobs j
- JOIN runs r ON r.id = j.run_id WHERE j.id = {id}`,
+ fastify.get('/partials/tasks/:id/log', async (req, reply) => {
+ const user = await currentUser(req);
+ const task = await db.get(
+ `SELECT t.*, j.visibility, j.project_id FROM tasks t
+ JOIN jobs j ON j.id = t.job_id WHERE t.id = {id}`,
{ id: req.params.id }
);
- if (!job) return reply.send('');
+ if (!task) return reply.send('');
- const project = await projects.get(job.project_id);
- if (job.visibility !== 'public' && !canManageProject(user, project)) return reply.send('');
+ const project = await projects.get(task.project_id);
+ if (task.visibility !== 'public' && !canManageProject(user, project)) return reply.send('');
- return send(reply, jobLog(job, await tailOf(job)));
+ return send(reply, taskLog(task, await tailOf(task)));
});
// The last window of output. Replacing the whole pane each poll keeps the
// fragment self contained, and a cap keeps a runaway log from being
// re-sent in full every two seconds.
- async function tailOf(job) {
- if (!job.log_key) {
- const size = await logs.size(job.run_id, job.id);
+ async function tailOf(task) {
+ if (!task.log_key) {
+ const size = await logs.size(task.job_id, task.id);
const offset = Math.max(0, size - LOG_TAIL_BYTES);
- const chunk = await logs.read(job.run_id, job.id, { offset, limit: LOG_TAIL_BYTES });
+ const chunk = await logs.read(task.job_id, task.id, { offset, limit: LOG_TAIL_BYTES });
return chunk.data.toString('utf8');
}
try {
- const start = Math.max(0, (job.log_size ?? 0) - LOG_TAIL_BYTES);
- const object = await services.storage.get(job.log_key, { range: { start } });
+ const start = Math.max(0, (task.log_size ?? 0) - LOG_TAIL_BYTES);
+ const object = await services.storage.get(task.log_key, { range: { start } });
const parts = [];
for await (const part of object.stream) parts.push(part);
return Buffer.concat(parts).toString('utf8');
@@ -316,7 +316,7 @@ export default async function uiRoutes(fastify, services) {
// --- session ---
fastify.get('/login', async (req, reply) => {
- const user = await viewer(req);
+ const user = await currentUser(req);
if (user) return reply.redirect('/', 302);
// With the provider in charge, asking for the sign in page is asking to
@@ -525,7 +525,7 @@ export default async function uiRoutes(fastify, services) {
const body = req.body ?? {};
const values = {};
- for (const key of ['artifact_keep_runs', 'artifact_keep_days', 'log_keep_days']) {
+ for (const key of ['artifact_keep_jobs', 'artifact_keep_days', 'log_keep_days']) {
if (!Object.hasOwn(body, key)) continue;
const raw = String(body[key] ?? '').trim();
// An empty field means "follow the server default", which is a null
@@ -614,7 +614,7 @@ export default async function uiRoutes(fastify, services) {
const name = String(req.body?.name ?? '').trim();
if (!name) return fail(reply, 'A name is required.');
- // Only an administrator can create capacity that runs anyone's work.
+ // Only an administrator can create capacity that jobs anyone's work.
const shared = user.role === 'admin' && req.body?.shared === 'true';
const created = await workerTokens.create(name, { ownerId: shared ? null : user.id });
@@ -688,7 +688,7 @@ export default async function uiRoutes(fastify, services) {
await users.create({
username: String(req.body?.username ?? ''),
password: String(req.body?.password ?? ''),
- role: req.body?.role === 'admin' ? 'admin' : 'viewer',
+ role: req.body?.role === 'admin' ? 'admin' : 'user',
});
} catch (e) {
return fail(reply, e.message);
diff --git a/src/lib/auth/oidc.js b/src/lib/auth/oidc.js
@@ -16,7 +16,7 @@
// groups, and there is no standard.
const ADMIN = 'admin';
-const VIEWER = 'viewer';
+const USER_ROLE = 'user';
const DISCOVERY_TIMEOUT = 10000;
const TOKEN_TIMEOUT = 10000;
@@ -106,7 +106,7 @@ export function createOidc(cfg) {
issuer,
subject: claims.sub,
username: claims.preferred_username ?? claims.email ?? claims.sub,
- role: roles.includes(adminRole) ? ADMIN : VIEWER,
+ role: roles.includes(adminRole) ? ADMIN : USER_ROLE,
roles,
};
}
diff --git a/src/lib/config.js b/src/lib/config.js
@@ -81,11 +81,11 @@ const DEFAULTS = {
max_size: 64 * 1024 * 1024,
},
// What happens to build output over time. A project may override any of
- // these; see migrations/sqlite/004_retention.sql for how the artifact
+ // these; see migrations/sqlite/001_initial.sql for how the artifact
// rules combine. Zero means keep forever.
retention: {
// The last this many runs keep their artifacts whatever their age.
- artifact_keep_runs: 10,
+ artifact_keep_jobs: 10,
// Artifacts younger than this are kept whatever has followed them.
artifact_keep_days: 30,
// Logs go purely by age, and are the reason this exists at all: they
@@ -103,14 +103,14 @@ const DEFAULTS = {
// considered lost and is requeued or failed.
heartbeat_timeout: 120,
reap_interval: 30,
- default_job_timeout: 3600,
+ default_task_timeout: 3600,
max_attempts: 3,
},
};
// [environment variable, dotted config path, parser]
const ENV_MAP = [
- ['CONDUCTOR_RETENTION_ARTIFACT_RUNS', 'retention.artifact_keep_runs', toInt],
+ ['CONDUCTOR_RETENTION_ARTIFACT_JOBS', 'retention.artifact_keep_jobs', toInt],
['CONDUCTOR_RETENTION_ARTIFACT_DAYS', 'retention.artifact_keep_days', toInt],
['CONDUCTOR_RETENTION_LOG_DAYS', 'retention.log_keep_days', toInt],
['CONDUCTOR_RETENTION_SWEEP_INTERVAL','retention.sweep_interval', toInt],
@@ -265,7 +265,7 @@ function validate(cfg) {
// Retention deletes things, so a nonsensical value here is worth
// refusing at startup rather than discovering from missing artifacts.
- for (const key of ['artifact_keep_runs', 'artifact_keep_days', 'log_keep_days']) {
+ for (const key of ['artifact_keep_jobs', 'artifact_keep_days', 'log_keep_days']) {
const value = cfg.retention[key];
if (!Number.isInteger(value) || value < 0) {
errors.push(`retention.${key} must be a whole number of zero or more, got ${JSON.stringify(value)}`);
diff --git a/src/lib/db/index.js b/src/lib/db/index.js
@@ -11,7 +11,7 @@
// Params is an object, and queries use named {name} markers which each driver
// compiles to its own placeholder syntax. See query.js.
//
-// db.get('SELECT * FROM jobs WHERE run_id = {run}', { run: runId })
+// db.get('SELECT * FROM tasks WHERE job_id = {job}', { job: jobId })
//
// Portability rules that apply to every query written against this layer:
// - named {name} markers only, never ? or $n directly
diff --git a/src/lib/db/query.js b/src/lib/db/query.js
@@ -2,7 +2,7 @@
//
// Queries are written once, against every dialect, using {name} markers:
//
-// db.get('SELECT * FROM jobs WHERE run_id = {run} AND state = {state}',
+// db.get('SELECT * FROM tasks WHERE job_id = {job} AND state = {state}',
// { run: runId, state: 'queued' })
//
// The marker is compiled to whatever the driver wants (? for sqlite and
diff --git a/src/lib/ids.js b/src/lib/ids.js
@@ -18,7 +18,7 @@ export function randomId(length = 24) {
return out;
}
-// Sortable by creation time, which keeps run listings stable without relying
+// Sortable by creation time, which keeps job listings stable without relying
// on a clock skewed created_at. 8 chars of millisecond timestamp in base32
// followed by 12 random chars.
export function timeOrderedId(now = Date.now()) {
@@ -31,29 +31,21 @@ export function timeOrderedId(now = Date.now()) {
return stamp + randomId(12);
}
-export const newRunId = () => timeOrderedId();
+export const newJobId = () => timeOrderedId();
export const newProjectId = () => randomId(16);
export const newArtifactId = () => randomId(24);
export const newUserId = () => randomId(16);
export const newWorkerTokenId = () => randomId(16);
-// Job ids stay human readable, since they appear in logs, storage keys and
-// URLs. The jobs.id column is 96 characters, so an unusually long job name
-// falls back to a truncated form with a hash suffix rather than being
-// rejected or silently colliding.
-export const JOB_ID_MAX = 96;
-
-export function jobId(runId, jobName) {
- const direct = `${runId}:${jobName}`;
- if (direct.length <= JOB_ID_MAX) return direct;
-
- const digest = crypto.createHash('sha256').update(jobName).digest('hex').slice(0, 12);
- const room = JOB_ID_MAX - runId.length - 1 - 1 - digest.length;
- if (room < 1) {
- throw new Error(`run id ${runId} leaves no room for a job id`);
- }
- return `${runId}:${jobName.slice(0, room)}~${digest}`;
-}
+// Task ids carry no structure at all. A worker holding one cannot tell which
+// job or project it belongs to, which is the point: a worker is given a
+// context and a script, and has no business knowing what they are for.
+//
+// It also removes a class of nuisance. The previous form embedded the task
+// name, so an id could contain ':' ',' and '=', and every use had to escape
+// it for a URL, sanitize it for a container name and encode it for a storage
+// key. Time ordered so that tasks of one job still group naturally.
+export const newTaskId = () => timeOrderedId();
// Worker and session tokens. The plaintext is shown once and only its hash
// is stored, so a database leak does not yield usable credentials.
diff --git a/src/lib/log.js b/src/lib/log.js
@@ -1,8 +1,8 @@
-// src/lib/log.js - job log spool
+// src/lib/log.js - task log spool
//
-// Logs are written to local disk while a job runs, then copied to object
+// Logs are written to local disk while a task runs, then copied to object
// storage when it finishes. The spool is always local even when S3 is
-// configured, because a running job produces many small appends and a live
+// configured, because a running task produces many small appends and a live
// tail wants cheap random reads; neither suits an object store.
//
// Appends carry the offset the worker believes it is writing at, which makes
@@ -44,10 +44,11 @@ export function createLogStore(cfg) {
}
}
- // Job ids contain a colon, which is fine on every filesystem we target,
- // but the run id still gives a directory per run to keep listings sane.
- function spoolPath(runId, jobId) {
- return path.join(root, encodeURIComponent(runId), `${encodeURIComponent(jobId)}.log`);
+ // A directory per job keeps listings sane, and lets a whole job's spool
+ // be removed in one go. Both ids are encoded anyway, which costs nothing
+ // now that neither carries a separator.
+ function spoolPath(jobId, taskId) {
+ return path.join(root, encodeURIComponent(jobId), `${encodeURIComponent(taskId)}.log`);
}
async function currentSize(file) {
@@ -62,15 +63,15 @@ export function createLogStore(cfg) {
return {
spoolPath,
- async size(runId, jobId) {
- return currentSize(spoolPath(runId, jobId));
+ async size(jobId, taskId) {
+ return currentSize(spoolPath(jobId, taskId));
},
// Returns the new total size. When offset is omitted the chunk is simply
// appended, which is what a worker that never retries will do.
- async append(runId, jobId, chunk, offset) {
- const file = spoolPath(runId, jobId);
- const key = `${runId}/${jobId}`;
+ async append(jobId, taskId, chunk, offset) {
+ const file = spoolPath(jobId, taskId);
+ const key = `${jobId}/${taskId}`;
const data = Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk), 'utf8');
return withLock(key, async () => {
@@ -112,8 +113,8 @@ export function createLogStore(cfg) {
// Reads a window for the live tail. Returns the total size too, so the
// caller knows whether more is already available.
- async read(runId, jobId, { offset = 0, limit = 256 * 1024 } = {}) {
- const file = spoolPath(runId, jobId);
+ async read(jobId, taskId, { offset = 0, limit = 256 * 1024 } = {}) {
+ const file = spoolPath(jobId, taskId);
const size = await currentSize(file);
if (size === 0 || offset >= size) {
return { data: Buffer.alloc(0), offset: Math.min(offset, size), size };
@@ -133,8 +134,8 @@ export function createLogStore(cfg) {
// Copies the finished log into object storage and drops the spool copy.
// Storage is the durable home; the spool only exists to serve a tail.
- async finalize(runId, jobId, storage, storageKey) {
- const file = spoolPath(runId, jobId);
+ async finalize(jobId, taskId, storage, storageKey) {
+ const file = spoolPath(jobId, taskId);
const size = await currentSize(file);
if (size === 0) return { key: null, size: 0 };
@@ -152,12 +153,12 @@ export function createLogStore(cfg) {
return { key: storageKey, size };
},
- async remove(runId, jobId) {
- await fs.rm(spoolPath(runId, jobId), { force: true });
+ async remove(jobId, taskId) {
+ await fs.rm(spoolPath(jobId, taskId), { force: true });
},
- async removeRun(runId) {
- await fs.rm(path.join(root, encodeURIComponent(runId)), { recursive: true, force: true });
+ async removeJob(jobId) {
+ await fs.rm(path.join(root, encodeURIComponent(jobId)), { recursive: true, force: true });
},
};
}
diff --git a/src/lib/pipeline/dag.js b/src/lib/pipeline/dag.js
@@ -1,7 +1,7 @@
// src/lib/pipeline/dag.js - dependency graph checks and traversal
//
-// The prototype stored a single integer level per job and treated the lowest
-// queued level as the next runnable job. That serialized every run and, on
+// The prototype stored a single integer level per task and treated the lowest
+// queued level as the next runnable task. That serialized every run and, on
// failure, only skipped direct dependents, so a transitive dependent could
// still be dispatched with its dependency missing. Both problems come from
// throwing the graph away, so the graph is kept here and used directly.
@@ -14,10 +14,10 @@ export class CycleError extends Error {
}
}
-// Returns the jobs in an order where every dependency precedes its
+// Returns the tasks in an order where every dependency precedes its
// dependents. Throws CycleError naming the cycle when none exists.
-export function topologicalOrder(jobs) {
- const byName = new Map(jobs.map((j) => [j.name, j]));
+export function topologicalOrder(tasks) {
+ const byName = new Map(tasks.map((j) => [j.name, j]));
const state = new Map();
const order = [];
const stack = [];
@@ -45,26 +45,26 @@ export function topologicalOrder(jobs) {
return order.map((name) => byName.get(name));
}
-// Depth of each job, where a job with no dependencies is 0. Used only for
+// 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(jobs) {
- const byName = new Map(jobs.map((j) => [j.name, j]));
+export function depths(tasks) {
+ const byName = new Map(tasks.map((j) => [j.name, j]));
const out = new Map();
- for (const job of topologicalOrder(jobs)) {
- const deps = job.needs.filter((d) => byName.has(d));
- out.set(job.name, deps.length === 0 ? 0 : Math.max(...deps.map((d) => out.get(d))) + 1);
+ for (const task of topologicalOrder(tasks)) {
+ const deps = task.needs.filter((d) => byName.has(d));
+ out.set(task.name, deps.length === 0 ? 0 : Math.max(...deps.map((d) => out.get(d))) + 1);
}
return out;
}
-// Every job reachable by following dependents from the given names. This is
+// 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(jobs, startNames) {
- const dependents = new Map(jobs.map((j) => [j.name, []]));
- for (const job of jobs) {
- for (const dep of job.needs) {
- if (dependents.has(dep)) dependents.get(dep).push(job.name);
+export function transitiveDependents(tasks, startNames) {
+ const dependents = new Map(tasks.map((j) => [j.name, []]));
+ for (const task of tasks) {
+ for (const dep of task.needs) {
+ if (dependents.has(dep)) dependents.get(dep).push(task.name);
}
}
@@ -81,11 +81,11 @@ export function transitiveDependents(jobs, startNames) {
return seen;
}
-// Jobs whose dependencies have all reached a satisfying state. Passed the
-// current state of every job by name.
-export function runnable(jobs, stateByName, { satisfied = ['success'] } = {}) {
- return jobs.filter((job) => {
- if (stateByName.get(job.name) !== 'queued') return false;
- return job.needs.every((dep) => satisfied.includes(stateByName.get(dep)));
+// 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'] } = {}) {
+ return tasks.filter((task) => {
+ if (stateByName.get(task.name) !== 'queued') return false;
+ return task.needs.every((dep) => satisfied.includes(stateByName.get(dep)));
});
}
diff --git a/src/lib/pipeline/expand.js b/src/lib/pipeline/expand.js
@@ -1,23 +1,23 @@
// src/lib/pipeline/expand.js - matrix and architecture expansion
//
-// Turns each job template into one concrete job per combination of its
-// dimensions, then resolves needs between the concrete jobs.
+// Turns each task template into one concrete task per combination of its
+// dimensions, then resolves needs between the concrete tasks.
//
-// Dimension matching is the part worth understanding. When a job declares
+// Dimension matching is the part worth understanding. When a task declares
// needs on a template that shares a dimension with it, the dependency is
// matched on that dimension rather than fanned out across all of it. So an
-// aarch64 package job needs the aarch64 build, not every build. Dimensions
+// aarch64 package task needs the aarch64 build, not every build. Dimensions
// the dependency has but the dependent does not are fanned out, which is
-// what you want when a single publish job waits for every architecture.
-// Write `{ job: name, match: all }` to opt out.
+// what you want when a single publish task waits for every architecture.
+// Write `{ task: name, match: all }` to opt out.
import { Problems, MATRIX_KEY_PATTERN } from './schema.js';
// Matches ${{ arch }} and ${{ matrix.pkg }}, tolerating inner whitespace.
const INTERPOLATION = /\$\{\{\s*([^}]*?)\s*\}\}/g;
-// jobs.name is 191 characters in the mysql schema.
-export const MAX_JOB_NAME = 191;
+// tasks.name is 191 characters in the mysql schema.
+export const MAX_TASK_NAME = 191;
export function interpolate(problems, path, text, context) {
if (typeof text !== 'string') return text;
@@ -25,7 +25,7 @@ export function interpolate(problems, path, text, context) {
return text.replace(INTERPOLATION, (whole, expr) => {
if (expr === 'arch') {
if (context.arch === null) {
- problems.add(path, 'refers to ${{ arch }} but the job declares no arch');
+ problems.add(path, 'refers to ${{ arch }} but the task declares no arch');
return whole;
}
return context.arch;
@@ -38,7 +38,7 @@ export function interpolate(problems, path, text, context) {
const known = Object.keys(context.matrix);
problems.add(
path,
- `refers to \${{ matrix.${key} }} which the job does not define` +
+ `refers to \${{ matrix.${key} }} which the task does not define` +
(known.length > 0 ? `; available: ${known.join(', ')}` : '')
);
return whole;
@@ -70,10 +70,10 @@ function instanceName(baseName, dims) {
return `${baseName}:${entries.map(([k, v]) => `${k}=${v}`).join(',')}`;
}
-function dimensionsOf(job) {
+function dimensionsOf(task) {
const dims = [];
- if (job.arch.length > 0) dims.push(['arch', job.arch]);
- for (const [key, values] of Object.entries(job.matrix)) dims.push([key, values]);
+ if (task.arch.length > 0) dims.push(['arch', task.arch]);
+ for (const [key, values] of Object.entries(task.matrix)) dims.push([key, values]);
return dims;
}
@@ -82,12 +82,12 @@ export function expandPipeline(pipeline, options = {}) {
const defaultTimeout = options.defaultTimeout ?? 3600;
const defaultAttempts = options.defaultAttempts ?? 1;
- // Pass one: build every concrete job.
+ // Pass one: build every concrete task.
const byBase = new Map();
const instances = [];
- for (const [baseName, job] of Object.entries(pipeline.jobs)) {
- const dims = dimensionsOf(job);
+ for (const [baseName, task] of Object.entries(pipeline.tasks)) {
+ const dims = dimensionsOf(task);
const combos = cartesian(dims);
const made = [];
@@ -97,18 +97,18 @@ export function expandPipeline(pipeline, options = {}) {
delete matrix.arch;
const name = instanceName(baseName, combo);
- const path = `jobs.${baseName}`;
- if (name.length > MAX_JOB_NAME) {
- problems.add(path, `expands to a job name longer than ${MAX_JOB_NAME} characters: ${name}`);
+ const path = `tasks.${baseName}`;
+ if (name.length > MAX_TASK_NAME) {
+ problems.add(path, `expands to a task name longer than ${MAX_TASK_NAME} characters: ${name}`);
continue;
}
const context = { arch, matrix };
const env = {};
- for (const [key, value] of Object.entries(job.env)) {
+ for (const [key, value] of Object.entries(task.env)) {
env[key] = interpolate(problems, `${path}.env.${key}`, value, context);
}
- // Dimension values are exposed to the script, so a matrix job rarely
+ // Dimension values are exposed to the script, so a matrix task rarely
// needs interpolation at all.
if (arch !== null) env.ARCH = arch;
for (const [key, value] of Object.entries(matrix)) {
@@ -121,10 +121,10 @@ export function expandPipeline(pipeline, options = {}) {
arch,
matrix,
dims: combo,
- image: interpolate(problems, `${path}.image`, job.image, context),
- script: job.script.map((line, i) => interpolate(problems, `${path}.script[${i}]`, line, context)),
- requires: job.requires.map((r, i) => interpolate(problems, `${path}.requires[${i}]`, r, context)),
- services: job.services.map((service, i) => ({
+ 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)),
+ services: task.services.map((service, i) => ({
image: interpolate(problems, `${path}.services[${i}].image`, service.image, context),
alias: service.alias,
env: Object.fromEntries(Object.entries(service.env).map(([k, v]) => [
@@ -134,59 +134,59 @@ export function expandPipeline(pipeline, options = {}) {
command: service.command,
})),
env,
- artifacts: job.artifacts
+ artifacts: task.artifacts
? {
- ...job.artifacts,
- paths: job.artifacts.paths.map((p, i) => interpolate(problems, `${path}.artifacts.paths[${i}]`, p, context)),
+ ...task.artifacts,
+ paths: task.artifacts.paths.map((p, i) => interpolate(problems, `${path}.artifacts.paths[${i}]`, p, context)),
}
: null,
- allow_failure: job.allow_failure,
- timeout: job.timeout ?? defaultTimeout,
- max_attempts: job.max_attempts ?? defaultAttempts,
- // Every instance of a matrix job inherits the ref rule, so a
- // restricted job stays restricted across all of its expansions.
- only: job.only ?? null,
+ allow_failure: task.allow_failure,
+ timeout: task.timeout ?? defaultTimeout,
+ max_attempts: task.max_attempts ?? defaultAttempts,
+ // Every instance of a matrix task inherits the ref rule, so a
+ // restricted task stays restricted across all of its expansions.
+ only: task.only ?? null,
needs: [],
});
}
- byBase.set(baseName, { job, dimensionNames: dims.map(([k]) => k), instances: made });
+ byBase.set(baseName, { task, dimensionNames: dims.map(([k]) => k), instances: made });
instances.push(...made);
}
const byName = new Map(instances.map((i) => [i.name, i]));
- // Pass two: resolve needs now that every concrete job exists.
+ // Pass two: resolve needs 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.job.needs) {
- const path = `jobs.${baseName}.needs`;
+ for (const need of entry.task.needs) {
+ const path = `tasks.${baseName}.needs`;
- // An exact concrete job name wins, which is the escape hatch for
+ // An exact concrete task name wins, which is the escape hatch for
// depending on one specific combination.
- if (byName.has(need.job)) {
- if (need.job === instance.name) {
- problems.add(path, `job ${JSON.stringify(baseName)} cannot depend on itself`);
+ 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.job);
+ resolved.add(need.task);
}
continue;
}
- const target = byBase.get(need.job);
+ const target = byBase.get(need.task);
if (!target) {
const known = [...byBase.keys()].filter((n) => n !== baseName);
problems.add(
path,
- `refers to unknown job ${JSON.stringify(need.job)}` +
- (known.length > 0 ? `; defined jobs are ${known.join(', ')}` : '')
+ `refers to unknown task ${JSON.stringify(need.task)}` +
+ (known.length > 0 ? `; defined tasks are ${known.join(', ')}` : '')
);
continue;
}
- if (need.job === baseName) {
- problems.add(path, `job ${JSON.stringify(baseName)} cannot depend on itself`);
+ if (need.task === baseName) {
+ problems.add(path, `task ${JSON.stringify(baseName)} cannot depend on itself`);
continue;
}
@@ -198,8 +198,8 @@ export function expandPipeline(pipeline, options = {}) {
if (candidates.length === 0) {
problems.add(
path,
- `no instance of ${JSON.stringify(need.job)} matches ${instance.name} on ` +
- `${shared.join(', ')}; add the missing value, or use { job: ${need.job}, match: all }`
+ `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;
}
diff --git a/src/lib/pipeline/index.js b/src/lib/pipeline/index.js
@@ -1,6 +1,6 @@
// src/lib/pipeline/index.js - pipeline compilation
//
-// compilePipeline turns the text of a .conductor.yml into the concrete job
+// compilePipeline turns the text of a .conductor.yml into the concrete task
// graph the scheduler stores. Every failure mode raises PipelineError with
// the full list of problems, so a bad pipeline is reported once rather than
// one mistake per push.
@@ -14,7 +14,7 @@ import { PipelineError } from './schema.js';
export { parsePipeline } from './parse.js';
export { expandPipeline, interpolate } from './expand.js';
export { topologicalOrder, depths, transitiveDependents, runnable, CycleError } from './dag.js';
-export { refMatches, jobRunsOnRef, selectForRef } from './only.js';
+export { refMatches, taskRunsOnRef, selectForRef } from './only.js';
export { PipelineError } from './schema.js';
export { SUPPORTED_VERSION, VISIBILITIES, DEFAULT_WORKDIR, parseWorkdir } from './parse.js';
@@ -22,29 +22,29 @@ export function compilePipeline(text, options = {}) {
const pipeline = parsePipeline(text, options);
const expanded = expandPipeline(pipeline, options);
- // Jobs restricted to other refs drop out before the graph is checked, so
+ // Tasks restricted to other refs drop out before the graph is checked, so
// ordering and depth describe the run that will actually happen. Without
// a ref nothing is filtered, which keeps validation of a pipeline
// separate from deciding what one push will run.
- const jobs = options.ref === undefined ? expanded : selectForRef(expanded, options.ref);
+ const tasks = options.ref === undefined ? expanded : selectForRef(expanded, options.ref);
try {
- topologicalOrder(jobs);
+ topologicalOrder(tasks);
} catch (e) {
if (e instanceof CycleError) {
- throw new PipelineError([{ path: 'jobs', message: e.message }], pipeline.source);
+ throw new PipelineError([{ path: 'tasks', message: e.message }], pipeline.source);
}
throw e;
}
- const depth = depths(jobs);
- for (const job of jobs) job.depth = depth.get(job.name);
+ const depth = depths(tasks);
+ for (const task of tasks) task.depth = depth.get(task.name);
return {
version: pipeline.version,
source: pipeline.source,
visibility: pipeline.visibility,
workdir: pipeline.workdir,
- jobs,
+ tasks,
};
}
diff --git a/src/lib/pipeline/only.js b/src/lib/pipeline/only.js
@@ -1,16 +1,16 @@
-// src/lib/pipeline/only.js - restricting jobs to particular refs
+// src/lib/pipeline/only.js - restricting tasks to particular refs
//
-// A job carrying an `only` rule runs when the ref that triggered the run
+// 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 job as a reason to fail the run: that is 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.
//
-// Anything depending on an excluded job is excluded with it. The
-// alternative, quietly dropping the dependency, would let a job run
+// Anything depending on 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 job not running.
+// the task not running.
import { transitiveDependents } from './dag.js';
@@ -29,23 +29,23 @@ export function refMatches(pattern, ref) {
return new RegExp(`^${source}$`).test(ref);
}
-export function jobRunsOnRef(job, ref) {
- if (!job.only) return true;
+export function taskRunsOnRef(task, ref) {
+ if (!task.only) return true;
// A run with no ref at all, which a manual or api trigger may produce,
- // cannot match a ref pattern. Restricted jobs stay out rather than
+ // cannot match a ref pattern. Restricted tasks stay out rather than
// being handed a ref they were never written for.
if (typeof ref !== 'string' || ref === '') return false;
- return job.only.refs.some((pattern) => refMatches(pattern, ref));
+ return task.only.refs.some((pattern) => refMatches(pattern, ref));
}
-// The jobs that belong in a run for this ref, in their original order.
-export function selectForRef(jobs, ref) {
- const excluded = new Set(jobs.filter((job) => !jobRunsOnRef(job, ref)).map((job) => job.name));
- if (excluded.size === 0) return jobs;
+// The tasks that belong in a run for this ref, in their original order.
+export function selectForRef(tasks, ref) {
+ const excluded = new Set(tasks.filter((task) => !taskRunsOnRef(task, ref)).map((task) => task.name));
+ if (excluded.size === 0) return tasks;
- for (const name of transitiveDependents(jobs, [...excluded])) excluded.add(name);
+ for (const name of transitiveDependents(tasks, [...excluded])) excluded.add(name);
- return jobs.filter((job) => !excluded.has(job.name));
+ return tasks.filter((task) => !excluded.has(task.name));
}
diff --git a/src/lib/pipeline/parse.js b/src/lib/pipeline/parse.js
@@ -1,6 +1,6 @@
// src/lib/pipeline/parse.js - reads and validates a .conductor.yml
//
-// Produces a normalized document: every job has every field filled in from
+// Produces a normalized document: every task has every field filled in from
// defaults, so later stages never have to ask whether something was set.
// Nothing here expands matrices or resolves dependencies; see expand.js.
@@ -22,13 +22,13 @@ import {
NAME_PATTERN,
} from './schema.js';
-const TOP_KEYS = ['version', 'visibility', 'workdir', 'defaults', 'jobs'];
+const TOP_KEYS = ['version', 'visibility', 'workdir', 'defaults', 'tasks'];
// A repository decides whether its own build results are readable without
// signing in. Unset means the project's setting stands.
export const VISIBILITIES = ['public', 'private'];
-// Where the tree is put inside a job container, and the directory every
+// Where the tree is put inside a task container, and the directory every
// script starts in. Unset means the project's setting stands, and failing
// that the server default.
export const DEFAULT_WORKDIR = '/work';
@@ -72,7 +72,7 @@ const JOB_KEYS = [
'only',
];
-// Fields a job may inherit from defaults.
+// Fields a task may inherit from defaults.
const DEFAULT_KEYS = [
'image', 'arch', 'requires', 'env', 'services',
'allow_failure', 'timeout', 'max_attempts',
@@ -104,6 +104,16 @@ export function parsePipeline(text, options = {}) {
throw new PipelineError([{ path: '', message: `expected a mapping at the top level, got ${typeName(doc)}` }], source);
}
+ // The key was called jobs before a job became the thing that contains
+ // these. Named outright rather than left to checkUnknown, whose nearest
+ // match would be a guess at a word that used to be correct.
+ if (doc.jobs !== undefined && doc.tasks === undefined) {
+ throw new PipelineError(
+ [{ path: 'jobs', message: 'was renamed to tasks; a job is now the pipeline run that these tasks belong to' }],
+ source
+ );
+ }
+
checkUnknown(problems, '', doc, TOP_KEYS);
if (doc.version === undefined) {
@@ -124,34 +134,34 @@ export function parsePipeline(text, options = {}) {
const defaults = parseDefaults(problems, doc.defaults);
- if (doc.jobs === undefined) {
- problems.add('jobs', 'is required');
- } else if (!isPlainObject(doc.jobs)) {
- problems.add('jobs', `expected a mapping of job names, got ${typeName(doc.jobs)}`);
- } else if (Object.keys(doc.jobs).length === 0) {
- problems.add('jobs', 'must define at least one job');
+ if (doc.tasks === undefined) {
+ problems.add('tasks', 'is required');
+ } else if (!isPlainObject(doc.tasks)) {
+ problems.add('tasks', `expected a mapping of task names, got ${typeName(doc.tasks)}`);
+ } else if (Object.keys(doc.tasks).length === 0) {
+ problems.add('tasks', 'must define at least one task');
}
- const jobs = {};
- if (isPlainObject(doc.jobs)) {
- for (const [name, raw] of Object.entries(doc.jobs)) {
- const path = `jobs.${name}`;
+ const tasks = {};
+ if (isPlainObject(doc.tasks)) {
+ for (const [name, raw] of Object.entries(doc.tasks)) {
+ const path = `tasks.${name}`;
if (!NAME_PATTERN.test(name) || name.length > 100) {
- problems.add(path, 'job name must start with a letter or digit and contain only letters, digits, underscore, dot and hyphen');
+ problems.add(path, 'task name must start with a letter or digit and contain only letters, digits, underscore, dot and hyphen');
continue;
}
if (!isPlainObject(raw)) {
problems.add(path, `expected a mapping, got ${typeName(raw)}`);
continue;
}
- jobs[name] = parseJob(problems, path, raw, defaults);
+ tasks[name] = parseTask(problems, path, raw, defaults);
}
}
const workdir = doc.workdir === undefined ? null : parseWorkdir(problems, 'workdir', doc.workdir);
problems.throwIfAny(source);
- return { version: SUPPORTED_VERSION, visibility, workdir, defaults, jobs, source };
+ return { version: SUPPORTED_VERSION, visibility, workdir, defaults, tasks, source };
}
function parseDefaults(problems, raw) {
@@ -202,12 +212,12 @@ function parseArch(problems, path, raw) {
return out;
}
-function parseJob(problems, path, raw, defaults) {
+function parseTask(problems, path, raw, defaults) {
checkUnknown(problems, path, raw, JOB_KEYS);
const image = raw.image === undefined ? defaults.image : asString(problems, `${path}.image`, raw.image);
if (image === undefined) {
- problems.add(`${path}.image`, 'is required; set it on the job or under defaults');
+ problems.add(`${path}.image`, 'is required; set it on the task or under defaults');
}
if (raw.script === undefined) {
@@ -246,15 +256,15 @@ function parseJob(problems, path, raw, defaults) {
};
}
-// Restricts a job to certain refs. Deliberately not inheritable from
-// defaults: a rule that silently applied to every job is the kind of thing
+// Restricts a task to certain refs. Deliberately not inheritable from
+// defaults: a rule that silently applied to every task is the kind of thing
// that stops a pipeline running at all and takes an afternoon to find.
//
// only:
// refs: [refs/heads/main, 'refs/tags/v*']
//
// Patterns match the whole ref, so refs/heads/main rather than main, with
-// * standing for any run of characters. Absent means the job always runs.
+// * standing for any run of characters. Absent means the task always runs.
function parseOnly(problems, path, raw) {
if (raw === undefined || raw === null) return null;
if (!isPlainObject(raw)) {
@@ -282,7 +292,7 @@ function parseOnly(problems, path, raw) {
return { refs };
}
-// A need is either a job name, or a mapping for the cases where the default
+// 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) {
if (raw === undefined) return [];
@@ -292,21 +302,21 @@ function parseNeeds(problems, path, raw) {
list.forEach((item, i) => {
const itemPath = `${path}[${i}]`;
if (typeof item === 'string') {
- out.push({ job: item, match: 'shared' });
+ out.push({ task: item, match: 'shared' });
return;
}
if (!isPlainObject(item)) {
- problems.add(itemPath, `expected a job name or a mapping, got ${typeName(item)}`);
+ problems.add(itemPath, `expected a task name or a mapping, got ${typeName(item)}`);
return;
}
- checkUnknown(problems, itemPath, item, ['job', 'match']);
- const job = asString(problems, `${itemPath}.job`, item.job);
+ checkUnknown(problems, itemPath, item, ['task', 'match']);
+ const task = asString(problems, `${itemPath}.task`, item.task);
const match = item.match === undefined ? 'shared' : asString(problems, `${itemPath}.match`, item.match);
if (match !== undefined && !['shared', 'all'].includes(match)) {
problems.add(`${itemPath}.match`, `expected shared or all, got ${JSON.stringify(match)}`);
return;
}
- if (job !== undefined) out.push({ job, match: match ?? 'shared' });
+ if (task !== undefined) out.push({ task, match: match ?? 'shared' });
});
return out;
@@ -346,7 +356,7 @@ function parseMatrix(problems, path, raw) {
problems.add(
`${keyPath}[${i}]`,
`${JSON.stringify(s)} may only contain letters, digits, underscore, dot and hyphen, ` +
- 'because it becomes part of the job name'
+ 'because it becomes part of the task name'
);
return;
}
@@ -383,7 +393,7 @@ function parseServices(problems, path, raw) {
if (image === undefined) return;
// Default alias is the image name without registry, path or tag, which
- // is what a job would naturally use as a hostname.
+ // is what a task would naturally use as a hostname.
const derived = image.split('/').pop().split(':')[0];
const alias = service.alias === undefined
? derived
diff --git a/src/lib/pipeline/schema.js b/src/lib/pipeline/schema.js
@@ -180,7 +180,7 @@ export function typeName(v) {
return `a ${typeof v}`;
}
-// Job and matrix value names appear in generated job names, storage keys and
+// Task and matrix value names appear in generated task names, storage keys and
// environment variables, so the character set is deliberately narrow.
export const NAME_PATTERN = /^[A-Za-z0-9][A-Za-z0-9_.-]*$/;
export const MATRIX_KEY_PATTERN = /^[A-Za-z_][A-Za-z0-9_]*$/;
diff --git a/src/lib/projects.js b/src/lib/projects.js
@@ -31,8 +31,8 @@ export function assertWorkdir(value) {
const COLUMNS = `
id, name, repo_url, default_branch, config_path, workdir,
- trigger_secret, enabled, run_counter, owner_id, visibility,
- artifact_keep_runs, artifact_keep_days, log_keep_days,
+ trigger_secret, enabled, job_counter, owner_id, visibility,
+ artifact_keep_jobs, artifact_keep_days, log_keep_days,
created_at, updated_at
`;
@@ -83,7 +83,7 @@ export function createProjects({ db, secrets }) {
await db.run(
`INSERT INTO projects
(id, name, repo_url, default_branch, config_path, workdir,
- trigger_secret, enabled, run_counter, owner_id, visibility,
+ trigger_secret, enabled, job_counter, owner_id, visibility,
created_at, updated_at)
VALUES
({id}, {name}, {repo_url}, {branch}, {config_path}, {workdir},
@@ -140,7 +140,7 @@ export function createProjects({ db, secrets }) {
// default; zero means keep forever, and has to stay distinguishable
// from null or a project could not opt out of a default that deletes.
async setRetention(id, values) {
- const columns = ['artifact_keep_runs', 'artifact_keep_days', 'log_keep_days'];
+ const columns = ['artifact_keep_jobs', 'artifact_keep_days', 'log_keep_days'];
const updates = [];
const params = { id, now: Date.now() };
@@ -184,13 +184,13 @@ export function createProjects({ db, secrets }) {
await db.run('DELETE FROM projects WHERE id = {id}', { id });
},
- // Allocates the next run number. Must run inside the same transaction
- // as the run insert, or two concurrent pushes can collide on the
+ // Allocates the next job number. Must run inside the same transaction
+ // as the job insert, or two concurrent pushes can collide on the
// unique (project_id, number) index.
- async nextRunNumber(tx, projectId) {
- await tx.run('UPDATE projects SET run_counter = run_counter + 1 WHERE id = {id}', { id: projectId });
- const row = await tx.get('SELECT run_counter FROM projects WHERE id = {id}', { id: projectId });
- return row.run_counter;
+ async nextJobNumber(tx, projectId) {
+ await tx.run('UPDATE projects SET job_counter = job_counter + 1 WHERE id = {id}', { id: projectId });
+ const row = await tx.get('SELECT job_counter FROM projects WHERE id = {id}', { id: projectId });
+ return row.job_counter;
},
};
}
diff --git a/src/lib/storage/index.js b/src/lib/storage/index.js
@@ -23,6 +23,6 @@ export function createStorage(cfg) {
// Key layout, kept in one place so every caller agrees.
export const keys = {
- artifact: (runId, jobId, relPath) => `artifacts/${runId}/${jobId}/${relPath}`,
- log: (runId, jobId) => `logs/${runId}/${jobId}.log`,
+ artifact: (jobId, taskId, relPath) => `artifacts/${jobId}/${taskId}/${relPath}`,
+ log: (jobId, taskId) => `logs/${jobId}/${taskId}.log`,
};
diff --git a/src/lib/users.js b/src/lib/users.js
@@ -6,7 +6,7 @@
import { newUserId } from './ids.js';
import { hashPassword, verifyPassword, generatePassword, validatePassword } from './auth/password.js';
-export const ROLES = ['admin', 'viewer'];
+export const ROLES = ['admin', 'user'];
// A provider may hand over anything as a display name, so it is reduced to
// the character set local accounts already use.
@@ -52,7 +52,7 @@ export function createUsers({ db, logger = console }) {
// administrator a way to lock someone out immediately.
async upsertExternal({ externalId, username, role }) {
if (!externalId) throw new Error('an external account needs a stable identifier');
- const desired = ROLES.includes(role) ? role : 'viewer';
+ const desired = ROLES.includes(role) ? role : 'user';
return db.transaction(async (tx) => {
const existing = await tx.get(
@@ -106,7 +106,7 @@ export function createUsers({ db, logger = console }) {
return (await db.get('SELECT COUNT(*) AS c FROM users')).c;
},
- async create({ username, password, role = 'viewer' }) {
+ async create({ username, password, role = 'user' }) {
if (typeof username !== 'string' || !/^[A-Za-z0-9][A-Za-z0-9_.@-]{1,63}$/.test(username)) {
throw new Error('username must be 2 to 64 characters of letters, digits, underscore, dot, at or hyphen');
}
diff --git a/src/worker/agent.js b/src/worker/agent.js
@@ -16,7 +16,7 @@ import { loadWorkerConfig, featureNames } from './config.js';
import { createClient } from './client.js';
import { createRuntime } from './docker.js';
-import { runJob } from './job.js';
+import { runTask } from './task.js';
const stamp = () => new Date().toISOString().replace('T', ' ').slice(0, 19);
@@ -49,11 +49,11 @@ export async function main(argv = process.argv.slice(2)) {
const client = createClient(cfg);
const runtime = createRuntime(cfg, { logger });
- // Fail at startup rather than on the first job, when a missing tool would
+ // Fail at startup rather than on the first task, when a missing tool would
// otherwise look like a broken pipeline.
//
// The runtime is the only thing a worker needs. The tree is unpacked
- // into the job container over the docker API rather than on this side,
+ // into the task container over the docker API rather than on this side,
// so there is nothing else to look for on PATH.
if (!(await runtime.available())) {
throw new Error(`container runtime ${JSON.stringify(cfg.docker)} is not usable; is the daemon running?`);
@@ -92,20 +92,20 @@ export async function main(argv = process.argv.slice(2)) {
process.exit(1);
}
stopping = true;
- logger.info(`${signal} received, finishing ${running.size} running job(s)`);
+ logger.info(`${signal} received, finishing ${running.size} running task(s)`);
for (const wake of [...sleepers]) wake();
};
for (const signal of ['SIGINT', 'SIGTERM']) process.once(signal, () => shutdown(signal));
- // One loop per concurrency slot. Each polls independently, so a slow job
+ // One loop per concurrency slot. Each polls independently, so a slow task
// in one slot does not stall the others.
async function slot(index) {
let backoff = cfg.poll_interval;
while (!stopping) {
- let job = null;
+ let task = null;
try {
- job = await client.poll({ arches: cfg.arches, features, name: cfg.name });
+ task = await client.claim({ arches: cfg.arches, features, name: cfg.name });
backoff = cfg.poll_interval;
} catch (e) {
// A conductor that is down or restarting should not become a busy
@@ -116,23 +116,23 @@ export async function main(argv = process.argv.slice(2)) {
continue;
}
- if (!job) {
+ if (!task) {
if (args.once) return;
await sleep(cfg.poll_interval * 1000);
continue;
}
- running.add(job.id);
- logger.info(`[${index}] ${job.name} (${job.image})`);
+ running.add(task.id);
+ logger.info(`[${index}] ${task.name} (${task.image})`);
const started = Date.now();
try {
- const result = await runJob({ cfg, client, runtime, job, logger });
+ const result = await runTask({ cfg, client, runtime, task, logger });
const seconds = ((Date.now() - started) / 1000).toFixed(1);
- logger.info(`[${index}] ${job.name} ${result.success ? 'succeeded' : 'failed'} in ${seconds}s`);
+ logger.info(`[${index}] ${task.name} ${result.success ? 'succeeded' : 'failed'} in ${seconds}s`);
} catch (e) {
- logger.error(`[${index}] ${job.name} crashed the worker loop: ${e.stack ?? e.message}`);
+ logger.error(`[${index}] ${task.name} crashed the worker loop: ${e.stack ?? e.message}`);
} finally {
- running.delete(job.id);
+ running.delete(task.id);
}
if (args.once) return;
diff --git a/src/worker/artifacts.js b/src/worker/artifacts.js
@@ -1,9 +1,9 @@
-// src/worker/artifacts.js - collecting and uploading job output
+// src/worker/artifacts.js - collecting and uploading task output
//
// Patterns are matched with the glob support built into node, so there is
-// nothing to install. Matches are confined to the workspace: a job can write
+// nothing to install. Matches are confined to the workspace: a task can write
// a symlink pointing anywhere it likes, and following one would upload a
-// file the job was never given.
+// file the task was never given.
import fs from 'node:fs/promises';
import path from 'node:path';
@@ -72,7 +72,7 @@ export async function uploadArtifacts(client, url, files, { logger = console } =
const result = await client.uploadArtifact(url, file.path, body, file.size);
uploaded.push(result);
} catch (e) {
- // An artifact that cannot be stored is worth reporting, but the job
+ // An artifact that cannot be stored is worth reporting, but the task
// itself already succeeded or failed on its own merits.
logger.warn?.(`could not upload ${file.path}: ${e.message}`);
}
diff --git a/src/worker/client.js b/src/worker/client.js
@@ -2,7 +2,7 @@
//
// Built on fetch, so there is nothing to install. Every call carries the
// worker token. Network errors are surfaced rather than retried here; the
-// agent decides what a failure means, since a failed poll and a failed log
+// agent decides what a failure means, since a failed claim and a failed log
// append want very different handling.
export class ConductorError extends Error {
@@ -35,19 +35,20 @@ export function createClient(cfg) {
}
return {
- // Returns a job, or null when there is nothing to do.
- async poll({ arches, features, name }) {
- const res = await fetch(`${base}/api/v1/workers/jobs`, {
+ // Returns a task, or null when there is nothing to do. This is the only
+ // URL the worker knows; everything else it needs arrives with the task.
+ async claim({ arches, features, name }) {
+ const res = await fetch(`${base}/api/v1/tasks/claim`, {
method: 'POST',
headers: { ...auth, 'content-type': 'application/json' },
body: JSON.stringify({ arches, features, name }),
});
if (res.status === 204) return null;
- if (!res.ok) throw await readError(res, 'poll');
- return (await res.json()).job;
+ if (!res.ok) throw await readError(res, 'claim');
+ return (await res.json()).task;
},
- // The tree at the job's commit. Returned as a web stream so it can be
+ // The tree at the task's commit. Returned as a web stream so it can be
// piped straight into tar without buffering.
async source(url) {
const res = await fetch(url, { headers: auth });
@@ -88,7 +89,7 @@ export function createClient(cfg) {
return await res.json();
},
- // Reports that the job is alive, and learns whether it should stop.
+ // Reports that the task is alive, and learns whether it should stop.
async heartbeat(url) {
const res = await fetch(url, {
method: 'POST',
diff --git a/src/worker/config.js b/src/worker/config.js
@@ -7,7 +7,7 @@
// which it is in the official image.
//
// Features are the important part. A feature is a name the worker
-// advertises, plus whatever local resources a job asking for it should get:
+// advertises, plus whatever local resources a task asking for it should get:
// mounts, environment, or a privileged container. The conductor only ever
// sees the name, so a signing key can be handed to a build without the
// server holding it.
@@ -27,9 +27,9 @@ const DEFAULTS = {
poll_interval: 5,
// Container runtime CLI. Podman and nerdctl are argument compatible.
docker: 'docker',
- // Shell used to run a job script inside its image.
+ // Shell used to run a task script inside its image.
shell: 'sh',
- // Applied when a job does not ask for something longer.
+ // Applied when a task does not ask for something longer.
default_timeout: 3600,
};
diff --git a/src/worker/docker.js b/src/worker/docker.js
@@ -14,9 +14,9 @@ import { promisify } from 'node:util';
const execFileAsync = promisify(execFile);
-// Docker requires [a-zA-Z0-9][a-zA-Z0-9_.-]*, while job names contain
-// colons, commas and equals signs. A hash suffix keeps names unique after
-// the unsafe characters are folded away.
+// Task ids are already within what docker accepts, but the sanitising stays
+// because the suffix is a caller supplied service alias and the id is not
+// this function's to trust. The hash keeps names unique if either is folded.
export function containerName(prefix, id, suffix = '') {
const digest = crypto.createHash('sha256').update(id).digest('hex').slice(0, 8);
const safe = id.replace(/[^a-zA-Z0-9_.-]/g, '-').replace(/^[^a-zA-Z0-9]+/, '').slice(0, 40);
@@ -38,7 +38,7 @@ export function createRuntime(cfg, { logger = console } = {}) {
}
}
- // Turns a job and the worker's feature definitions into run arguments.
+ // Turns a task and the worker's feature definitions into run arguments.
function buildRunArgs({ name, image, network, workdir = null, env = {}, mounts = [], privileged = false, devices = [], command = [], entrypoint = null, detach = false, networkAlias = null }) {
const args = ['run', '--rm', '--name', name];
if (detach) args.push('--detach');
@@ -63,7 +63,7 @@ export function createRuntime(cfg, { logger = console } = {}) {
return args;
}
- // A job container is created rather than run, so its filesystem can be
+ // A task container is created rather than run, so its filesystem can be
// populated before it starts and read after it exits. Nothing from the
// worker's own filesystem is mounted into it: the source arrives over
// the docker API, which means the worker needs no shared directory with
@@ -167,12 +167,12 @@ export function createRuntime(cfg, { logger = console } = {}) {
}
},
- // Services run detached; their output is not part of the job log.
+ // Services run detached; their output is not part of the task log.
async startService(options) {
return run(buildRunArgs({ ...options, detach: true }));
},
- // Starts the job container and hands back the process. The caller reads
+ // Starts the task container and hands back the process. The caller reads
// stdout and stderr, and awaits the exit code.
start(options) {
const args = buildRunArgs({ ...options, detach: false });
@@ -181,7 +181,7 @@ export function createRuntime(cfg, { logger = console } = {}) {
},
// Runs a container to completion and returns its output. Used for
- // short housekeeping tasks rather than for jobs.
+ // short housekeeping work rather than for running a task.
async runOnce(options, { timeout = 120000 } = {}) {
return run(buildRunArgs({ ...options, detach: false }), { timeout });
},
diff --git a/src/worker/logstream.js b/src/worker/logstream.js
@@ -1,6 +1,6 @@
// src/worker/logstream.js - batching log shipper with masking
//
-// Job output arrives as many small writes; shipping each one would be a
+// Task output arrives as many small writes; shipping each one would be a
// request per line. Output is accumulated and flushed on a timer, or as soon
// as it grows past a threshold.
//
@@ -114,7 +114,7 @@ export function createLogStream(client, url, { masked = [], flushMs = FLUSH_MS,
}
offset = result.size;
} catch (e) {
- // Losing log output must not fail an otherwise good job.
+ // Losing log output must not fail an otherwise good task.
failed = e;
logger.warn?.(`log shipping failed, dropping ${chunk.length} bytes: ${e.message}`);
}
@@ -139,7 +139,7 @@ export function createLogStream(client, url, { masked = [], flushMs = FLUSH_MS,
else schedule();
},
- // A line from the worker rather than the job, so the log can explain a
+ // A line from the worker rather than the task, so the log can explain a
// timeout, a cancellation or a setup failure.
note(line) {
this.write(`${line}\n`);
diff --git a/src/worker/script.js b/src/worker/script.js
@@ -1,11 +1,11 @@
-// src/worker/script.js - turning a job's commands into a shell script
+// src/worker/script.js - turning a task's commands into a shell script
//
// The script is copied into the container and run by path rather than
// piped to the shell's stdin, so that a command which itself reads stdin
// cannot swallow the rest of the script.
//
// It lives outside the tree, so that a repository cannot shadow it and
-// nothing the job does to its own working directory can lose it.
+// nothing the task does to its own working directory can lose it.
// Single quoting is the only form that is safe for arbitrary text in POSIX
// sh: everything inside is literal, and an embedded quote is closed,
diff --git a/src/worker/source.js b/src/worker/source.js
@@ -1,6 +1,6 @@
-// src/worker/source.js - getting the tree into the job container
+// src/worker/source.js - getting the tree into the task container
//
-// The conductor serves a tarball of the exact commit a job was created
+// The conductor serves a tarball of the exact commit a task was created
// for. The worker never sees a repository, holds no credentials for one,
// and cannot reach any ref other than the one it was given work for.
//
@@ -16,9 +16,9 @@
import { Readable } from 'node:stream';
import { createGunzip } from 'node:zlib';
-export async function sourceIntoContainer({ client, runtime, job, container, workdir }) {
- const url = job.endpoints?.source;
- if (!url) throw new Error('the job carries no source endpoint');
+export async function sourceIntoContainer({ client, runtime, task, container, workdir }) {
+ const url = task.endpoints?.source;
+ if (!url) throw new Error('the task carries no source endpoint');
const body = await client.source(url);
const stream = body instanceof Readable ? body : Readable.fromWeb(body);
diff --git a/src/worker/task.js b/src/worker/task.js
@@ -1,4 +1,4 @@
-// src/worker/job.js - running one job
+// src/worker/task.js - running one task
//
// Sequence: create the container, put the tree and the script into it,
// start any services on a private network, run it, ship its output,
@@ -9,10 +9,10 @@
// the worker needs no directory the host can also see, and therefore no
// mounts, no matching paths, and no storage of its own. What little it
// does write goes to a scratch directory that lives and dies with the
-// job.
+// task.
//
// Cleanup runs whatever happened. A worker that leaks containers or
-// networks stops being usable after a few dozen jobs.
+// networks stops being usable after a few dozen tasks.
import fs from 'node:fs/promises';
import os from 'node:os';
@@ -21,7 +21,7 @@ import { containerName } from './docker.js';
import { createLogStream } from './logstream.js';
import { sourceIntoContainer } from './source.js';
-// Where a job runs when the conductor did not say. The conductor resolves
+// Where a task runs when the conductor did not say. The conductor resolves
// this from the pipeline and the project, so this only covers a worker
// talking to one too old to have an opinion.
const FALLBACK_WORKDIR = '/work';
@@ -44,7 +44,7 @@ export function patternPrefix(pattern) {
return literal.join('/');
}
-// The set of paths to copy out for a job's artifact patterns, with any
+// The set of paths to copy out for a task's artifact patterns, with any
// path that is covered by another removed.
export function copyOutPaths(patterns) {
const prefixes = (patterns ?? []).map(patternPrefix);
@@ -54,15 +54,15 @@ export function copyOutPaths(patterns) {
return unique.filter((p, i) => !unique.slice(0, i).some((other) => p === other || p.startsWith(`${other}/`)));
}
-// Resolves the features a job asked for into local resources.
-export function resolveFeatures(job, available) {
+// Resolves the features a task asked for into local resources.
+export function resolveFeatures(task, available) {
const mounts = [];
const env = {};
const devices = [];
let privileged = false;
const missing = [];
- for (const name of job.requires ?? []) {
+ for (const name of task.requires ?? []) {
const feature = available[name];
if (!feature) {
missing.push(name);
@@ -77,17 +77,17 @@ export function resolveFeatures(job, available) {
return { mounts, env, devices, privileged, missing };
}
-export async function runJob({ cfg, client, runtime, job, logger = console }) {
- const network = containerName('conductor-net', job.id);
- const jobContainer = containerName('conductor', job.id);
- const workdir = job.workdir || FALLBACK_WORKDIR;
+export async function runTask({ cfg, client, runtime, task, logger = console }) {
+ const network = containerName('conductor-net', task.id);
+ const taskContainer = containerName('conductor', task.id);
+ const workdir = task.workdir || FALLBACK_WORKDIR;
// Worker local, and only ever written by the worker: the generated
// script on its way in, and artifacts on their way out. The container
// never sees it.
let scratch = null;
- const log = createLogStream(client, job.endpoints.log, { masked: job.masked ?? [], logger });
+ const log = createLogStream(client, task.endpoints.log, { masked: task.masked ?? [], logger });
const services = [];
let networkCreated = false;
@@ -100,15 +100,15 @@ export async function runJob({ cfg, client, runtime, job, logger = console }) {
let failure = null;
try {
- scratch = await fs.mkdtemp(path.join(os.tmpdir(), 'conductor-job-'));
+ scratch = await fs.mkdtemp(path.join(os.tmpdir(), 'conductor-task-'));
- log.note(`[conductor] job ${job.name} on ${cfg.name}`);
- log.note(`[conductor] commit ${String(job.sha).slice(0, 12)} attempt ${job.attempt}`);
- log.note(`[conductor] image ${job.image}`);
+ log.note(`[conductor] task ${task.name} on ${cfg.name}`);
+ log.note(`[conductor] commit ${String(task.sha).slice(0, 12)} attempt ${task.attempt}`);
+ log.note(`[conductor] image ${task.image}`);
- const features = resolveFeatures(job, cfg.features);
+ const features = resolveFeatures(task, cfg.features);
if (features.missing.length > 0) {
- // The conductor should never send such a job, so this means the two
+ // The conductor should never send such a task, so this means the two
// sides disagree about what this worker offers.
throw new Error(`worker does not provide required feature(s): ${features.missing.join(', ')}`);
}
@@ -116,8 +116,8 @@ export async function runJob({ cfg, client, runtime, job, logger = console }) {
await runtime.createNetwork(network);
networkCreated = true;
- for (const service of job.services ?? []) {
- const name = containerName('conductor-svc', job.id, service.alias);
+ for (const service of task.services ?? []) {
+ const name = containerName('conductor-svc', task.id, service.alias);
log.note(`[conductor] starting service ${service.image} as ${service.alias}`);
await runtime.startService({
name,
@@ -135,11 +135,11 @@ export async function runJob({ cfg, client, runtime, job, logger = console }) {
// starts and read after it exits. Feature mounts are still bind
// mounts, since those name host paths the operator chose deliberately.
await runtime.create({
- name: jobContainer,
- image: job.image,
+ name: taskContainer,
+ image: task.image,
network,
workdir,
- env: { ...features.env, ...job.env },
+ env: { ...features.env, ...task.env },
mounts: features.mounts,
devices: features.devices,
privileged: features.privileged,
@@ -148,38 +148,38 @@ export async function runJob({ cfg, client, runtime, job, logger = console }) {
});
containerCreated = true;
- await sourceIntoContainer({ client, runtime, job, container: jobContainer, workdir });
+ await sourceIntoContainer({ client, runtime, task, container: taskContainer, workdir });
const scriptPath = path.join(scratch, SCRIPT_NAME);
- await fs.writeFile(scriptPath, buildScript(job.script), { mode: 0o755 });
- await runtime.copyIn(jobContainer, scriptPath, SCRIPT_PATH);
+ await fs.writeFile(scriptPath, buildScript(task.script), { mode: 0o755 });
+ await runtime.copyIn(taskContainer, scriptPath, SCRIPT_PATH);
- const { child } = runtime.startCreated(jobContainer);
+ const { child } = runtime.startCreated(taskContainer);
child.stdout.on('data', (chunk) => log.write(chunk));
child.stderr.on('data', (chunk) => log.write(chunk));
// Report in, and find out whether the run was cancelled underneath us.
- const beatEvery = Math.max(5, Number(job.heartbeat_interval) || 30) * 1000;
+ const beatEvery = Math.max(5, Number(task.heartbeat_interval) || 30) * 1000;
heartbeat = setInterval(async () => {
try {
- const result = await client.heartbeat(job.endpoints.heartbeat);
+ const result = await client.heartbeat(task.endpoints.heartbeat);
if (result.cancelled && !cancelled) {
cancelled = true;
log.note('[conductor] cancelled, stopping the container');
- await runtime.kill(jobContainer);
+ await runtime.kill(taskContainer);
}
} catch (e) {
- logger.warn?.(`heartbeat failed for ${job.name}: ${e.message}`);
+ logger.warn?.(`heartbeat failed for ${task.name}: ${e.message}`);
}
}, beatEvery);
heartbeat.unref?.();
- const limit = (Number(job.timeout) || cfg.default_timeout) * 1000;
+ const limit = (Number(task.timeout) || cfg.default_timeout) * 1000;
timeoutTimer = setTimeout(async () => {
timedOut = true;
log.note(`[conductor] timed out after ${Math.round(limit / 1000)}s, stopping the container`);
- await runtime.kill(jobContainer);
+ await runtime.kill(taskContainer);
}, limit);
timeoutTimer.unref?.();
@@ -200,31 +200,31 @@ export async function runJob({ cfg, client, runtime, job, logger = console }) {
// Artifacts are collected before teardown, and even after a failure
// when the pipeline asked for that. They come out of the stopped
// container, which still has its filesystem until it is removed.
- if (containerCreated && job.artifacts && shouldCollect(job.artifacts.when, success)) {
+ if (containerCreated && task.artifacts && shouldCollect(task.artifacts.when, success)) {
try {
const staged = await stageArtifacts();
- const files = await collectArtifacts(staged, job.artifacts.paths, { logger });
+ const files = await collectArtifacts(staged, task.artifacts.paths, { logger });
if (files.length > 0) {
log.note(`[conductor] uploading ${files.length} artifact(s)`);
- await uploadArtifacts(client, job.endpoints.artifact, files, { logger });
+ await uploadArtifacts(client, task.endpoints.artifact, files, { logger });
} else {
log.note('[conductor] no files matched the artifact patterns');
}
} catch (e) {
log.note(`[conductor] artifact collection failed: ${e.message}`);
- logger.warn?.(`artifact collection failed for ${job.name}: ${e.message}`);
+ logger.warn?.(`artifact collection failed for ${task.name}: ${e.message}`);
}
}
const error = failure
? failure.message
: timedOut
- ? 'job timed out'
+ ? 'task timed out'
: cancelled
- ? 'job cancelled'
+ ? 'task cancelled'
: exitCode === 0
? null
- : `job exited ${exitCode}`;
+ : `task exited ${exitCode}`;
if (error) log.note(`[conductor] ${error}`);
else log.note('[conductor] done');
@@ -235,9 +235,9 @@ export async function runJob({ cfg, client, runtime, job, logger = console }) {
await cleanup();
try {
- await client.done(job.endpoints.done, { success, exitCode, error });
+ await client.done(task.endpoints.done, { success, exitCode, error });
} catch (e) {
- logger.error?.(`could not report completion of ${job.name}: ${e.message}`);
+ logger.error?.(`could not report completion of ${task.name}: ${e.message}`);
}
return { success, exitCode, error, cancelled, timedOut };
@@ -253,10 +253,10 @@ export async function runJob({ cfg, client, runtime, job, logger = console }) {
const staged = path.join(scratch, 'artifacts');
await fs.mkdir(staged, { recursive: true });
- for (const relative of copyOutPaths(job.artifacts.paths)) {
+ for (const relative of copyOutPaths(task.artifacts.paths)) {
if (relative === '') {
// The trailing /. copies the contents rather than the directory.
- await runtime.copyOut(jobContainer, `${workdir}/.`, staged);
+ await runtime.copyOut(taskContainer, `${workdir}/.`, staged);
break;
}
@@ -264,7 +264,7 @@ export async function runJob({ cfg, client, runtime, job, logger = console }) {
// unlike copying in.
const parent = path.join(staged, path.dirname(relative));
await fs.mkdir(parent, { recursive: true });
- await runtime.copyOut(jobContainer, `${workdir}/${relative}`, parent);
+ await runtime.copyOut(taskContainer, `${workdir}/${relative}`, parent);
}
return staged;
@@ -272,10 +272,10 @@ export async function runJob({ cfg, client, runtime, job, logger = console }) {
async function cleanup() {
for (const name of services) await runtime.remove(name);
- await runtime.remove(jobContainer);
+ await runtime.remove(taskContainer);
if (networkCreated) await runtime.removeNetwork(network);
- // Nothing here was written by the job, so there is no root owned
+ // Nothing here was written by the task, so there is no root owned
// output to work around: removing the container took that with it.
if (scratch) {
await fs.rm(scratch, { recursive: true, force: true })
