commit 31012e50c40db26f7667e225ad085952ab4ff358
parent 24f70da111d0df55b42d34576093b00659ba8a43
Author: finwo <finwo@pm.me>
Date: Sat, 26 Sep 2026 02:54:52 +0200
Serve task outputs, artifacts and logs as pages of their own
Diffstat:
5 files changed, 254 insertions(+), 130 deletions(-)
diff --git a/src/conductor/retention.js b/src/conductor/retention.js
@@ -183,12 +183,22 @@ export function createRetention({ cfg, db, storage, logs, logger = {} }) {
return res.changes ?? 0;
}
+ // A session nobody can present any more, expired rather than revoked, so
+ // there is a window in which it can still be listed as ended.
+ async function sweepRefreshTokens(now) {
+ const res = await db.run(
+ 'DELETE FROM refresh_tokens WHERE expires_at < {cutoff}',
+ { cutoff: now - DAY }
+ );
+ return res.changes ?? 0;
+ }
+
return {
// One pass over everything. Returns what it did, which the tests read
// and the log line summarises.
async sweep({ now = Date.now(), batch = defaults.batch } = {}) {
const total = {
- artifacts: 0, artifactBytes: 0, logs: 0, logBytes: 0, agents: 0,
+ artifacts: 0, artifactBytes: 0, logs: 0, logBytes: 0, agents: 0, sessions: 0,
};
const expired = await sweepExpiredArtifacts(now, batch);
@@ -212,11 +222,12 @@ export function createRetention({ cfg, db, storage, logs, logger = {} }) {
}
total.agents = await sweepWorkerAgents(now);
+ total.sessions = await sweepRefreshTokens(now);
- if (total.artifacts > 0 || total.logs > 0 || total.agents > 0) {
+ if (total.artifacts > 0 || total.logs > 0 || total.agents > 0 || total.sessions > 0) {
logger.info?.(
`retention: removed ${total.artifacts} artifact(s), ${total.logs} log(s) and ` +
- `${total.agents} worker record(s), ` +
+ `${total.agents} worker record(s) and ${total.sessions} session(s), ` +
`freeing ${Math.round((total.artifactBytes + total.logBytes) / 1024)} KiB`
);
}
diff --git a/src/conductor/routes/tasks.js b/src/conductor/routes/tasks.js
@@ -36,7 +36,9 @@ import { UnknownTask } from '../../lib/errors.js';
import { outputAad } from '../../lib/outputs.js';
import { limitOf, offsetOfQuery, pageBlock, paginateOffset } from '../../lib/page.js';
-export default async function taskRoutes(fastify, { cfg, db, projects, secrets, auth = null }) {
+const LOG_WINDOW = 256 * 1024;
+
+export default async function taskRoutes(fastify, { cfg, db, logs, storage, projects, secrets, auth = null }) {
const base = cfg.server.public_url.replace(/\/+$/, '');
// Explicit columns rather than t.*, which would pull spec and
@@ -45,7 +47,7 @@ export default async function taskRoutes(fastify, { cfg, db, projects, secrets,
const TASK_QUERY = `
SELECT t.id, t.name, t.base_name, t.state, t.arch, t.image,
t.attempt, t.max_attempts, t.allow_failure, t.exit_code, t.error,
- t.worker_name, t.timeout, t.spec, t.log_size,
+ t.worker_name, t.timeout, t.spec, t.log_size, t.log_key, t.log_expired_at,
t.created_at, t.started_at, t.finished_at,
j.id AS job_id, j.number AS job_number, j.ref, j.head_sha,
j.visibility, j.project_id
@@ -62,42 +64,22 @@ export default async function taskRoutes(fastify, { cfg, db, projects, secrets,
}
// Names always, values only where the task said they could be shown.
- async function readOutputs(taskId) {
- const rows = await db.all(
- 'SELECT task_id, name, value, masked FROM task_outputs WHERE task_id = {task} ORDER BY name',
- { task: taskId }
- );
- return rows.map((row) => {
- const masked = row.masked === 1;
- let value = null;
- if (!masked) {
- try {
- value = secrets.open(row.value, outputAad(row.task_id, row.name));
- } catch {
- value = null;
- }
- }
- return { name: row.name, masked, value };
- });
+ function readOutput(row) {
+ const masked = row.masked === 1;
+ if (masked) return { name: row.name, masked, value: null };
+
+ let value = null;
+ try {
+ value = secrets.open(row.value, outputAad(row.task_id, row.name));
+ } catch {
+ value = null;
+ }
+ return { name: row.name, masked, value };
}
- // Everything a caller is allowed to see, assembled once so the two routes
- // cannot drift into answering differently.
- async function represent(req, reply, row) {
+ function represent(reply, row) {
const spec = readSpec(row.spec);
- const outputs = await readOutputs(row.id);
-
- const artifacts = await paginateOffset(db, {
- select: 'id, path, size, sha256, created_at, expires_at',
- from: 'artifacts',
- where: 'task_id = {task}',
- order: 'path',
- params: { task: row.id },
- offset: offsetOfQuery(req.query),
- limit: limitOf(req.query),
- });
-
return ok(reply, {
task: {
id: row.id,
@@ -119,11 +101,10 @@ export default async function taskRoutes(fastify, { cfg, db, projects, secrets,
needs: spec.needs,
after: spec.after,
matrix: spec.matrix,
- // What the task was told to collect, which is not the same as what
- // it managed to store. The list below is what exists.
+ // What it was told to collect, not what it managed to store.
artifact_paths: spec.artifactPaths,
- outputs,
log_size: row.log_size,
+ log_expired: row.log_expired_at !== null,
created_at: row.created_at,
started_at: row.started_at,
finished_at: row.finished_at,
@@ -134,52 +115,100 @@ export default async function taskRoutes(fastify, { cfg, db, projects, secrets,
sha: row.head_sha,
visibility: row.visibility,
},
- artifacts: artifacts.rows.map((a) => ({
- id: a.id,
- path: a.path,
- size: a.size,
- sha256: a.sha256,
- created_at: a.created_at,
- expires_at: a.expires_at,
- // Ready to fetch. The download route checks visibility again for
- // itself, so this is a convenience and not the permission.
- url: `${base}/api/v1/projects/${encodeURIComponent(row.project_id)}/jobs/${row.job_id}`
- + `/tasks/${row.id}/artifacts/${a.id}`,
- })),
- }, pageBlock({ ...artifacts, url: base + req.url }));
+ });
}
- // A task id is unique by itself, so the project need not be named. Without
- // one there is no trigger secret to check against, leaving a public task
- // or a session that may manage the project.
- fastify.get('/tasks/:task', async (req, reply) => {
+ // Naming a project asserts it rather than decorating the path.
+ async function readableTask(req) {
const row = await db.get(`${TASK_QUERY} WHERE t.id = {id}`, { id: req.params.task });
if (!row) throw new UnknownTask();
+ if (req.params.project !== undefined && req.params.project !== row.project_id) throw new UnknownTask();
if (row.visibility !== 'public') {
const project = await projects.get(row.project_id);
- if (!canManageProject(await currentUser(req), project)) throw new UnknownTask();
+ if (!(await mayRead(req, project))) throw new UnknownTask();
}
+ return row;
+ }
+
+ function both(suffix, handler) {
+ fastify.get(`/tasks/:task${suffix}`, handler);
+ fastify.get(`/projects/:project/tasks/:task${suffix}`, handler);
+ }
+
+ both('', async (req, reply) => represent(reply, await readableTask(req)));
+
+ both('/outputs', async (req, reply) => {
+ const task = await readableTask(req);
+ const window = await paginateOffset(db, {
+ select: 'task_id, name, value, masked',
+ from: 'task_outputs',
+ where: 'task_id = {task}',
+ order: 'name',
+ params: { task: task.id },
+ offset: offsetOfQuery(req.query),
+ limit: limitOf(req.query),
+ });
- return represent(req, reply, row);
+ return ok(reply, window.rows.map(readOutput), pageBlock({ ...window, url: base + req.url }));
});
- // The same task named in full. The project in the path must be the one the
- // task belongs to, so a guessed id cannot be read through a project that
- // happens to be readable.
- fastify.get('/projects/:project/tasks/:task', async (req, reply) => {
- const project = await projects.get(req.params.project);
- if (!project) throw new UnknownTask();
-
- const row = await db.get(
- `${TASK_QUERY} WHERE t.id = {id} AND j.project_id = {project}`,
- { id: req.params.task, project: project.id }
- );
- if (!row) throw new UnknownTask();
+ both('/artifacts', async (req, reply) => {
+ const task = await readableTask(req);
+ const window = await paginateOffset(db, {
+ select: 'id, path, size, sha256, created_at, expires_at',
+ from: 'artifacts',
+ where: 'task_id = {task}',
+ order: 'path',
+ params: { task: task.id },
+ offset: offsetOfQuery(req.query),
+ limit: limitOf(req.query),
+ });
- if (row.visibility !== 'public' && !(await mayRead(req, project))) throw new UnknownTask();
+ const rows = window.rows.map((a) => ({
+ ...a,
+ // A convenience, not the permission: the download route checks again.
+ url: `${base}/api/v1/projects/${encodeURIComponent(task.project_id)}/jobs/${task.job_id}`
+ + `/tasks/${task.id}/artifacts/${a.id}`,
+ }));
- return represent(req, reply, row);
+ return ok(reply, rows, pageBlock({ ...window, url: base + req.url }));
+ });
+
+ both('/log', async (req, reply) => {
+ const task = await readableTask(req);
+ const offset = offsetOfQuery(req.query);
+ const limit = limitOf(req.query, LOG_WINDOW, LOG_WINDOW);
+
+ if (task.log_expired_at !== null) {
+ return ok(reply, { offset: 0, size: 0, data: '', expired: true });
+ }
+
+ // The spool only exists until the task finishes; after that the log is
+ // in storage under log_key.
+ if (!task.log_key) {
+ const chunk = await logs.read(task.job_id, task.id, { offset, limit });
+ return ok(reply, {
+ offset: chunk.offset,
+ size: chunk.size,
+ data: chunk.data.toString('utf8'),
+ expired: false,
+ });
+ }
+
+ const size = task.log_size ?? 0;
+ if (offset >= size) return ok(reply, { offset: size, size, data: '', expired: false });
+
+ const object = await storage.get(task.log_key, { range: { start: offset, end: Math.min(offset + limit, size) - 1 } });
+ const parts = [];
+ for await (const part of object.stream) parts.push(part);
+
+ return ok(reply, {
+ offset,
+ size,
+ data: Buffer.concat(parts).toString('utf8'),
+ expired: false,
+ });
});
async function mayRead(req, project) {
diff --git a/test/api.test.js b/test/api.test.js
@@ -185,13 +185,13 @@ test('a task reports its artifacts as a page', async () => {
await h.waitForJob(job);
const task = data(await h.claim({})).task;
- const res = await h.app.inject({ method: 'GET', url: `/api/v1/tasks/${task.id}` });
+ const res = await h.app.inject({ method: 'GET', url: `/api/v1/tasks/${task.id}/artifacts` });
assert.equal(res.statusCode, 200, res.body);
const body = res.json();
- assert.ok(Array.isArray(body.data.artifacts));
+ assert.ok(Array.isArray(body.data));
assert.equal(body.pagination.offset, 0);
- assert.equal(body.pagination.total, body.data.artifacts.length);
+ assert.equal(body.pagination.total, body.data.length);
});
});
diff --git a/test/conductor.test.js b/test/conductor.test.js
@@ -7,10 +7,6 @@ import test from 'node:test';
import assert from 'node:assert/strict';
import { startHarness, DEFAULT_PIPELINE, data, apiError } from './helpers/harness.js';
-// Tests that exercised the JSON read API, which has been removed. Kept
-// until the suite is rebuilt against the interface.
-const apiGone = { skip: 'the JSON API was removed' };
-
// Each test gets its own harness so that ordering never matters.
async function withHarness(options, fn) {
const h = await startHarness(options);
@@ -25,8 +21,8 @@ function tasksByName(payload) {
return Object.fromEntries(payload.tasks.map((j) => [j.name, j]));
}
-// Reads a job and its tasks straight from the database. There is no read API;
-// the interface is for people, and the tests inspect the data itself.
+// Straight from the database, so a test can assert on what the API chooses
+// not to publish.
async function jobState(h, jobId) {
const job = await h.services.db.get('SELECT * FROM jobs WHERE id = {id}', { id: jobId });
const tasks = await h.services.db.all(
@@ -443,10 +439,10 @@ test('a task lists the artifacts it stored, with a usable download url', async (
payload: Buffer.from('artifact bytes'),
});
- const res = await h.app.inject({ method: 'GET', url: `/api/v1/tasks/${task.id}` });
+ const res = await h.app.inject({ method: 'GET', url: `/api/v1/tasks/${task.id}/artifacts` });
assert.equal(res.statusCode, 200, res.body);
- const { artifacts } = data(res);
+ const artifacts = data(res);
assert.equal(artifacts.length, 1);
assert.equal(artifacts[0].path, 'dist/app');
assert.equal(artifacts[0].size, 14);
@@ -460,6 +456,86 @@ test('a task lists the artifacts it stored, with a usable download url', async (
});
});
+test('outputs are a page of their own, with masked values withheld', async () => {
+ await withHarness({}, async (h) => {
+ const tr = await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ await h.waitForJob(tr.json().data.job_id);
+ const task = await claimNamed(h, 'lint', {});
+
+ await h.services.db.run(
+ `INSERT INTO task_outputs (task_id, name, value, masked, created_at)
+ VALUES ({t}, 'version', {v}, 0, {now})`,
+ { t: task.id, v: h.services.secrets.seal('1.2.3', `task:${task.id}/output:version`), now: Date.now() }
+ );
+ await h.services.db.run(
+ `INSERT INTO task_outputs (task_id, name, value, masked, created_at)
+ VALUES ({t}, 'password', {v}, 1, {now})`,
+ { t: task.id, v: h.services.secrets.seal('hunter2', `task:${task.id}/output:password`), now: Date.now() }
+ );
+
+ const res = await h.app.inject({ method: 'GET', url: `/api/v1/tasks/${task.id}/outputs` });
+ assert.equal(res.statusCode, 200, res.body);
+
+ const outputs = Object.fromEntries(data(res).map((o) => [o.name, o]));
+ assert.equal(outputs.version.value, '1.2.3');
+ assert.equal(outputs.password.masked, true);
+ assert.equal(outputs.password.value, null);
+ assert.ok(!res.body.includes('hunter2'));
+ assert.equal(res.json().pagination.total, 2);
+ });
+});
+
+test('a log is read by byte offset and reports the total size', async () => {
+ await withHarness({}, async (h) => {
+ const tr = await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ await h.waitForJob(tr.json().data.job_id);
+ const task = await claimNamed(h, 'lint', {});
+
+ await h.app.inject({
+ method: 'POST',
+ url: `/api/v1/tasks/${task.id}/log`,
+ headers: { ...h.auth, 'content-type': 'application/octet-stream', 'x-log-offset': '0' },
+ payload: Buffer.from('hello world\n'),
+ });
+
+ const whole = data(await h.app.inject({ method: 'GET', url: `/api/v1/tasks/${task.id}/log` }));
+ assert.equal(whole.data, 'hello world\n');
+ assert.equal(whole.size, 12);
+ assert.equal(whole.offset, 0);
+ assert.equal(whole.expired, false);
+
+ // A tail asks from where it stopped and is given only what is new.
+ const tail = data(await h.app.inject({ method: 'GET', url: `/api/v1/tasks/${task.id}/log?offset=6` }));
+ assert.equal(tail.data, 'world\n');
+ assert.equal(tail.offset, 6);
+ assert.equal(tail.size, 12);
+
+ // Asking from the end is how a caller learns there is nothing new.
+ const caught = data(await h.app.inject({ method: 'GET', url: `/api/v1/tasks/${task.id}/log?offset=12` }));
+ assert.equal(caught.data, '');
+ assert.equal(caught.size, 12);
+ });
+});
+
+test('a swept log reports that it expired rather than that it is empty', async () => {
+ await withHarness({}, async (h) => {
+ const tr = await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ await h.waitForJob(tr.json().data.job_id);
+ const task = await claimNamed(h, 'lint', {});
+
+ await h.services.db.run(
+ 'UPDATE tasks SET log_expired_at = {now} WHERE id = {id}',
+ { id: task.id, now: Date.now() }
+ );
+
+ const log = data(await h.app.inject({ method: 'GET', url: `/api/v1/tasks/${task.id}/log` }));
+ assert.equal(log.expired, true);
+
+ const detail = data(await h.app.inject({ method: 'GET', url: `/api/v1/tasks/${task.id}` }));
+ assert.equal(detail.task.log_expired, true);
+ });
+});
+
test('reading a task back needs no worker token', async () => {
await withHarness({}, async (h) => {
const res = await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
@@ -497,10 +573,17 @@ test('a private task is absent without a credential, and needs its project named
assert.equal(scoped.statusCode, 200, scoped.body);
assert.equal(scoped.json().data.task.visibility, 'private');
- // Without a project in the path there is no secret to check against, so
- // a signature proves nothing here.
+ // The task names its own project, so the secret is checked either way.
const bare = await h.app.inject({ method: 'GET', url: `/api/v1/tasks/${task.id}`, headers: signed });
- assert.equal(bare.statusCode, 404, 'a signature is meaningless without a project');
+ assert.equal(bare.statusCode, 200, bare.body);
+
+ // A signature from another project is still no use.
+ const wrong = await h.app.inject({
+ method: 'GET',
+ url: `/api/v1/tasks/${task.id}`,
+ headers: { 'content-type': 'application/json', 'x-hub-signature-256': 'sha256=' + '0'.repeat(64) },
+ });
+ assert.equal(wrong.statusCode, 404);
});
});
diff --git a/test/retention.test.js b/test/retention.test.js
@@ -465,62 +465,63 @@ test('artifacts.expire in a pipeline reaches the stored artifact', async () => {
});
});
-test('the api reports and accepts project retention', { skip: 'the JSON API was removed' }, async () => {
+test('the api reports and accepts project retention', async () => {
await withHarness({
bootstrap: true,
retention: { artifact_keep_jobs: 7, artifact_keep_days: 21, log_keep_days: 9 },
}, async (h) => {
- const login = await h.app.inject({
- method: 'POST',
- url: '/api/auth/login',
- payload: { username: 'admin', password: 'bootstrap-password' },
- });
- assert.equal(login.statusCode, 200);
- const auth = { cookie: login.headers['set-cookie'].split(';')[0] };
-
- // The defaults have to be discoverable, or a null on a project is
- // meaningless to whoever is reading it.
- const defaults = await h.app.inject({ method: 'GET', url: '/api/retention', headers: auth });
- assert.equal(defaults.statusCode, 200);
- assert.deepEqual(defaults.json().defaults, {
- artifact_keep_jobs: 7,
- artifact_keep_days: 21,
- log_keep_days: 9,
+ const { headers } = await h.login();
+ const json = { ...headers, 'content-type': 'application/json' };
+
+ const patch = (payload) => h.app.inject({
+ method: 'PATCH',
+ url: `/api/v1/projects/${h.project.id}`,
+ headers: json,
+ payload: JSON.stringify(payload),
});
const before = await h.app.inject({
- method: 'GET', url: `/api/projects/${h.project.id}`, headers: auth,
+ method: 'GET', url: `/api/v1/projects/${h.project.id}`, headers,
});
- assert.equal(before.json().project.artifact_keep_days, null, 'unset means inherited');
+ assert.equal(before.json().data.artifact_keep_days, null, 'unset means inherited');
- const patched = await h.app.inject({
- method: 'PATCH',
- url: `/api/projects/${h.project.id}`,
- headers: auth,
- payload: { artifact_keep_jobs: 3, artifact_keep_days: 0, log_keep_days: 30 },
- });
- assert.equal(patched.statusCode, 200);
- assert.equal(patched.json().project.artifact_keep_jobs, 3);
- assert.equal(patched.json().project.artifact_keep_days, 0, 'zero is a value, not an absence');
- assert.equal(patched.json().project.log_keep_days, 30);
+ const patched = await patch({ artifact_keep_jobs: 3, artifact_keep_days: 0, log_keep_days: 30 });
+ assert.equal(patched.statusCode, 200, patched.body);
+ assert.equal(patched.json().data.artifact_keep_jobs, 3);
+ assert.equal(patched.json().data.artifact_keep_days, 0, 'zero is a value, not an absence');
+ assert.equal(patched.json().data.log_keep_days, 30);
// And back to following the server default.
- const cleared = await h.app.inject({
- method: 'PATCH',
- url: `/api/projects/${h.project.id}`,
- headers: auth,
- payload: { artifact_keep_jobs: null },
- });
- assert.equal(cleared.json().project.artifact_keep_jobs, null);
+ const cleared = await patch({ artifact_keep_jobs: null });
+ assert.equal(cleared.json().data.artifact_keep_jobs, null);
- const rejected = await h.app.inject({
- method: 'PATCH',
- url: `/api/projects/${h.project.id}`,
- headers: auth,
- payload: { log_keep_days: -5 },
- });
+ const rejected = await patch({ log_keep_days: -5 });
assert.equal(rejected.statusCode, 400);
- assert.match(rejected.json().error, /zero or more/);
+ assert.match(rejected.json().error.message, /zero or more/);
+ });
+});
+
+test('a session outlives its expiry only until the next sweep', async () => {
+ await withHarness({ bootstrap: true }, async (h) => {
+ const grant = await h.login();
+ const live = await h.services.refreshTokens.page({ offset: 0, limit: 10 });
+ assert.equal(live.total, 1);
+
+ // Still within the grace window, so it can be reported as ended.
+ await h.services.db.run(
+ 'UPDATE refresh_tokens SET expires_at = {then}',
+ { then: Date.now() - 1000 }
+ );
+ assert.equal((await h.services.retention.sweep()).sessions, 0);
+
+ const refused = await h.token({ grant_type: 'refresh_token', refresh_token: grant.refresh_token });
+ assert.equal(refused.statusCode, 400);
+
+ await h.services.db.run(
+ 'UPDATE refresh_tokens SET expires_at = {then}',
+ { then: Date.now() - 3 * 24 * 60 * 60 * 1000 }
+ );
+ assert.equal((await h.services.retention.sweep()).sessions, 1);
});
});
