commit a7d3cdd18d7d49596731c3d1f4e886ad4e6bc2d2
parent 6b4544cc2b3b44c600abc16f870c728cff3e5786
Author: finwo <finwo@pm.me>
Date: Tue, 29 Sep 2026 22:02:47 +0200
Settle a job after the task outcome commits rather than inside it
Diffstat:
6 files changed, 167 insertions(+), 17 deletions(-)
diff --git a/docs/pipeline.md b/docs/pipeline.md
@@ -116,9 +116,9 @@ A `*` stands for any run of characters; everything else is literal, so a
dot is a dot rather than any character.
A task that does not match is left out of the job entirely, not recorded
-as skipped. A skipped task means a dependency collapsed and fails the job,
-which is the wrong reading for a publish step that was never meant to run
-on this branch.
+as skipped. A skipped task still satisfies `needs`, so everything below it
+runs anyway, which is the wrong reading for a publish step that was never
+meant to run on this branch.
Anything depending on an excluded task is excluded with it. Dropping the
dependency instead would let a task run without something it declared it
diff --git a/src/conductor/app.js b/src/conductor/app.js
@@ -113,6 +113,9 @@ export function startReaper(services) {
running = true;
try {
await scheduler.reap();
+ for (const job of await scheduler.settleStalled()) {
+ logger.warn?.(`settled stalled job ${job.id} as ${job.state}`);
+ }
} catch (e) {
logger.error?.(`reaper failed: ${e.message}`);
} finally {
diff --git a/src/conductor/scheduler.js b/src/conductor/scheduler.js
@@ -119,7 +119,8 @@ export function createScheduler({
return found[0] ?? null;
}
- // Marks the job finished once nothing is left to do.
+ // Needs a transaction opened after the task outcome committed: a REPEATABLE
+ // READ snapshot older than a sibling's commit still sees it running.
async function settleJob(tx, jobId) {
const outstanding = await tx.get(
"SELECT COUNT(*) AS c FROM tasks WHERE job_id = {job} AND state IN ('queued', 'running')",
@@ -135,13 +136,16 @@ export function createScheduler({
{ job: jobId }
);
const state = bad.c > 0 ? 'failed' : 'success';
- await tx.run(
+ const moved = await tx.run(
"UPDATE jobs SET state = {state}, finished_at = {now} WHERE id = {job} AND state = 'running'",
{ state, job: jobId, now: Date.now() }
);
- return state;
+ // Only the completion that moved the row reports it: settle fires once.
+ return moved.changes === 1 ? state : null;
}
+ const settle = (jobId) => db.transaction((tx) => settleJob(tx, jobId));
+
// Background compilation: git sync, pipeline read/compile, task creation.
// Updates the job from 'pending' to either 'running' (with tasks) or
// 'failed' (with an error). This is what lets a push hook return immediately.
@@ -436,7 +440,7 @@ export function createScheduler({
outputs = [], tokenId = null,
}) {
const outcome = reported ?? (success ? 'success' : 'failed');
- return db.transaction(async (tx) => {
+ const result = await db.transaction(async (tx) => {
const task = await tx.get(
`SELECT id, job_id, name, state, attempt, max_attempts, allow_failure, worker_token_id
FROM tasks WHERE id = {id}`,
@@ -460,8 +464,7 @@ export function createScheduler({
{ id: taskId, state: outcome, code: exitCode, now }
);
await writeOutputs(tx, taskId, outputs);
- const jobState = await settleJob(tx, task.job_id);
- return { ok: true, state: outcome, jobState };
+ return { ok: true, state: outcome, settleJobId: task.job_id };
}
// Transient failure with attempts left: back to the queue rather
@@ -492,15 +495,22 @@ export function createScheduler({
skipped = await skipDependents(tx, task.job_id, [taskId], now);
}
- const jobState = await settleJob(tx, task.job_id);
- return { ok: true, state: 'failed', skipped, jobState };
+ return { ok: true, state: 'failed', skipped, settleJobId: task.job_id };
});
+
+ if (result.settleJobId === undefined) return result;
+ const { settleJobId, ...rest } = result;
+ return { ...rest, jobState: await settle(settleJobId) };
},
// Every task reachable from the failed ones, not just their immediate
// dependents.
async skipDependents(jobId, fromTaskIds) {
- return db.transaction((tx) => skipDependents(tx, jobId, fromTaskIds, Date.now()));
+ const skipped = await db.transaction(
+ (tx) => skipDependents(tx, jobId, fromTaskIds, Date.now())
+ );
+ await settle(jobId);
+ return skipped;
},
// Requeues or fails tasks whose worker stopped reporting.
@@ -528,6 +538,26 @@ export function createScheduler({
return results;
},
+ // Recovers jobs whose settle was lost to a crash between the outcome and
+ // the settle transaction.
+ async settleStalled() {
+ const stalled = await db.all(
+ `SELECT id FROM jobs j
+ WHERE j.state = 'running'
+ AND NOT EXISTS (
+ SELECT 1 FROM tasks t
+ WHERE t.job_id = j.id AND t.state IN ('queued', 'running')
+ )`
+ );
+
+ const settled = [];
+ for (const job of stalled) {
+ const state = await settle(job.id);
+ if (state !== null) settled.push({ id: job.id, state });
+ }
+ return settled;
+ },
+
async cancelJob(jobId, reason = 'cancelled by request') {
return db.transaction(async (tx) => {
const job = await tx.get('SELECT id, state FROM jobs WHERE id = {id}', { id: jobId });
@@ -624,17 +654,18 @@ export function createScheduler({
// Never claimed, so it never ran: failed straight from queued.
async function failUnclaimable(candidate, message) {
- await db.transaction(async (tx) => {
+ const claimed = await db.transaction(async (tx) => {
const now = Date.now();
const marked = await tx.run(
`UPDATE tasks SET state = 'failed', error = {error}, finished_at = {now}
WHERE id = {id} AND state = 'queued'`,
{ id: candidate.id, error: message, now }
);
- if (marked.changes !== 1) return;
+ if (marked.changes !== 1) return false;
await skipDependents(tx, candidate.job_id, [candidate.id], now);
- await settleJob(tx, candidate.job_id);
+ return true;
});
+ if (claimed) await settle(candidate.job_id);
logger.warn?.(`task ${candidate.name} cannot run: ${message}`);
}
diff --git a/test/conductor.test.js b/test/conductor.test.js
@@ -1631,6 +1631,63 @@ tasks:
});
});
+// Regression: settling inside the completing transaction stranded jobs.
+test('a job whose settle was lost is recovered by the sweep', async () => {
+ const pipeline = `
+version: 1
+defaults:
+ image: alpine
+tasks:
+ build:
+ script: ['true']
+`;
+ await withHarness({ pipeline }, async (h) => {
+ const job = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().data.job_id;
+ await h.waitForJob(job);
+
+ const build = await claimNamed(h, 'build', {});
+ assert.equal((await finish(h, build.id, { success: true })).job_state, 'success');
+
+ await h.services.db.run(
+ "UPDATE jobs SET state = 'running', finished_at = NULL WHERE id = {id}",
+ { id: job }
+ );
+
+ const settled = await h.services.scheduler.settleStalled();
+ assert.deepEqual(settled, [{ id: job, state: 'success' }]);
+
+ const row = await h.services.db.get('SELECT state, finished_at FROM jobs WHERE id = {id}',
+ { id: job });
+ assert.equal(row.state, 'success');
+ assert.ok(row.finished_at > 0, 'the recovered job gets a finish time');
+ });
+});
+
+test('the sweep leaves a job that still has work to do alone', async () => {
+ const pipeline = `
+version: 1
+defaults:
+ image: alpine
+tasks:
+ build:
+ script: ['true']
+ publish:
+ script: ['true']
+ needs: [build]
+`;
+ await withHarness({ pipeline }, async (h) => {
+ const job = (await h.trigger({ sha: h.sha, ref: 'refs/heads/main' })).json().data.job_id;
+ await h.waitForJob(job);
+
+ const build = await claimNamed(h, 'build', {});
+ await finish(h, build.id, { success: true });
+
+ assert.deepEqual(await h.services.scheduler.settleStalled(), []);
+ const row = await h.services.db.get('SELECT state FROM jobs WHERE id = {id}', { id: job });
+ assert.equal(row.state, 'running');
+ });
+});
+
test('a skipped task that owed a value fails the one that needed it', async () => {
const pipeline = `
version: 1
diff --git a/test/dialects.test.js b/test/dialects.test.js
@@ -24,6 +24,7 @@ import { createLogStore } from '../src/lib/log.js';
import { createRetention } from '../src/conductor/retention.js';
import { createWorkerTokens } from '../src/lib/workers.js';
import { createRefreshTokens } from '../src/lib/auth/refresh.js';
+import { createScheduler } from '../src/conductor/scheduler.js';
const DAY = 24 * 60 * 60 * 1000;
const NOW = 1_800_000_000_000;
@@ -194,4 +195,62 @@ for (const dialect of ['postgres', 'mysql']) {
await server.stop();
}
});
+
+ // Snapshot-at-first-read dialects only; sqlite serialises and cannot reach this.
+ test(`simultaneous completions settle the job on ${dialect}`, { skip: !dialectAvailable(dialect) }, async () => {
+ const server = await startDatabase(dialect);
+ const cfg = { database: { url: server.url, dialect: dialectFromUrl(server.url) } };
+ const width = 8;
+
+ let db;
+ try {
+ db = await openDatabase(cfg);
+ await runMigrations(db, { logger: () => {} });
+
+ await db.run(
+ `INSERT INTO projects (id, name, repo_url, enabled, job_counter, created_at, updated_at)
+ VALUES ('demo', 'Demo', '/tmp/repo', 1, 0, {now}, {now})`,
+ { now: NOW }
+ );
+ await db.run(
+ `INSERT INTO jobs (id, project_id, number, head_sha, trigger_type, state,
+ visibility, created_at, started_at)
+ VALUES ('j1', 'demo', 1, {sha}, 'push', 'running', 'public', {now}, {now})`,
+ { sha: 'a'.repeat(40), now: NOW }
+ );
+ for (let i = 0; i < width; i += 1) {
+ await db.run(
+ `INSERT INTO tasks (id, job_id, name, base_name, image, requires, spec, state,
+ allow_failure, attempt, max_attempts, timeout, log_size, created_at)
+ VALUES ({id}, 'j1', {name}, {name}, 'alpine:3', '[]', '{}', 'running',
+ 0, 1, 1, 3600, 0, {now})`,
+ { id: `t${i}`, name: `build-${i}`, now: NOW }
+ );
+ }
+
+ const scheduler = createScheduler({
+ cfg: { ...cfg, scheduler: {} },
+ db,
+ git: {}, logs: {}, storage: {}, projects: {}, secrets: {},
+ logger: { warn() {}, error() {}, info() {}, debug() {} },
+ });
+
+ // Started together so every snapshot predates every commit; sequential
+ // completion hides the bug.
+ const outcomes = await Promise.all(
+ Array.from({ length: width }, (_, i) => scheduler.complete(`t${i}`, { success: true }))
+ );
+ assert.ok(outcomes.every((o) => o.ok), 'every completion should be accepted');
+
+ const job = await db.get("SELECT state FROM jobs WHERE id = 'j1'", {});
+ assert.equal(job.state, 'success', 'the last completion has to settle the job');
+ assert.equal(
+ outcomes.filter((o) => o.jobState === 'success').length, 1,
+ 'exactly one completion should report the settle'
+ );
+ } finally {
+ await db?.close().catch(() => {});
+ await server.stop();
+ }
+ });
}
diff --git a/test/only.test.js b/test/only.test.js
@@ -59,8 +59,8 @@ test('a restricted task is left out of jobs for other refs', () => {
});
test('it is left out rather than recorded as skipped', () => {
- // The scheduler reads a skipped task as a reason to fail the job, so a
- // publish step that was never meant to run here must not appear at all.
+ // A skipped task satisfies needs, so recording it that way would let the
+ // steps below a never-intended publish run anyway.
const { tasks } = compilePipeline(PIPELINE, { ref: 'refs/heads/feature' });
assert.ok(!tasks.some((j) => j.name === 'publish'));
});
