dialects.test.js (11169B)
1 // test/dialects.test.js - the same behaviour on postgres and mysql 2 // 3 // Everything else jobs on sqlite, so the other two dialects were only 4 // exercised by reading them. This applies every migration against a real 5 // server and then jobs the retention sweep, which is the query that most 6 // depends on the dialect: it joins, filters on nulls, orders, and binds a 7 // LIMIT, and the drivers do not all accept those the same way. Paging is 8 // checked here for the same reason, since it binds an OFFSET as well. 9 // 10 // Skipped when docker is unavailable, or when the optional drivers are 11 // not installed. 12 13 import test from 'node:test'; 14 import assert from 'node:assert/strict'; 15 import fs from 'node:fs/promises'; 16 import os from 'node:os'; 17 import path from 'node:path'; 18 19 import { startDatabase, dialectAvailable } from './helpers/database.js'; 20 import { openDatabase, runMigrations, dialectFromUrl } from '../src/lib/db/index.js'; 21 import { paginateOffset, pageBlock } from '../src/lib/page.js'; 22 import { createStorage } from '../src/lib/storage/index.js'; 23 import { createLogStore } from '../src/lib/log.js'; 24 import { createRetention } from '../src/conductor/retention.js'; 25 import { createWorkerTokens } from '../src/lib/workers.js'; 26 import { createRefreshTokens } from '../src/lib/auth/refresh.js'; 27 import { createScheduler } from '../src/conductor/scheduler.js'; 28 29 const DAY = 24 * 60 * 60 * 1000; 30 const NOW = 1_800_000_000_000; 31 32 // The same shape the sqlite tests use, so the expectations below can be 33 // compared against what that dialect already proves. 34 async function seed(db, storage) { 35 await db.run( 36 `INSERT INTO projects (id, name, repo_url, enabled, job_counter, created_at, updated_at) 37 VALUES ('demo', 'Demo', '/tmp/repo', 1, 0, {now}, {now})`, 38 { now: NOW } 39 ); 40 41 for (let i = 1; i <= 5; i += 1) { 42 // Job 5 is recent; the rest are long past. 43 const at = NOW - (i === 5 ? 1 : 100) * DAY; 44 await db.run( 45 `INSERT INTO jobs (id, project_id, number, head_sha, trigger_type, state, 46 visibility, created_at, started_at, finished_at) 47 VALUES ({id}, 'demo', {number}, {sha}, 'push', 'failed', 'public', {at}, {at}, {at})`, 48 { id: `r${i}`, number: i, sha: 'a'.repeat(40), at } 49 ); 50 await db.run( 51 `INSERT INTO tasks (id, job_id, name, base_name, image, requires, spec, state, 52 allow_failure, attempt, max_attempts, timeout, 53 log_key, log_size, created_at, finished_at) 54 VALUES ({id}, {job}, 'build', 'build', 'alpine:3', '[]', '{}', 'failed', 55 0, 1, 1, 3600, {logKey}, 10, {at}, {at})`, 56 { id: `r${i}:build`, job: `r${i}`, logKey: `logs/r${i}`, at } 57 ); 58 await db.run( 59 `INSERT INTO artifacts (id, task_id, job_id, path, storage_key, size, sha256, 60 created_at, expires_at) 61 VALUES ({id}, {task}, {job}, 'out.txt', {key}, 5, {sha}, {at}, {expires})`, 62 { 63 id: `a${i}`, 64 task: `r${i}:build`, 65 job: `r${i}`, 66 key: `art/r${i}`, 67 sha: 'b'.repeat(64), 68 at, 69 // Job 1 carries an explicit deadline, which nothing protects. 70 expires: i === 1 ? NOW - 1 : null, 71 } 72 ); 73 await storage.put(`art/r${i}`, Buffer.from('artifact')); 74 await storage.put(`logs/r${i}`, Buffer.from('log')); 75 } 76 } 77 78 for (const dialect of ['postgres', 'mysql']) { 79 test(`migrations apply and retention sweeps on ${dialect}`, { skip: !dialectAvailable(dialect) }, async () => { 80 const server = await startDatabase(dialect); 81 const root = await fs.mkdtemp(path.join(os.tmpdir(), `conductor-${dialect}-`)); 82 83 const cfg = { 84 database: { url: server.url, dialect: dialectFromUrl(server.url) }, 85 storage: { path: path.join(root, 'storage') }, 86 log: { spool_path: path.join(root, 'logs'), max_size: 1 << 20 }, 87 retention: { 88 artifact_keep_jobs: 2, 89 artifact_keep_days: 7, 90 log_keep_days: 7, 91 worker_agent_keep_hours: 24, 92 sweep_interval: 3600, 93 batch: 2, 94 }, 95 }; 96 97 let db; 98 try { 99 db = await openDatabase(cfg); 100 await runMigrations(db, { logger: () => {} }); 101 102 const storage = await createStorage(cfg); 103 const logs = createLogStore(cfg); 104 await seed(db, storage); 105 106 // Paging binds both LIMIT and OFFSET, and reads a COUNT back. The 107 // count has to arrive as a number: postgres returns int8, which the 108 // driver would otherwise hand over as a string and turn the page 109 // arithmetic into concatenation. Checked before the sweep, while 110 // all five jobs are still there. 111 const firstPage = await paginateOffset(db, { 112 select: 'id', from: 'jobs', order: 'number', offset: 0, limit: 2, 113 }); 114 assert.equal(firstPage.rows.length, 2); 115 assert.equal(typeof firstPage.total, 'number', 'COUNT came back as a string'); 116 assert.equal(firstPage.total, 5); 117 118 const lastPage = await paginateOffset(db, { 119 select: 'id', from: 'jobs', order: 'number', offset: 4, limit: 2, 120 }); 121 assert.deepEqual(lastPage.rows.map((r) => r.id), ['r5'], 'the offset landed wrong'); 122 123 const window = await paginateOffset(db, { 124 select: 'id', from: 'jobs', order: 'number', offset: 2, limit: 2, 125 }); 126 assert.deepEqual(window.rows.map((r) => r.id), ['r3', 'r4']); 127 assert.equal(typeof window.total, 'number', 'COUNT came back as a string'); 128 assert.equal(window.total, 5); 129 130 const tail = await paginateOffset(db, { 131 select: 'id', from: 'jobs', order: 'number', offset: 4, limit: 2, 132 }); 133 assert.deepEqual(tail.rows.map((r) => r.id), ['r5']); 134 assert.ok(!('next' in pageBlock({ url: 'https://x.test/j', ...tail })), 'that was the last page'); 135 136 // The unique index is what makes a repeat claim update, not duplicate. 137 const workerTokens = createWorkerTokens({ db }); 138 const token = await workerTokens.create('dialect-worker'); 139 140 await workerTokens.recordAgent({ tokenId: token.id, name: 'builder-01', arches: ['x86_64'], features: ['dind'] }); 141 await workerTokens.recordAgent({ tokenId: token.id, name: 'builder-01', arches: ['x86_64'], features: ['dind', 'sign'] }); 142 await workerTokens.recordAgent({ tokenId: token.id, name: 'builder-02', arches: [], features: [] }); 143 144 const agents = await workerTokens.agentsFor(token.id); 145 assert.deepEqual(agents.map((a) => a.name).sort(), ['builder-01', 'builder-02']); 146 assert.deepEqual( 147 agents.find((a) => a.name === 'builder-01').features, 148 ['dind', 'sign'], 149 'the repeat claim should have updated in place' 150 ); 151 152 await db.run( 153 "UPDATE worker_agents SET last_seen_at = {then} WHERE name = 'builder-02'", 154 { then: NOW - 2 * DAY } 155 ); 156 await db.run( 157 "UPDATE worker_agents SET last_seen_at = {then} WHERE name = 'builder-01'", 158 { then: NOW - 1000 } 159 ); 160 161 const retention = createRetention({ 162 cfg, db, storage, logs, refreshTokens: createRefreshTokens({ db }), logger: {}, 163 }); 164 165 const first = await retention.sweep({ now: NOW }); 166 167 assert.equal(first.agents, 1, 'only the machine that stopped claiming should go'); 168 assert.deepEqual((await workerTokens.agentsFor(token.id)).map((a) => a.name), ['builder-01']); 169 170 // Job 1 by its own deadline, jobs 2 and 3 by policy. Jobs 4 and 5 171 // are the last two, so they are kept whatever their age. 172 assert.equal(first.artifacts, 3, 'three artifacts should go on the first pass'); 173 174 const left = (await db.all('SELECT id FROM artifacts ORDER BY id', {})).map((r) => r.id); 175 assert.deepEqual(left, ['a4', 'a5']); 176 177 // Logs are batched, so the four old ones take two passes. 178 assert.equal(first.logs, 2, 'the batch limit should hold on this dialect too'); 179 const second = await retention.sweep({ now: NOW }); 180 assert.equal(second.artifacts, 0, 'nothing further to delete'); 181 assert.equal(second.logs, 2); 182 183 const third = await retention.sweep({ now: NOW }); 184 assert.equal(third.logs, 0, 'a swept log must not be found again'); 185 186 const withLogs = await db.all('SELECT id FROM tasks WHERE log_key IS NOT NULL ORDER BY id', {}); 187 assert.deepEqual(withLogs.map((r) => r.id), ['r5:build'], 'only the recent job keeps its log'); 188 189 // The tasks themselves are history and stay. 190 const tasks = await db.get('SELECT COUNT(*) AS c FROM tasks', {}); 191 assert.equal(Number(tasks.c), 5, 'sweeping a log must not remove the task'); 192 } finally { 193 await db?.close().catch(() => {}); 194 await fs.rm(root, { recursive: true, force: true }); 195 await server.stop(); 196 } 197 }); 198 199 // Snapshot-at-first-read dialects only; sqlite serialises and cannot reach this. 200 test(`simultaneous completions settle the job on ${dialect}`, { skip: !dialectAvailable(dialect) }, async () => { 201 const server = await startDatabase(dialect); 202 const cfg = { database: { url: server.url, dialect: dialectFromUrl(server.url) } }; 203 const width = 8; 204 205 let db; 206 try { 207 db = await openDatabase(cfg); 208 await runMigrations(db, { logger: () => {} }); 209 210 await db.run( 211 `INSERT INTO projects (id, name, repo_url, enabled, job_counter, created_at, updated_at) 212 VALUES ('demo', 'Demo', '/tmp/repo', 1, 0, {now}, {now})`, 213 { now: NOW } 214 ); 215 await db.run( 216 `INSERT INTO jobs (id, project_id, number, head_sha, trigger_type, state, 217 visibility, created_at, started_at) 218 VALUES ('j1', 'demo', 1, {sha}, 'push', 'running', 'public', {now}, {now})`, 219 { sha: 'a'.repeat(40), now: NOW } 220 ); 221 for (let i = 0; i < width; i += 1) { 222 await db.run( 223 `INSERT INTO tasks (id, job_id, name, base_name, image, requires, spec, state, 224 allow_failure, attempt, max_attempts, timeout, log_size, created_at) 225 VALUES ({id}, 'j1', {name}, {name}, 'alpine:3', '[]', '{}', 'running', 226 0, 1, 1, 3600, 0, {now})`, 227 { id: `t${i}`, name: `build-${i}`, now: NOW } 228 ); 229 } 230 231 const scheduler = createScheduler({ 232 cfg: { ...cfg, scheduler: {} }, 233 db, 234 git: {}, logs: {}, storage: {}, projects: {}, secrets: {}, 235 logger: { warn() {}, error() {}, info() {}, debug() {} }, 236 }); 237 238 // Started together so every snapshot predates every commit; sequential 239 // completion hides the bug. 240 const outcomes = await Promise.all( 241 Array.from({ length: width }, (_, i) => scheduler.complete(`t${i}`, { success: true })) 242 ); 243 assert.ok(outcomes.every((o) => o.ok), 'every completion should be accepted'); 244 245 const job = await db.get("SELECT state FROM jobs WHERE id = 'j1'", {}); 246 assert.equal(job.state, 'success', 'the last completion has to settle the job'); 247 assert.equal( 248 outcomes.filter((o) => o.jobState === 'success').length, 1, 249 'exactly one completion should report the settle' 250 ); 251 } finally { 252 await db?.close().catch(() => {}); 253 await server.stop(); 254 } 255 }); 256 }