commit 10b8758dd218edef0becaf1da5e9ecb4545e8350
parent 24d7089bae258cfd6b9386c0fc2408cccb4cb67c
Author: finwo <finwo@pm.me>
Date: Sat, 19 Sep 2026 18:18:07 +0200
Run migrations and retention against real postgres and mysql
Diffstat:
2 files changed, 300 insertions(+), 0 deletions(-)
diff --git a/test/dialects.test.js b/test/dialects.test.js
@@ -0,0 +1,132 @@
+// test/dialects.test.js - the same behaviour on postgres and mysql
+//
+// Everything else runs on sqlite, so the other two dialects were only
+// exercised by reading them. This applies every migration against a real
+// server and then runs the retention sweep, which is the query that most
+// depends on the dialect: it joins, filters on nulls, orders, and binds a
+// LIMIT, and the drivers do not all accept those the same way.
+//
+// Skipped when docker is unavailable, or when the optional drivers are
+// not installed.
+
+import test 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 { startDatabase, dialectAvailable } from './helpers/database.js';
+import { openDatabase, runMigrations, dialectFromUrl } from '../src/lib/db/index.js';
+import { createStorage } from '../src/lib/storage/index.js';
+import { createLogStore } from '../src/lib/log.js';
+import { createRetention } from '../src/conductor/retention.js';
+
+const DAY = 24 * 60 * 60 * 1000;
+const NOW = 1_800_000_000_000;
+
+// The same shape the sqlite tests use, so the expectations below can be
+// compared against what that dialect already proves.
+async function seed(db, storage) {
+ await db.run(
+ `INSERT INTO projects (id, name, repo_url, enabled, run_counter, created_at, updated_at)
+ VALUES ('demo', 'Demo', '/tmp/repo', 1, 0, {now}, {now})`,
+ { now: NOW }
+ );
+
+ for (let i = 1; i <= 5; i += 1) {
+ // Run 5 is recent; the rest are long past.
+ const at = NOW - (i === 5 ? 1 : 100) * DAY;
+ await db.run(
+ `INSERT INTO runs (id, project_id, number, head_sha, trigger_type, state,
+ visibility, created_at, started_at, finished_at)
+ VALUES ({id}, 'demo', {number}, {sha}, 'push', 'failed', 'public', {at}, {at}, {at})`,
+ { id: `r${i}`, number: i, sha: 'a'.repeat(40), at }
+ );
+ await db.run(
+ `INSERT INTO jobs (id, run_id, name, base_name, image, requires, spec, state,
+ allow_failure, attempt, max_attempts, timeout,
+ log_key, log_size, created_at, finished_at)
+ VALUES ({id}, {run}, 'build', 'build', 'alpine:3', '[]', '{}', 'failed',
+ 0, 1, 1, 3600, {logKey}, 10, {at}, {at})`,
+ { id: `r${i}:build`, run: `r${i}`, logKey: `logs/r${i}`, at }
+ );
+ await db.run(
+ `INSERT INTO artifacts (id, job_id, run_id, path, storage_key, size, sha256,
+ created_at, expires_at)
+ VALUES ({id}, {job}, {run}, 'out.txt', {key}, 5, {sha}, {at}, {expires})`,
+ {
+ id: `a${i}`,
+ job: `r${i}:build`,
+ run: `r${i}`,
+ key: `art/r${i}`,
+ sha: 'b'.repeat(64),
+ at,
+ // Run 1 carries an explicit deadline, which nothing protects.
+ expires: i === 1 ? NOW - 1 : null,
+ }
+ );
+ await storage.put(`art/r${i}`, Buffer.from('artifact'));
+ await storage.put(`logs/r${i}`, Buffer.from('log'));
+ }
+}
+
+for (const dialect of ['postgres', 'mysql']) {
+ test(`migrations apply and retention sweeps on ${dialect}`, { skip: !dialectAvailable(dialect) }, async () => {
+ const server = await startDatabase(dialect);
+ const root = await fs.mkdtemp(path.join(os.tmpdir(), `conductor-${dialect}-`));
+
+ const cfg = {
+ database: { url: server.url, dialect: dialectFromUrl(server.url) },
+ storage: { path: path.join(root, 'storage') },
+ log: { spool_path: path.join(root, 'logs'), max_size: 1 << 20 },
+ retention: {
+ artifact_keep_runs: 2,
+ artifact_keep_days: 7,
+ log_keep_days: 7,
+ sweep_interval: 3600,
+ batch: 2,
+ },
+ };
+
+ let db;
+ try {
+ db = await openDatabase(cfg);
+ await runMigrations(db, { logger: () => {} });
+
+ const storage = await createStorage(cfg);
+ const logs = createLogStore(cfg);
+ await seed(db, storage);
+
+ const retention = createRetention({ cfg, db, storage, logs, logger: {} });
+
+ const first = await retention.sweep({ now: NOW });
+
+ // Run 1 by its own deadline, runs 2 and 3 by policy. Runs 4 and 5
+ // are the last two, so they are kept whatever their age.
+ assert.equal(first.artifacts, 3, 'three artifacts should go on the first pass');
+
+ const left = (await db.all('SELECT id FROM artifacts ORDER BY id', {})).map((r) => r.id);
+ assert.deepEqual(left, ['a4', 'a5']);
+
+ // Logs are batched, so the four old ones take two passes.
+ assert.equal(first.logs, 2, 'the batch limit should hold on this dialect too');
+ const second = await retention.sweep({ now: NOW });
+ assert.equal(second.artifacts, 0, 'nothing further to delete');
+ assert.equal(second.logs, 2);
+
+ const third = await retention.sweep({ now: NOW });
+ assert.equal(third.logs, 0, 'a swept log must not be found again');
+
+ const withLogs = await db.all('SELECT id FROM jobs WHERE log_key IS NOT NULL ORDER BY id', {});
+ assert.deepEqual(withLogs.map((r) => r.id), ['r5:build'], 'only the recent run keeps its log');
+
+ // The jobs themselves are history and stay.
+ const jobs = await db.get('SELECT COUNT(*) AS c FROM jobs', {});
+ assert.equal(Number(jobs.c), 5, 'sweeping a log must not remove the job');
+ } finally {
+ await db?.close().catch(() => {});
+ await fs.rm(root, { recursive: true, force: true });
+ await server.stop();
+ }
+ });
+}
diff --git a/test/helpers/database.js b/test/helpers/database.js
@@ -0,0 +1,168 @@
+// test/helpers/database.js - real postgres and mysql for the tests
+//
+// The suite otherwise runs entirely on sqlite, which means the postgres
+// and mysql migrations and the dialect specific query compilation were
+// only ever read, never executed. A migration that does not apply, or a
+// query that binds its parameters in a way one driver dislikes, would
+// reach whoever runs that database rather than the test suite.
+//
+// Set CONDUCTOR_TEST_POSTGRES_URL or CONDUCTOR_TEST_MYSQL_URL to use a
+// server you already have 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 { openDatabase, dialectFromUrl } from '../../src/lib/db/index.js';
+import { reapStale } from './containers.js';
+
+const execFileAsync = promisify(execFile);
+
+// Pinned, so an upstream release cannot change what is being tested
+// against without the change being visible here.
+export const IMAGES = {
+ postgres: 'postgres:16-alpine',
+ mysql: 'mysql:8.4',
+};
+
+const READY_TIMEOUT = 180 * 1000;
+const NAME_PREFIX = 'conductor-db-test-';
+const PASSWORD = 'conductortestpassword';
+
+// 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;
+ }
+}
+
+// Whether a given dialect can be exercised: a url was supplied, or docker
+// is here to start one. The drivers are optional dependencies, so a
+// checkout that skipped them cannot run these either.
+export function dialectAvailable(dialect) {
+ if (envUrl(dialect)) return true;
+ if (!dockerAvailableSync()) return false;
+ try {
+ // require rather than import, to stay synchronous for the same reason.
+ import.meta.resolve(dialect === 'postgres' ? 'pg' : 'mysql2');
+ return true;
+ } catch {
+ return false;
+ }
+}
+
+function envUrl(dialect) {
+ return dialect === 'postgres'
+ ? process.env.CONDUCTOR_TEST_POSTGRES_URL
+ : process.env.CONDUCTOR_TEST_MYSQL_URL;
+}
+
+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));
+ });
+ });
+}
+
+// Whether the server is actually usable yet.
+//
+// Not pg_isready or mysqladmin ping over docker exec, which is the
+// obvious choice and the wrong one: both images run a temporary server
+// during initialisation to create the database, and those clients answer
+// for it over the local socket. The probe then passes, the temporary
+// server shuts down, and the first real query dies with ECONNRESET.
+//
+// Both images keep that temporary server off the network, so connecting
+// over the published port and running a statement is the thing that
+// cannot be satisfied early.
+async function probe(url) {
+ let db;
+ try {
+ db = await openDatabase({ database: { url, dialect: dialectFromUrl(url) } });
+ await db.get('SELECT 1 AS ok', {});
+ return true;
+ } catch {
+ return false;
+ } finally {
+ await db?.close().catch(() => {});
+ }
+}
+
+// Starts one, returning its url and a stop function. Each caller gets its
+// own database so tests cannot see each other's rows.
+export async function startDatabase(dialect) {
+ const supplied = envUrl(dialect);
+ if (supplied) {
+ return { url: supplied, async stop() {} };
+ }
+
+ await reapStale(NAME_PREFIX);
+
+ const name = `${NAME_PREFIX}${dialect}-${crypto.randomBytes(4).toString('hex')}`;
+ const port = await freePort();
+ const inner = dialect === 'postgres' ? 5432 : 3306;
+
+ const env = dialect === 'postgres'
+ ? [
+ '--env', 'POSTGRES_USER=conductor',
+ '--env', `POSTGRES_PASSWORD=${PASSWORD}`,
+ '--env', 'POSTGRES_DB=conductor',
+ ]
+ : [
+ '--env', `MYSQL_ROOT_PASSWORD=${PASSWORD}`,
+ '--env', 'MYSQL_DATABASE=conductor',
+ ];
+
+ await execFileAsync('docker', [
+ 'run', '--detach', '--rm',
+ '--name', name,
+ '--publish', `127.0.0.1:${port}:${inner}`,
+ // Nothing here outlives the test, so durability is only a cost.
+ '--tmpfs', dialect === 'postgres' ? '/var/lib/postgresql/data' : '/var/lib/mysql',
+ ...env,
+ IMAGES[dialect],
+ ], { timeout: 300000 });
+
+ const stop = async () => {
+ await execFileAsync('docker', ['rm', '-f', name], { timeout: 60000 }).catch(() => {});
+ };
+
+ const url = dialect === 'postgres'
+ ? `postgres://conductor:${PASSWORD}@127.0.0.1:${port}/conductor`
+ : `mysql://root:${PASSWORD}@127.0.0.1:${port}/conductor`;
+
+ const deadline = Date.now() + READY_TIMEOUT;
+ let ready = false;
+ while (Date.now() < deadline) {
+ if (await probe(url)) {
+ ready = true;
+ break;
+ }
+ await new Promise((r) => { setTimeout(r, 500); });
+ }
+
+ if (!ready) {
+ const diagnosis = await execFileAsync('docker', ['logs', '--tail', '20', name], { timeout: 10000 })
+ .then(({ stdout, stderr }) => `${stdout}${stderr}`.trim())
+ .catch((e) => `could not read container logs: ${e.message}`);
+ await stop();
+ throw new Error(`${dialect} did not become ready within ${READY_TIMEOUT / 1000}s\n${diagnosis}`);
+ }
+
+ return { url, stop };
+}
