commit e60515fff046bad256797d6b04d16ee15c244205
parent ec84d4a080f05c2053c7b9a2a40686f8b010016d
Author: finwo <finwo@pm.me>
Date: Sat, 19 Sep 2026 14:14:06 +0200
Stream large artifacts to S3 with multipart upload
Diffstat:
8 files changed, 994 insertions(+), 102 deletions(-)
diff --git a/README.md b/README.md
@@ -206,13 +206,28 @@ Testing
npm test
```
-Tests that need a container skip themselves when docker is unavailable.
-With docker present the suite starts what it needs and cleans up after
-itself: real containers for the worker to run jobs in, and a real OIDC
-provider to authenticate against. Nothing is stubbed, because the failures
-worth catching are in the parts a stub would replace.
+That is the whole setup. With docker available the suite starts what it
+needs and cleans up after itself:
-S3 coverage is opt in, since it needs a server to point at:
+| Dependency | Used for |
+| ------------------- | -------------------------------------------- |
+| `alpine` | containers for the worker to actually run jobs in |
+| MinIO | object storage, artifacts and archived logs |
+| mock-oauth2-server | a real OIDC provider to authenticate against |
+
+Nothing is stubbed, because the failures worth catching live in the parts a
+stub would replace. Hand written SigV4 signing is only meaningfully tested
+by a server that rejects a bad signature, and the JWKS path was wrong
+against every provider except the one it was written for.
+
+Without docker those tests skip and the rest still run:
+
+```sh
+CONDUCTOR_TEST_NO_DOCKER=1 npm test
+```
+
+To test against a different object store, such as Garage or AWS, point the
+same tests at it instead of starting MinIO:
```sh
docker run -d -p 9000:9000 -e MINIO_ROOT_USER=testkey \
diff --git a/docs/pipeline.md b/docs/pipeline.md
@@ -222,6 +222,13 @@ A bare list is shorthand for `paths`. Paths are relative to the workspace;
absolute paths are rejected. `when` is `on_success`, `on_failure` or
`always`.
+Artifacts are streamed rather than buffered, so size is bounded by where
+they are stored rather than by the conductor's memory. Object storage
+receives anything over 32 MiB as a multipart upload, which keeps memory
+flat and lifts the five gigabyte limit that applies to a single request.
+Disk images and other multi gigabyte output are fine; what they cost is
+storage, not resident memory.
+
Cache
-----
diff --git a/src/lib/storage/s3.js b/src/lib/storage/s3.js
@@ -4,16 +4,84 @@
// and MinIO, which both need path style addressing; set force_path_style to
// false for AWS S3 proper.
//
-// Requests go out over fetch with SigV4 headers. Uploads of a known Buffer
-// are signed over their content hash; streamed uploads use UNSIGNED-PAYLOAD,
-// which every S3 implementation accepts and which avoids buffering an entire
-// artifact in memory just to hash it.
+// Requests go out over fetch with SigV4 headers.
+//
+// Large uploads go through multipart, one part at a time. Handing a stream
+// to fetch does not stream it: undici collects the whole body before
+// sending, which was measured at just over a gigabyte of live heap for a
+// one gigabyte artifact. Build output is routinely that size, so a single
+// upload could exhaust the conductor. Multipart also lifts the five
+// gigabyte ceiling on a single PUT, and lets an upload proceed when the
+// total length is not known in advance.
+//
+// Small bodies still go as one request, since three round trips to save
+// buffering a few megabytes is a poor trade.
import crypto from 'node:crypto';
import { Readable } from 'node:stream';
-import { signRequest, presignUrl, sha256Hex, uriEncode, UNSIGNED_PAYLOAD } from './sigv4.js';
+import { signRequest, presignUrl, sha256Hex, uriEncode } from './sigv4.js';
import { StorageNotFound, assertKey } from './key.js';
+// Replays what has already been read, then continues with the rest.
+async function* concatSources(head, rest) {
+ yield head;
+ yield* rest;
+}
+
+// S3 requires every part except the last to be at least 5 MiB, and allows
+// at most 10000 parts. The default part size is comfortably above the
+// minimum, and is scaled up when the total is large enough that 10000
+// parts would not cover it.
+const MIN_PART_SIZE = 5 * 1024 * 1024;
+const DEFAULT_PART_SIZE = 16 * 1024 * 1024;
+const MAX_PARTS = 10000;
+
+// Below this a stream is collected and sent as one request.
+const SINGLE_PUT_MAX = 32 * 1024 * 1024;
+
+export function partSizeFor(size) {
+ if (!Number.isFinite(size) || size <= 0) return DEFAULT_PART_SIZE;
+ const needed = Math.ceil(size / (MAX_PARTS - 1));
+ const rounded = Math.ceil(needed / (1024 * 1024)) * 1024 * 1024;
+ return Math.max(DEFAULT_PART_SIZE, MIN_PART_SIZE, rounded);
+}
+
+// Regroups an arbitrarily chunked stream into buffers of a fixed size, so
+// memory stays bounded by the part size no matter how the source behaves.
+export async function* inParts(source, partSize) {
+ let held = [];
+ let heldLength = 0;
+
+ for await (const chunk of source) {
+ const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
+ held.push(buffer);
+ heldLength += buffer.length;
+
+ while (heldLength >= partSize) {
+ const joined = Buffer.concat(held, heldLength);
+ yield joined.subarray(0, partSize);
+ const rest = joined.subarray(partSize);
+ held = rest.length > 0 ? [rest] : [];
+ heldLength = rest.length;
+ }
+ }
+
+ if (heldLength > 0) yield Buffer.concat(held, heldLength);
+}
+
+// Values inside S3 XML are hex digests and quotes, but escaping keeps a
+// surprising key or ETag from breaking the document.
+function xmlEscape(value) {
+ return String(value).replace(/[<>&'"]/g, (c) => (
+ { '<': '<', '>': '>', '&': '&', "'": ''', '"': '"' }[c]
+ ));
+}
+
+function firstTag(xml, tag) {
+ const match = new RegExp(`<${tag}>([\\s\\S]*?)</${tag}>`).exec(xml);
+ return match ? match[1] : null;
+}
+
export function createS3Storage(cfg) {
const s3 = cfg.storage.s3;
const endpoint = new URL(s3.endpoint);
@@ -42,8 +110,12 @@ export function createS3Storage(cfg) {
return url;
}
- async function send(method, key, { headers = {}, body, payloadHash } = {}) {
+ async function send(method, key, { headers = {}, body, payloadHash, query } = {}) {
const url = objectUrl(key);
+ // Query parameters are part of the signature, so they must be set
+ // before signing and not touched afterwards.
+ for (const [name, value] of Object.entries(query ?? {})) url.searchParams.set(name, value);
+
const signed = signRequest({
method,
url,
@@ -53,11 +125,7 @@ export function createS3Storage(cfg) {
});
const init = { method, headers: signed };
- if (body !== undefined) {
- init.body = body;
- // Required by undici when streaming a request body.
- if (body instanceof ReadableStream) init.duplex = 'half';
- }
+ if (body !== undefined) init.body = body;
return fetch(url, init);
}
@@ -71,42 +139,159 @@ export function createS3Storage(cfg) {
return new Error(`s3 ${action} failed for ${key}: ${res.status} ${res.statusText}\n${detail}`);
}
+ // One request, for a body already in memory.
+ async function putBuffer(key, buffer, contentType) {
+ const digest = crypto.createHash('sha256').update(buffer).digest('hex');
+ const headers = { 'content-length': String(buffer.length) };
+ if (contentType) headers['content-type'] = contentType;
+
+ const res = await send('PUT', key, { headers, body: buffer, payloadHash: digest });
+ if (!res.ok) throw await errorFrom(res, 'put', key);
+ await res.arrayBuffer();
+
+ return { key, size: buffer.length, sha256: digest };
+ }
+
+ async function createMultipart(key, contentType) {
+ const headers = {};
+ if (contentType) headers['content-type'] = contentType;
+
+ const res = await send('POST', key, { headers, query: { uploads: '' }, body: undefined });
+ if (!res.ok) throw await errorFrom(res, 'create multipart upload', key);
+
+ const uploadId = firstTag(await res.text(), 'UploadId');
+ if (!uploadId) throw new Error(`s3 did not return an upload id for ${key}`);
+ return uploadId;
+ }
+
+ async function uploadPart(key, uploadId, partNumber, buffer) {
+ const res = await send('PUT', key, {
+ headers: { 'content-length': String(buffer.length) },
+ query: { partNumber: String(partNumber), uploadId },
+ body: buffer,
+ payloadHash: crypto.createHash('sha256').update(buffer).digest('hex'),
+ });
+ if (!res.ok) throw await errorFrom(res, `upload part ${partNumber}`, key);
+ await res.arrayBuffer();
+
+ const etag = res.headers.get('etag');
+ if (!etag) throw new Error(`s3 did not return an ETag for part ${partNumber} of ${key}`);
+ return etag;
+ }
+
+ async function completeMultipart(key, uploadId, parts) {
+ const body = Buffer.from(
+ '<CompleteMultipartUpload>' +
+ parts.map((p) => `<Part><PartNumber>${p.number}</PartNumber><ETag>${xmlEscape(p.etag)}</ETag></Part>`).join('') +
+ '</CompleteMultipartUpload>',
+ 'utf8'
+ );
+
+ const res = await send('POST', key, {
+ headers: { 'content-length': String(body.length), 'content-type': 'application/xml' },
+ query: { uploadId },
+ body,
+ payloadHash: crypto.createHash('sha256').update(body).digest('hex'),
+ });
+ if (!res.ok) throw await errorFrom(res, 'complete multipart upload', key);
+
+ // S3 can report failure inside a 200 response, because the connection
+ // is held open while the parts are assembled.
+ const text = await res.text();
+ if (/<Error>/.test(text)) {
+ const code = firstTag(text, 'Code') ?? 'unknown';
+ throw new Error(`s3 complete multipart upload failed for ${key}: ${code}`);
+ }
+ }
+
+ async function abortMultipart(key, uploadId) {
+ // Parts left behind are billable and invisible, so this matters even
+ // though nothing depends on its result.
+ const res = await send('DELETE', key, { query: { uploadId } });
+ await res.arrayBuffer().catch(() => {});
+ }
+
+ async function putMultipart(key, source, { contentType, size }) {
+ const uploadId = await createMultipart(key, contentType);
+ const partSize = partSizeFor(size);
+ const hash = crypto.createHash('sha256');
+ const parts = [];
+ let total = 0;
+
+ // Parts go up one at a time. Overlapping them was measured at about
+ // eight percent faster while doubling peak memory, which is not a
+ // trade worth the extra failure handling.
+ try {
+ for await (const part of inParts(source, partSize)) {
+ hash.update(part);
+ total += part.length;
+ if (parts.length >= MAX_PARTS) {
+ throw new Error(`s3 upload of ${key} exceeded ${MAX_PARTS} parts`);
+ }
+ const etag = await uploadPart(key, uploadId, parts.length + 1, part);
+ parts.push({ number: parts.length + 1, etag });
+ }
+
+ // A multipart upload must have at least one part.
+ if (parts.length === 0) {
+ parts.push({ number: 1, etag: await uploadPart(key, uploadId, 1, Buffer.alloc(0)) });
+ }
+
+ if (size !== undefined && size !== total) {
+ throw new Error(`s3 put size mismatch for ${key}: declared ${size}, read ${total}`);
+ }
+
+ await completeMultipart(key, uploadId, parts);
+ } catch (e) {
+ await abortMultipart(key, uploadId).catch(() => {});
+ throw e;
+ }
+
+ return { key, size: total, sha256: hash.digest('hex') };
+ }
+
return {
driver: 's3',
async put(key, body, opts = {}) {
- const headers = {};
- if (opts.contentType) headers['content-type'] = opts.contentType;
-
- let payload;
- let payloadHash;
- let size = opts.size;
- let sha256 = opts.sha256;
+ assertKey(key);
if (Buffer.isBuffer(body) || typeof body === 'string') {
- const buf = Buffer.isBuffer(body) ? body : Buffer.from(body, 'utf8');
- payload = buf;
- sha256 = crypto.createHash('sha256').update(buf).digest('hex');
- payloadHash = sha256;
- size = buf.length;
- headers['content-length'] = String(size);
- } else {
- if (size === undefined) {
- throw new Error(
- `s3 put of a stream needs an explicit size for ${key}; ` +
- 'buffer the object first if the length is unknown'
- );
- }
- payloadHash = UNSIGNED_PAYLOAD;
- headers['content-length'] = String(size);
- payload = body instanceof Readable ? Readable.toWeb(body) : body;
+ return putBuffer(key, Buffer.isBuffer(body) ? body : Buffer.from(body, 'utf8'), opts.contentType);
}
- const res = await send('PUT', key, { headers, body: payload, payloadHash });
- if (!res.ok) throw await errorFrom(res, 'put', key);
- await res.arrayBuffer();
+ const source = body instanceof Readable ? body : Readable.from(body);
+
+ // Small and of known length: collect it and send one request rather
+ // than paying for three.
+ if (opts.size !== undefined && opts.size <= SINGLE_PUT_MAX) {
+ const chunks = [];
+ let length = 0;
+ for await (const chunk of source) {
+ const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
+ chunks.push(buffer);
+ length += buffer.length;
+ if (length > SINGLE_PUT_MAX) break;
+ }
+
+ if (length <= SINGLE_PUT_MAX) {
+ const buffer = Buffer.concat(chunks, length);
+ if (opts.size !== buffer.length) {
+ throw new Error(`s3 put size mismatch for ${key}: declared ${opts.size}, read ${buffer.length}`);
+ }
+ return putBuffer(key, buffer, opts.contentType);
+ }
+
+ // It was larger than it claimed; carry on as multipart without
+ // losing what has already been read.
+ const head = Buffer.concat(chunks, length);
+ return putMultipart(key, concatSources(head, source), {
+ contentType: opts.contentType,
+ size: opts.size,
+ });
+ }
- return { key, size, sha256: sha256 ?? null };
+ return putMultipart(key, source, { contentType: opts.contentType, size: opts.size });
},
async get(key, opts = {}) {
diff --git a/test/conductor-s3.test.js b/test/conductor-s3.test.js
@@ -0,0 +1,246 @@
+// 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);
+ });
+});
diff --git a/test/conductor.test.js b/test/conductor.test.js
@@ -332,6 +332,25 @@ test('log appends are resumable and deduplicated by offset', async () => {
});
});
+test('an oversized log chunk is refused', async () => {
+ await withHarness({}, async (h) => {
+ await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
+ const job = await claimNamed(h, 'lint', {});
+
+ // Artifacts are streamed and may be large, but a single log append is
+ // buffered, so it has to be bounded or a worker can exhaust memory.
+ const res = await h.app.inject({
+ method: 'POST',
+ url: `/api/workers/jobs/${encodeURIComponent(job.id)}/log`,
+ headers: { ...h.auth, 'content-type': 'application/octet-stream' },
+ payload: Buffer.alloc(2 * 1024 * 1024, 0x41),
+ });
+
+ assert.equal(res.statusCode, 413);
+ assert.equal(await h.services.logs.size(job.run_id, job.id), 0, 'nothing should have been written');
+ });
+});
+
test('a finished log moves to storage and is still readable', async () => {
await withHarness({}, async (h) => {
await h.trigger({ sha: h.sha, ref: 'refs/heads/main' });
diff --git a/test/helpers/harness.js b/test/helpers/harness.js
@@ -74,6 +74,19 @@ export async function startHarness(options = {}) {
' path: ./state/conductor.db',
'storage:',
' path: ./state/storage',
+ // Given an s3 block, artifacts and finished logs go there instead of
+ // to the local path above.
+ ...(options.s3
+ ? [
+ ' s3:',
+ ` endpoint: ${options.s3.endpoint}`,
+ ` region: ${options.s3.region}`,
+ ` bucket: ${options.s3.bucket}`,
+ ` access_key_id: ${options.s3.access_key_id}`,
+ ` secret_access_key: ${options.s3.secret_access_key}`,
+ ` force_path_style: ${options.s3.force_path_style !== false}`,
+ ]
+ : []),
'git:',
' mirror_path: ./state/mirrors',
' fetch_interval: 0',
diff --git a/test/helpers/minio.js b/test/helpers/minio.js
@@ -0,0 +1,179 @@
+// test/helpers/minio.js - a real object store for the tests
+//
+// Runs MinIO, which speaks real S3. The signing code here is written
+// against node:crypto rather than taken from a library, so verifying it
+// against a real server is the only way to know it works: a stub would
+// accept whatever signature we happened to produce, including a wrong one.
+//
+// Set CONDUCTOR_TEST_S3_ENDPOINT to point the same tests at something else,
+// such as Garage or AWS, instead of starting a container.
+
+import { execFile, execFileSync } from 'node:child_process';
+import { promisify } from 'node:util';
+import crypto from 'node:crypto';
+import net from 'node:net';
+import { signRequest, sha256Hex } from '../../src/lib/storage/sigv4.js';
+
+const execFileAsync = promisify(execFile);
+
+// Pinned, so a MinIO release cannot change what the suite is testing
+// against without the change being visible here.
+export const IMAGE = 'quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z';
+const NAME_PREFIX = 'conductor-minio-test-';
+
+const ACCESS_KEY = 'conductortest';
+const SECRET_KEY = 'conductortestsecret';
+
+// Checked synchronously: the runner needs to know whether to skip while it
+// is collecting tests, and an unsettled top-level await at that point
+// aborts the whole file.
+export function dockerAvailableSync() {
+ if (process.env.CONDUCTOR_TEST_NO_DOCKER) return false;
+ try {
+ execFileSync('docker', ['version', '--format', '{{.Server.Version}}'], {
+ timeout: 15000,
+ stdio: 'ignore',
+ });
+ return true;
+ } catch {
+ return false;
+ }
+}
+
+// True when the suite can exercise S3 at all: either an endpoint was given,
+// or docker is here to start one.
+export function s3Available() {
+ return Boolean(process.env.CONDUCTOR_TEST_S3_ENDPOINT) || dockerAvailableSync();
+}
+
+async function freePort() {
+ return new Promise((resolve, reject) => {
+ const server = net.createServer();
+ server.unref();
+ server.on('error', reject);
+ server.listen(0, '127.0.0.1', () => {
+ const { port } = server.address();
+ server.close(() => resolve(port));
+ });
+ });
+}
+
+// A run that crashes never reaches its cleanup hook. Anything still
+// carrying the prefix is from a previous run and can go.
+async function reapStale() {
+ try {
+ const { stdout } = await execFileAsync('docker', ['ps', '-aq', '--filter', `name=${NAME_PREFIX}`], { timeout: 30000 });
+ const ids = stdout.trim().split('\n').filter(Boolean);
+ if (ids.length > 0) await execFileAsync('docker', ['rm', '-f', ...ids], { timeout: 60000 });
+ } catch {
+ // Best effort; a failure here must not fail the tests.
+ }
+}
+
+// Buckets are created with a signed request rather than the MinIO client,
+// which keeps the helper free of another dependency and exercises the
+// signer on the way in.
+export async function createBucket(endpoint, bucket, credentials) {
+ const url = new URL(`${endpoint.replace(/\/+$/, '')}/${bucket}`);
+ const headers = signRequest({
+ method: 'PUT',
+ url,
+ payloadHash: sha256Hex(''),
+ region: credentials.region,
+ accessKeyId: credentials.accessKeyId,
+ secretAccessKey: credentials.secretAccessKey,
+ });
+
+ const res = await fetch(url, { method: 'PUT', headers });
+ // 409 means it already exists, which is fine.
+ if (!res.ok && res.status !== 409) {
+ throw new Error(`could not create bucket ${bucket}: ${res.status} ${await res.text()}`);
+ }
+}
+
+export async function startMinio() {
+ const external = process.env.CONDUCTOR_TEST_S3_ENDPOINT;
+
+ // An endpoint supplied by the operator wins, and is left alone.
+ if (external) {
+ const credentials = {
+ region: process.env.CONDUCTOR_TEST_S3_REGION || 'us-east-1',
+ accessKeyId: process.env.CONDUCTOR_TEST_S3_KEY,
+ secretAccessKey: process.env.CONDUCTOR_TEST_S3_SECRET,
+ };
+ return {
+ endpoint: external.replace(/\/+$/, ''),
+ credentials,
+ managed: false,
+ async bucket(name = `citest${Date.now().toString(36)}`) {
+ await createBucket(this.endpoint, name, credentials);
+ return name;
+ },
+ async stop() {},
+ };
+ }
+
+ await reapStale();
+
+ const port = await freePort();
+ const name = `${NAME_PREFIX}${crypto.randomBytes(4).toString('hex')}`;
+ const endpoint = `http://127.0.0.1:${port}`;
+ const credentials = { region: 'us-east-1', accessKeyId: ACCESS_KEY, secretAccessKey: SECRET_KEY };
+
+ await execFileAsync('docker', [
+ 'run', '--detach', '--rm',
+ '--name', name,
+ '--publish', `${port}:9000`,
+ '--env', `MINIO_ROOT_USER=${ACCESS_KEY}`,
+ '--env', `MINIO_ROOT_PASSWORD=${SECRET_KEY}`,
+ IMAGE, 'server', '/data',
+ ], { timeout: 180000 });
+
+ let ready = false;
+ for (let i = 0; i < 60; i += 1) {
+ try {
+ const res = await fetch(`${endpoint}/minio/health/live`, { signal: AbortSignal.timeout(2000) });
+ if (res.ok) {
+ ready = true;
+ break;
+ }
+ } catch {
+ // Still starting.
+ }
+ await new Promise((r) => { setTimeout(r, 500).unref?.(); });
+ }
+
+ if (!ready) {
+ await execFileAsync('docker', ['rm', '-f', name]).catch(() => {});
+ throw new Error(`MinIO did not become ready at ${endpoint}`);
+ }
+
+ return {
+ endpoint,
+ credentials,
+ name,
+ managed: true,
+
+ // A fresh bucket per test keeps them independent.
+ async bucket(bucketName = `citest${crypto.randomBytes(6).toString('hex')}`) {
+ await createBucket(endpoint, bucketName, credentials);
+ return bucketName;
+ },
+
+ // Configuration block for createStorage and for the conductor.
+ storageConfig(bucketName) {
+ return {
+ endpoint,
+ region: credentials.region,
+ bucket: bucketName,
+ access_key_id: credentials.accessKeyId,
+ secret_access_key: credentials.secretAccessKey,
+ force_path_style: true,
+ };
+ },
+
+ async stop() {
+ await execFileAsync('docker', ['rm', '-f', name], { timeout: 60000 }).catch(() => {});
+ },
+ };
+}
diff --git a/test/storage.test.js b/test/storage.test.js
@@ -1,22 +1,25 @@
// test/storage.test.js - object storage backends
//
-// The local backend is always exercised. The S3 backend runs against a live
-// server only when CONDUCTOR_TEST_S3_ENDPOINT is set, for example:
+// The local backend is always exercised. The S3 backend runs against a real
+// MinIO, started automatically when docker is available, because the
+// signing code is written by hand here and a stub would happily accept a
+// signature a real server rejects.
//
-// docker run -d -p 9000:9000 -e MINIO_ROOT_USER=testkey \
-// -e MINIO_ROOT_PASSWORD=testsecret123 quay.io/minio/minio server /data
-// CONDUCTOR_TEST_S3_ENDPOINT=http://127.0.0.1:9000 \
-// CONDUCTOR_TEST_S3_KEY=testkey CONDUCTOR_TEST_S3_SECRET=testsecret123 \
-// npm test
+// Point the same tests at another implementation, such as Garage or AWS,
+// with CONDUCTOR_TEST_S3_ENDPOINT, CONDUCTOR_TEST_S3_KEY and
+// CONDUCTOR_TEST_S3_SECRET.
-import test from 'node:test';
+import test, { before, after } from 'node:test';
import assert from 'node:assert/strict';
import fs from 'node:fs/promises';
import os from 'node:os';
import path from 'node:path';
import { Readable } from 'node:stream';
+import crypto from 'node:crypto';
import { createStorage, keys, StorageNotFound, assertKey, sanitizeRelativePath } from '../src/lib/storage/index.js';
-import { signRequest, sha256Hex } from '../src/lib/storage/sigv4.js';
+import { sha256Hex, signRequest } from '../src/lib/storage/sigv4.js';
+import { partSizeFor, inParts } from '../src/lib/storage/s3.js';
+import { startMinio, s3Available } from './helpers/minio.js';
async function drain(stream) {
let out = '';
@@ -119,75 +122,300 @@ test('a partial write leaves no object behind', async () => {
}
});
-// S3 coverage, opt in.
-const S3_ENDPOINT = process.env.CONDUCTOR_TEST_S3_ENDPOINT;
-const s3Options = { skip: S3_ENDPOINT ? false : 'set CONDUCTOR_TEST_S3_ENDPOINT to run S3 tests' };
+// --- multipart mechanics, no server needed ---
-test('s3 backend round trips awkward keys against a live server', s3Options, async () => {
- const bucket = `citest${Date.now().toString(36)}`;
- const creds = {
- region: 'us-east-1',
- accessKeyId: process.env.CONDUCTOR_TEST_S3_KEY,
- secretAccessKey: process.env.CONDUCTOR_TEST_S3_SECRET,
- };
+test('part size grows so that a huge object still fits in 10000 parts', () => {
+ const MiB = 1024 * 1024;
- const bucketUrl = new URL(`${S3_ENDPOINT.replace(/\/+$/, '')}/${bucket}`);
- const created = await fetch(bucketUrl, {
- method: 'PUT',
- headers: signRequest({ method: 'PUT', url: bucketUrl, payloadHash: sha256Hex(''), ...creds }),
- });
- assert.ok(created.ok, `failed to create bucket: ${created.status}`);
+ // Small and unknown sizes use the default.
+ assert.equal(partSizeFor(undefined), 16 * MiB);
+ assert.equal(partSizeFor(0), 16 * MiB);
+ assert.equal(partSizeFor(100 * MiB), 16 * MiB);
- const st = createStorage({
- storage: {
- driver: 's3',
- s3: {
- endpoint: S3_ENDPOINT,
- bucket,
- region: creds.region,
- access_key_id: creds.accessKeyId,
- secret_access_key: creds.secretAccessKey,
- force_path_style: true,
- },
- },
- });
+ // 10000 parts of 16 MiB covers about 156 GiB; past that the part size
+ // has to grow or the upload cannot be expressed.
+ const huge = 1024 * 1024 * MiB; // 1 TiB
+ const size = partSizeFor(huge);
+ assert.ok(size > 16 * MiB, 'part size should scale up');
+ assert.ok(Math.ceil(huge / size) <= 10000, 'must fit within the part limit');
+ assert.equal(size % MiB, 0, 'part size should stay a whole number of MiB');
+});
+
+test('inParts regroups an awkwardly chunked stream into fixed parts', async () => {
+ const source = Readable.from([
+ Buffer.from('aaa'), Buffer.from('bb'), Buffer.from('cccccc'), Buffer.from('d'),
+ ]);
- // Spaces and plus signs are exactly what a naive signer gets wrong.
+ const parts = [];
+ for await (const part of inParts(source, 4)) parts.push(part.toString());
+
+ // Everything is preserved, in order, in fixed sized pieces with a
+ // remainder at the end.
+ assert.deepEqual(parts, ['aaab', 'bccc', 'cccd']);
+ assert.equal(parts.join(''), 'aaabbccccccd');
+});
+
+test('inParts handles a chunk far larger than the part size', async () => {
+ const source = Readable.from([Buffer.alloc(10, 0x41)]);
+ const parts = [];
+ for await (const part of inParts(source, 4)) parts.push(part.length);
+ assert.deepEqual(parts, [4, 4, 2]);
+});
+
+test('inParts yields nothing for an empty stream', async () => {
+ const parts = [];
+ for await (const part of inParts(Readable.from([]), 4)) parts.push(part);
+ assert.deepEqual(parts, []);
+});
+
+// S3 coverage against a real server.
+const s3Options = { 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 storage handle onto a fresh bucket.
+async function s3Storage() {
+ const bucket = await minio.bucket();
+ return createStorage({ storage: { driver: 's3', s3: minio.storageConfig(bucket) } });
+}
+
+test('s3 backend round trips awkward keys against a real server', s3Options, async () => {
+ const st = await s3Storage();
+
+ // Spaces, plus signs and parentheses are exactly what a naive signer
+ // gets wrong, and what a stubbed server would never catch.
const key = 'artifacts/r1/r1:build/dist/app bin+v1(final)!.tar.gz';
const put = await st.put(key, Buffer.from('hello from s3'), { contentType: 'application/gzip' });
assert.equal(put.size, 13);
+ assert.equal(put.sha256, sha256Hex(Buffer.from('hello from s3')));
assert.equal(await drain((await st.get(key)).stream), 'hello from s3');
- assert.equal(await drain((await st.get(key, { range: { start: 6, end: 9 } })).stream), 'from');
assert.equal((await st.head(key)).size, 13);
+});
+
+test('s3 serves byte ranges', s3Options, async () => {
+ const st = await s3Storage();
+ await st.put('r.txt', Buffer.from('hello conductor'));
+
+ assert.equal(await drain((await st.get('r.txt', { range: { start: 6, end: 14 } })).stream), 'conductor');
+ assert.equal(await drain((await st.get('r.txt', { range: { start: 6 } })).stream), 'conductor');
+});
+
+test('s3 accepts a stream when the size is known', s3Options, async () => {
+ const st = await s3Storage();
+ const body = Readable.from([Buffer.from('streamed '), Buffer.from('upload')]);
+ const put = await st.put('logs/r1/j.log', body, { size: 15 });
- const streamed = await st.put('logs/r1/j.log', Readable.from([Buffer.from('streamed '), Buffer.from('upload')]), { size: 15 });
- assert.equal(streamed.size, 15);
+ assert.equal(put.size, 15);
+ assert.equal(await drain((await st.get('logs/r1/j.log')).stream), 'streamed upload');
+});
+
+test('a presigned url is usable without credentials', s3Options, async () => {
+ const st = await s3Storage();
+ const key = 'artifacts/run/job/dist/app bin+v1.tar.gz';
+ await st.put(key, Buffer.from('presigned content'));
+
+ const url = await st.presign(key, { expires: 300 });
+ assert.match(url, /X-Amz-Signature=[0-9a-f]{64}/);
+
+ // Plain fetch, no Authorization header.
+ const res = await fetch(url);
+ assert.equal(res.status, 200);
+ assert.equal(await res.text(), 'presigned content');
+});
- const presigned = await fetch(await st.presign(key, { expires: 300 }));
- assert.equal(presigned.status, 200);
- assert.equal(await presigned.text(), 'hello from s3');
+test('a tampered presigned url is rejected by the server', s3Options, async () => {
+ const st = await s3Storage();
+ await st.put('secret.txt', Buffer.from('do not serve this'));
+ const url = new URL(await st.presign('secret.txt', { expires: 300 }));
+ const signature = url.searchParams.get('X-Amz-Signature');
+ // Flip a character of the signature.
+ url.searchParams.set('X-Amz-Signature', `${signature.slice(0, -1)}${signature.endsWith('a') ? 'b' : 'a'}`);
+
+ const res = await fetch(url);
+ assert.ok(!res.ok, 'a forged signature must not be honoured');
+});
+
+test('s3 reports a missing object and deletes idempotently', s3Options, async () => {
+ const st = await s3Storage();
await assert.rejects(st.get('missing/object'), (e) => e instanceof StorageNotFound);
+ assert.equal(await st.head('missing/object'), null);
+
+ await st.put('gone.txt', Buffer.from('x'));
+ await st.delete('gone.txt');
+ assert.equal(await st.head('gone.txt'), null);
+ await st.delete('gone.txt');
+});
+
+test('a stream of unknown length is accepted', s3Options, async () => {
+ const st = await s3Storage();
+
+ // Multipart does not need the total up front, so a body whose length is
+ // not known in advance no longer has to be buffered by the caller.
+ const put = await st.put('unknown.bin', Readable.from([
+ Buffer.from('one '), Buffer.from('two '), Buffer.from('three'),
+ ]));
- await st.delete(key);
- assert.equal(await st.head(key), null);
- await st.delete(key);
+ assert.equal(put.size, 13);
+ assert.equal(await drain((await st.get('unknown.bin')).stream), 'one two three');
});
-test('s3 put of a stream without a size is refused', s3Options, async () => {
+test('a body past the single request threshold is uploaded in parts', s3Options, async () => {
+ const st = await s3Storage();
+
+ // Above SINGLE_PUT_MAX, so this exercises create, several uploads and
+ // complete rather than one PUT.
+ const size = 40 * 1024 * 1024;
+ const pattern = Buffer.alloc(1024 * 1024);
+ for (let i = 0; i < pattern.length; i += 4) pattern.writeUInt32BE(i, i);
+
+ const expected = crypto.createHash('sha256');
+ let produced = 0;
+ const source = new Readable({
+ read() {
+ if (produced >= size) return this.push(null);
+ // Vary each block so a part written out of order would be visible.
+ const block = Buffer.from(pattern);
+ block.writeUInt32BE(produced / 1024 / 1024, 0);
+ produced += block.length;
+ expected.update(block);
+ this.push(block);
+ },
+ });
+
+ const put = await st.put('big/multipart.bin', source, { size });
+ assert.equal(put.size, size);
+ assert.equal(put.sha256, expected.digest('hex'), 'the assembled object must match what was sent');
+
+ // Read it back independently rather than trusting the reported digest.
+ const back = crypto.createHash('sha256');
+ let read = 0;
+ for await (const chunk of (await st.get('big/multipart.bin')).stream) {
+ back.update(chunk);
+ read += chunk.length;
+ }
+ assert.equal(read, size);
+ assert.equal(back.digest('hex'), put.sha256);
+});
+
+test('a large body really is sent as several parts, and a small one is not', s3Options, async () => {
+ const bucket = await minio.bucket();
+ const st = createStorage({ storage: { driver: 's3', s3: minio.storageConfig(bucket) } });
+
+ // A multipart ETag is the digest of the part digests with the part
+ // count appended, so it says plainly how the object was uploaded. That
+ // makes it a reliable check that streaming is still in use, rather than
+ // inferring it from timing or memory.
+ async function etagOf(key) {
+ const url = new URL(`${minio.endpoint}/${bucket}/${key}`);
+ const headers = signRequest({
+ method: 'HEAD',
+ url,
+ payloadHash: sha256Hex(''),
+ region: minio.credentials.region,
+ accessKeyId: minio.credentials.accessKeyId,
+ secretAccessKey: minio.credentials.secretAccessKey,
+ });
+ const res = await fetch(url, { method: 'HEAD', headers });
+ assert.ok(res.ok, `HEAD ${key} failed: ${res.status}`);
+ return res.headers.get('etag').replace(/"/g, '');
+ }
+
+ // 40 MiB at a 16 MiB part size is three parts.
+ const size = 40 * 1024 * 1024;
+ const source = Readable.from((function* () {
+ for (let sent = 0; sent < size; sent += 1024 * 1024) yield Buffer.alloc(1024 * 1024, 0x43);
+ })());
+ await st.put('big/parted.bin', source, { size });
+
+ const multipart = await etagOf('big/parted.bin');
+ assert.match(multipart, /-3$/, `expected a three part upload, got ETag ${multipart}`);
+
+ // Something small must not pay for three round trips.
+ await st.put('small/plain.bin', Buffer.alloc(1024, 0x44));
+ const single = await etagOf('small/plain.bin');
+ assert.ok(!/-\d+$/.test(single), `expected a single request upload, got ETag ${single}`);
+});
+
+test('a multipart object still serves ranges', s3Options, async () => {
+ const st = await s3Storage();
+ const size = 40 * 1024 * 1024;
+
+ const source = Readable.from((function* () {
+ for (let sent = 0; sent < size; sent += 1024 * 1024) {
+ const block = Buffer.alloc(1024 * 1024);
+ block.writeUInt32BE(sent, 0);
+ yield block;
+ }
+ })());
+
+ await st.put('big/ranged.bin', source, { size });
+
+ // A boundary between parts is the interesting place to read across.
+ const start = 16 * 1024 * 1024 - 2;
+ const chunk = await st.get('big/ranged.bin', { range: { start, end: start + 5 } });
+ const bytes = [];
+ for await (const part of chunk.stream) bytes.push(part);
+ assert.equal(Buffer.concat(bytes).length, 6);
+});
+
+test('a body that does not match its declared size is rejected', s3Options, async () => {
+ const st = await s3Storage();
+ const size = 40 * 1024 * 1024;
+
+ // Claims 40 MiB, sends 20 MiB.
+ const short = Readable.from((function* () {
+ for (let sent = 0; sent < size / 2; sent += 1024 * 1024) yield Buffer.alloc(1024 * 1024, 0x41);
+ })());
+
+ await assert.rejects(st.put('big/short.bin', short, { size }), /size mismatch/);
+ assert.equal(await st.head('big/short.bin'), null, 'no partial object should be visible');
+});
+
+test('a failed multipart upload does not leave parts behind', s3Options, async () => {
+ const bucket = await minio.bucket();
+ const st = createStorage({ storage: { driver: 's3', s3: minio.storageConfig(bucket) } });
+ const size = 40 * 1024 * 1024;
+
+ const short = Readable.from((function* () {
+ for (let sent = 0; sent < size / 2; sent += 1024 * 1024) yield Buffer.alloc(1024 * 1024, 0x42);
+ })());
+ await assert.rejects(st.put('big/abandoned.bin', short, { size }), /size mismatch/);
+
+ // Abandoned parts are billable and invisible, so the upload must have
+ // been aborted rather than simply dropped.
+ const url = new URL(`${minio.endpoint}/${bucket}`);
+ url.searchParams.set('uploads', '');
+ const headers = signRequest({
+ method: 'GET',
+ url,
+ payloadHash: sha256Hex(''),
+ region: minio.credentials.region,
+ accessKeyId: minio.credentials.accessKeyId,
+ secretAccessKey: minio.credentials.secretAccessKey,
+ });
+
+ const listing = await (await fetch(url, { headers })).text();
+ assert.ok(!listing.includes('<Upload>'), `an upload was left pending:\n${listing}`);
+});
+
+test('bad credentials fail loudly rather than silently storing nothing', s3Options, async () => {
+ const bucket = await minio.bucket();
const st = createStorage({
storage: {
driver: 's3',
- s3: {
- endpoint: S3_ENDPOINT,
- bucket: 'irrelevant',
- region: 'us-east-1',
- access_key_id: 'k',
- secret_access_key: 's',
- force_path_style: true,
- },
+ s3: { ...minio.storageConfig(bucket), secret_access_key: 'wrong-secret' },
},
});
- await assert.rejects(st.put('k', Readable.from(['x'])), /needs an explicit size/);
+
+ await assert.rejects(st.put('k.txt', Buffer.from('x')), /s3 put failed/);
});
