commit 4fd94b10dcba620030e498bfcd1f09488e321bb1
parent ddd5a60e50eeaed261604f455636d05e08fad1ca
Author: finwo <finwo@pm.me>
Date: Sun, 27 Sep 2026 19:54:01 +0200
Restore the object storage suite against the task API
Diffstat:
2 files changed, 208 insertions(+), 246 deletions(-)
diff --git a/test/conductor-s3.test.js b/test/conductor-s3.test.js
@@ -0,0 +1,208 @@
+// test/conductor-s3.test.js - the conductor running on object storage
+
+import test, { before, after } from 'node:test';
+import assert from 'node:assert/strict';
+import fs from 'node:fs/promises';
+import { startHarness, data } from './helpers/harness.js';
+import { startMinio, s3Available } from './helpers/minio.js';
+import { createStorage, keys } from '../src/lib/storage/index.js';
+
+const opts = { skip: s3Available() ? false : 'docker is not available' };
+
+let minio = null;
+
+before(async () => {
+ if (s3Available()) minio = await startMinio();
+}, { timeout: 240000 });
+
+after(async () => {
+ if (minio) await minio.stop();
+});
+
+async function withS3Conductor(options, fn) {
+ const bucket = await minio.bucket();
+ const s3 = minio.storageConfig(bucket);
+ const h = await startHarness({ visibility: 'public', s3, ...options });
+ const store = createStorage({ storage: { driver: 's3', s3 } });
+
+ try {
+ return await fn(h, { store, bucket });
+ } finally {
+ await h.stop();
+ }
+}
+
+const octet = (h) => ({ ...h.auth, 'content-type': 'application/octet-stream' });
+
+async function drain(stream) {
+ let out = '';
+ for await (const chunk of stream) out += chunk;
+ return out;
+}
+
+async function taskWithOutput(h, { log = 'building\ndone\n', artifact = 'artifact bytes' } = {}) {
+ const triggered = await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const jobId = data(triggered).job_id;
+ await h.waitForJob(jobId);
+
+ const claimed = await h.claim({ arches: ['x86_64'] });
+ assert.equal(claimed.statusCode, 200, claimed.body);
+ const task = data(claimed).task;
+
+ await h.app.inject({
+ method: 'POST',
+ url: `/api/v1/tasks/${encodeURIComponent(task.id)}/log`,
+ headers: octet(h),
+ payload: Buffer.from(log),
+ });
+
+ if (artifact !== null) {
+ const upload = await h.app.inject({
+ method: 'POST',
+ url: `/api/v1/tasks/${encodeURIComponent(task.id)}/artifacts`,
+ headers: { ...octet(h), 'x-artifact-path': 'dist/app bin+v1.tar.gz' },
+ payload: Buffer.from(artifact),
+ });
+ assert.equal(upload.statusCode, 200, upload.body);
+ }
+
+ return { jobId, task };
+}
+
+function complete(h, taskId) {
+ return h.app.inject({
+ method: 'POST',
+ url: `/api/v1/tasks/${encodeURIComponent(taskId)}/complete`,
+ headers: { ...h.auth, 'content-type': 'application/json' },
+ payload: JSON.stringify({ state: 'success', success: true, exit_code: 0 }),
+ });
+}
+
+function readLog(h, taskId, query = '') {
+ return h.app.inject({ method: 'GET', url: `/api/v1/tasks/${encodeURIComponent(taskId)}/log${query}` });
+}
+
+test('the conductor reports object storage when configured for it', opts, async () => {
+ await withS3Conductor({}, async (h) => {
+ const health = await h.app.inject({ method: 'GET', url: '/health' });
+ assert.equal(data(health).storage, 's3');
+ });
+});
+
+test('an artifact uploaded by a worker really lands in the bucket', opts, async () => {
+ await withS3Conductor({}, async (h, { store }) => {
+ const { jobId, task } = await taskWithOutput(h);
+
+ const listed = await h.app.inject({
+ method: 'GET', url: `/api/v1/tasks/${encodeURIComponent(task.id)}/artifacts`,
+ });
+ const artifact = data(listed)[0];
+ assert.equal(artifact.path, 'dist/app bin+v1.tar.gz');
+
+ const key = keys.artifact(jobId, task.id, artifact.path);
+ const stored = await store.get(key);
+ assert.equal(await drain(stored.stream), 'artifact bytes');
+ assert.equal((await store.head(key)).size, artifact.size);
+ });
+});
+
+test('an artifact download is handed off with a redirect that works', opts, async () => {
+ await withS3Conductor({}, async (h) => {
+ const { jobId, task } = await taskWithOutput(h);
+ const listed = await h.app.inject({
+ method: 'GET', url: `/api/v1/tasks/${encodeURIComponent(task.id)}/artifacts`,
+ });
+ const artifact = data(listed)[0];
+
+ const res = await h.app.inject({
+ method: 'GET',
+ url: `/api/v1/projects/demo/jobs/${jobId}/tasks/${encodeURIComponent(task.id)}`
+ + `/artifacts/${artifact.id}`,
+ });
+ assert.equal(res.statusCode, 302, res.body);
+ assert.match(res.headers.location, /X-Amz-Signature=/);
+
+ const followed = await fetch(res.headers.location);
+ assert.equal(followed.status, 200);
+ assert.equal(await followed.text(), 'artifact bytes');
+ });
+});
+
+test('a finished log is moved to the bucket and stays readable', opts, async () => {
+ await withS3Conductor({}, async (h, { store }) => {
+ const { jobId, task } = await taskWithOutput(h, { log: 'compiling\nlinking\n' });
+
+ const live = await readLog(h, task.id);
+ assert.equal(data(live).data, 'compiling\nlinking\n');
+
+ assert.equal((await complete(h, task.id)).statusCode, 200);
+
+ const archived = await readLog(h, task.id);
+ assert.equal(data(archived).data, 'compiling\nlinking\n');
+
+ const row = await h.services.db.get('SELECT log_key FROM tasks WHERE id = {id}', { id: task.id });
+ assert.equal(row.log_key, keys.log(jobId, task.id));
+
+ const stored = await store.get(row.log_key);
+ assert.equal(await drain(stored.stream), 'compiling\nlinking\n');
+ });
+});
+
+test('the local spool is released once the log is archived', opts, async () => {
+ await withS3Conductor({}, async (h) => {
+ const { jobId, task } = await taskWithOutput(h, { log: 'output\n', artifact: null });
+
+ const spool = h.services.logs.spoolPath(jobId, task.id);
+ assert.ok(await fs.stat(spool).catch(() => null), 'the spool should exist while running');
+
+ await complete(h, task.id);
+
+ assert.equal(await fs.stat(spool).catch(() => null), null, 'the spool should be gone');
+ });
+});
+
+test('an archived log still serves incremental ranges', opts, async () => {
+ await withS3Conductor({}, async (h) => {
+ const { task } = await taskWithOutput(h, { log: 'first line\nsecond line\n', artifact: null });
+ await complete(h, task.id);
+
+ const tail = await readLog(h, task.id, '?offset=11');
+ assert.equal(data(tail).data, 'second line\n');
+ });
+});
+
+test('a private job in a bucket is not readable by a stranger', opts, async () => {
+ await withS3Conductor({ visibility: 'private' }, async (h) => {
+ const { jobId, task } = await taskWithOutput(h);
+
+ const detail = await h.app.inject({ method: 'GET', url: `/api/v1/jobs/${jobId}` });
+ assert.equal(detail.statusCode, 404);
+
+ const log = await readLog(h, task.id);
+ assert.equal(log.statusCode, 404);
+ });
+});
+
+test('a large artifact streams through to the bucket', opts, async () => {
+ await withS3Conductor({}, async (h, { store }) => {
+ const triggered = await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const jobId = data(triggered).job_id;
+ await h.waitForJob(jobId);
+
+ const claimed = await h.claim({ arches: ['x86_64'] });
+ const task = data(claimed).task;
+ const payload = Buffer.alloc(3 * 1024 * 1024, 0x41);
+
+ const upload = await h.app.inject({
+ method: 'POST',
+ url: `/api/v1/tasks/${encodeURIComponent(task.id)}/artifacts`,
+ headers: { ...octet(h), 'x-artifact-path': 'dist/big.bin' },
+ payload,
+ });
+ assert.equal(upload.statusCode, 200, upload.body);
+ assert.equal(data(upload).size, payload.length);
+
+ const stored = await store.head(keys.artifact(jobId, task.id, 'dist/big.bin'));
+ assert.equal(stored.size, payload.length);
+ });
+});
diff --git a/test/conductor-s3.test.js.disabled b/test/conductor-s3.test.js.disabled
@@ -1,246 +0,0 @@
-// test/conductor-s3.test.js - the conductor running on object storage
-//
-// The storage module is covered on its own elsewhere. What is checked here
-// is the conductor configured to use it: that artifacts actually land in
-// the bucket, that a finished log is moved there and remains readable, and
-// that a download is handed off with a redirect instead of being proxied.
-//
-// That handoff only happens on S3, since the local backend has no URL to
-// give out, so none of it is exercised by the default configuration.
-
-import test, { before, after } from 'node:test';
-import assert from 'node:assert/strict';
-import fs from 'node:fs/promises';
-import { startHarness } from './helpers/harness.js';
-import { startMinio, s3Available } from './helpers/minio.js';
-import { createStorage, keys } from '../src/lib/storage/index.js';
-
-const opts = { skip: s3Available() ? false : 'docker is not available' };
-
-let minio = null;
-
-before(async () => {
- if (s3Available()) minio = await startMinio();
-}, { timeout: 240000 });
-
-after(async () => {
- if (minio) await minio.stop();
-});
-
-// A conductor whose storage is a fresh bucket, plus a handle on that
-// bucket so a test can look at what actually landed in it.
-async function withS3Conductor(options, fn) {
- const bucket = await minio.bucket();
- const s3 = minio.storageConfig(bucket);
- const h = await startHarness({ visibility: 'public', s3, ...options });
- const store = createStorage({ storage: { driver: 's3', s3 } });
-
- try {
- return await fn(h, { store, bucket });
- } finally {
- await h.stop();
- }
-}
-
-const octet = (h) => ({ ...h.auth, 'content-type': 'application/octet-stream' });
-
-async function drain(stream) {
- let out = '';
- for await (const chunk of stream) out += chunk;
- return out;
-}
-
-// Runs a job far enough to have a log and an artifact.
-async function jobWithOutput(h, { log = 'building\ndone\n', artifact = 'artifact bytes' } = {}) {
- await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
- const job = (await h.poll({ arches: 'x86_64' })).json().job;
-
- await h.app.inject({
- method: 'POST',
- url: `/api/workers/jobs/${encodeURIComponent(job.id)}/log`,
- headers: octet(h),
- payload: Buffer.from(log),
- });
-
- if (artifact !== null) {
- await h.app.inject({
- method: 'POST',
- url: `/api/workers/jobs/${encodeURIComponent(job.id)}/artifact`,
- headers: { ...octet(h), 'x-artifact-path': 'dist/app bin+v1.tar.gz' },
- payload: Buffer.from(artifact),
- });
- }
-
- return job;
-}
-
-test('the conductor reports object storage when configured for it', opts, async () => {
- await withS3Conductor({}, async (h) => {
- const health = (await h.app.inject({ method: 'GET', url: '/health' })).json();
- assert.equal(health.storage, 's3');
- });
-});
-
-test('an artifact uploaded by a worker really lands in the bucket', opts, async () => {
- await withS3Conductor({}, async (h, { store }) => {
- const job = await jobWithOutput(h);
-
- const detail = await h.app.inject({ method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}` });
- const artifact = detail.json().artifacts[0];
- assert.equal(artifact.path, 'dist/app bin+v1.tar.gz');
-
- // Read it straight out of the bucket, not through the conductor.
- const key = keys.artifact(job.run_id, job.id, artifact.path);
- const stored = await store.get(key);
- assert.equal(await drain(stored.stream), 'artifact bytes');
- assert.equal((await store.head(key)).size, artifact.size);
- });
-});
-
-test('an artifact download is handed off with a redirect that works', opts, async () => {
- await withS3Conductor({}, async (h) => {
- const job = await jobWithOutput(h);
- const detail = await h.app.inject({ method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}` });
- const artifact = detail.json().artifacts[0];
-
- const res = await h.app.inject({ method: 'GET', url: `/api/artifacts/${artifact.id}` });
-
- // With local storage this streams a 200; on S3 the client is sent to
- // the store so the conductor does not carry the bytes.
- assert.equal(res.statusCode, 302);
- const location = res.headers.location;
- assert.match(location, /X-Amz-Signature=/);
-
- // The presigned URL has to be usable as given, with no credentials.
- const followed = await fetch(location);
- assert.equal(followed.status, 200);
- assert.equal(await followed.text(), 'artifact bytes');
- });
-});
-
-test('a finished log is moved to the bucket and stays readable', opts, async () => {
- await withS3Conductor({}, async (h, { store }) => {
- const job = await jobWithOutput(h, { log: 'compiling\nlinking\n' });
-
- // While running it is served from the local spool.
- const live = await h.app.inject({ method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}/log` });
- assert.equal(live.headers['x-log-complete'], 'false');
- assert.equal(live.body, 'compiling\nlinking\n');
-
- await h.app.inject({
- method: 'POST',
- url: `/api/workers/jobs/${encodeURIComponent(job.id)}/done`,
- headers: { ...h.auth, 'content-type': 'application/json' },
- payload: JSON.stringify({ success: true, exit_code: 0 }),
- });
-
- // Afterwards it comes from the bucket, and reads the same.
- const archived = await h.app.inject({ method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}/log` });
- assert.equal(archived.headers['x-log-complete'], 'true');
- assert.equal(archived.body, 'compiling\nlinking\n');
-
- const stored = await store.get(keys.log(job.run_id, job.id));
- assert.equal(await drain(stored.stream), 'compiling\nlinking\n');
- });
-});
-
-test('the local spool is released once the log is archived', opts, async () => {
- await withS3Conductor({}, async (h) => {
- const job = await jobWithOutput(h, { log: 'output\n', artifact: null });
-
- const spool = h.services.logs.spoolPath(job.run_id, job.id);
- assert.ok(await fs.stat(spool).catch(() => null), 'the spool should exist while running');
-
- await h.app.inject({
- method: 'POST',
- url: `/api/workers/jobs/${encodeURIComponent(job.id)}/done`,
- headers: { ...h.auth, 'content-type': 'application/json' },
- payload: JSON.stringify({ success: true, exit_code: 0 }),
- });
-
- // Storage is the durable copy; keeping the spool too would grow
- // without bound.
- assert.equal(await fs.stat(spool).catch(() => null), null, 'the spool should be gone');
- });
-});
-
-test('an archived log still serves incremental ranges', opts, async () => {
- await withS3Conductor({}, async (h) => {
- const job = await jobWithOutput(h, { log: 'first line\nsecond line\n', artifact: null });
-
- await h.app.inject({
- method: 'POST',
- url: `/api/workers/jobs/${encodeURIComponent(job.id)}/done`,
- headers: { ...h.auth, 'content-type': 'application/json' },
- payload: JSON.stringify({ success: true, exit_code: 0 }),
- });
-
- const tail = await h.app.inject({
- method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}/log?offset=11`,
- });
- assert.equal(tail.body, 'second line\n');
- });
-});
-
-test('the interface renders a log that lives in the bucket', opts, async () => {
- await withS3Conductor({}, async (h) => {
- const job = await jobWithOutput(h, { log: 'rendered from s3\n', artifact: null });
-
- await h.app.inject({
- method: 'POST',
- url: `/api/workers/jobs/${encodeURIComponent(job.id)}/done`,
- headers: { ...h.auth, 'content-type': 'application/json' },
- payload: JSON.stringify({ success: true, exit_code: 0 }),
- });
-
- const page = await h.app.inject({ method: 'GET', url: `/jobs/${encodeURIComponent(job.id)}` });
- assert.equal(page.statusCode, 200);
- assert.match(page.body, /rendered from s3/);
- });
-});
-
-test('a private run in a bucket is still not readable by a stranger', opts, async () => {
- await withS3Conductor({ visibility: 'private' }, async (h) => {
- const job = await jobWithOutput(h);
- const detail = await h.app.inject({
- method: 'GET',
- url: `/api/jobs/${encodeURIComponent(job.id)}`,
- headers: await h.login().catch(() => ({})),
- });
-
- // Anonymous access is refused before any URL is ever signed, so the
- // object store is never asked and no presigned link is handed out.
- const anonymous = await h.app.inject({ method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}` });
- assert.equal(anonymous.statusCode, 404);
-
- const anonymousLog = await h.app.inject({
- method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}/log`,
- });
- assert.equal(anonymousLog.statusCode, 404);
- assert.equal(detail.statusCode, 404);
- });
-});
-
-test('a large artifact streams through to the bucket', opts, async () => {
- await withS3Conductor({}, async (h, { store }) => {
- await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
- const job = (await h.poll({ arches: 'x86_64' })).json().job;
-
- // Comfortably past the server's body limit for parsed payloads, which
- // an artifact must not be subject to: build output is routinely far
- // larger than a JSON request.
- const payload = Buffer.alloc(3 * 1024 * 1024, 0x41);
-
- const upload = await h.app.inject({
- method: 'POST',
- url: `/api/workers/jobs/${encodeURIComponent(job.id)}/artifact`,
- headers: { ...octet(h), 'x-artifact-path': 'dist/big.bin' },
- payload,
- });
- assert.equal(upload.statusCode, 200, upload.body);
- assert.equal(upload.json().size, payload.length);
-
- const stored = await store.head(keys.artifact(job.run_id, job.id, 'dist/big.bin'));
- assert.equal(stored.size, payload.length);
- });
-});
