mirror your GitHub repos to tangled.org automatically
synchub.to
mirror
sync
tangled
github
git
fix: drain job queue concurrently and retire poison jobs
Daniel Roe
(Jul 10, 2026, 12:20 PM +0200)
cd8c9d43
c053c4a3
+136
-32
+7
-2
.env.example
+7
-2
.env.example
···
83
83
CRON_SECRET=<base64url-encoded 32 bytes>
84
84
85
85
# Optional: per-invocation worker time budget in milliseconds.
86
-
# Default 25_000. Set lower in dev so `pnpm jobs:tick` returns sooner when
87
-
# the queue is empty.
86
+
# Default 270_000 (just under the 300s Vercel maxDuration). Set lower in dev
87
+
# so `pnpm jobs:tick` returns sooner when the queue is empty.
88
88
# NUXT_WORKER_BUDGET_MS=5000
89
+
90
+
# Optional: how many jobs the worker runs concurrently per invocation.
91
+
# Default 6. Jobs are network-bound (SSH to the knot, HTTPS to GitHub), so
92
+
# concurrency raises throughput without CPU contention.
93
+
# NUXT_WORKER_CONCURRENCY=6
+1
nuxt.config.ts
+1
nuxt.config.ts
+73
-26
server/api/jobs/run.get.ts
+73
-26
server/api/jobs/run.get.ts
···
3
3
import { claim, complete, fail } from '#server/utils/queue'
4
4
5
5
const LEASE_MS = 5 * 60_000 // 5 min — generous for a sync job
6
-
const DEFAULT_BUDGET_MS = 25_000 // leave headroom under Vercel's 10s default; pro tiers can override
6
+
7
+
// Use most of Vercel's 300s `maxDuration` (see nuxt.config `functions`). The
8
+
// old 25s budget drained only a handful of jobs per minute, so the queue grew
9
+
// unboundedly under real push volume. Leave headroom for the response and for
10
+
// in-flight jobs to settle.
11
+
const DEFAULT_BUDGET_MS = 270_000
12
+
13
+
// How many jobs to run at once inside one invocation. Each job is mostly
14
+
// network wait (SSH to the knot, HTTPS to GitHub), so concurrency buys real
15
+
// throughput without CPU contention. A single slow/hung job no longer blocks
16
+
// the others for the whole budget.
17
+
const DEFAULT_CONCURRENCY = 6
18
+
19
+
// Hard cap on a single job's wall-clock time. A knot SSH connection can hang
20
+
// past its `readyTimeout`; without this the slot stays busy until the budget
21
+
// ends. On timeout the job is recorded as a failure so backoff and the attempt
22
+
// ceiling apply, and the slot is freed for the next job.
23
+
const JOB_TIMEOUT_MS = 45_000
24
+
25
+
class JobTimeoutError extends Error {
26
+
constructor(ms: number) {
27
+
super(`job exceeded ${ms}ms wall-clock cap`)
28
+
this.name = 'JobTimeoutError'
29
+
}
30
+
}
31
+
32
+
async function withTimeout<T>(p: Promise<T>, ms: number): Promise<T> {
33
+
let timer: ReturnType<typeof setTimeout> | undefined
34
+
try {
35
+
return await Promise.race([
36
+
p,
37
+
new Promise<never>((_, reject) => {
38
+
timer = setTimeout(() => reject(new JobTimeoutError(ms)), ms)
39
+
}),
40
+
])
41
+
}
42
+
finally {
43
+
if (timer) clearTimeout(timer)
44
+
}
45
+
}
7
46
8
47
export default defineEventHandler(async event => {
9
48
const cronSecret = process.env.CRON_SECRET
···
16
55
throw createError({ statusCode: 401, statusMessage: 'unauthorized' })
17
56
}
18
57
19
-
const workerId = `${process.env.VERCEL_DEPLOYMENT_ID ?? 'local'}:${crypto.randomUUID()}`
20
-
const budgetMs = Number(useRuntimeConfig().workerBudgetMs) || DEFAULT_BUDGET_MS
58
+
const config = useRuntimeConfig()
59
+
const budgetMs = Number(config.workerBudgetMs) || DEFAULT_BUDGET_MS
60
+
const concurrency = Number(config.workerConcurrency) || DEFAULT_CONCURRENCY
21
61
const deadline = Date.now() + budgetMs
22
62
23
63
let processed = 0
64
+
let failed = 0
24
65
let drained = false
25
66
26
-
// Sequential by design: each iteration claims one job, runs it, records the
27
-
// outcome. We don't parallelise because each Vercel invocation is a single
28
-
// small worker; concurrency comes from cron firing multiple invocations.
29
-
// eslint-disable-next-line no-await-in-loop
30
-
while (Date.now() < deadline) {
31
-
// eslint-disable-next-line no-await-in-loop
32
-
const job = await claim(workerId, LEASE_MS)
33
-
if (!job) {
34
-
drained = true
35
-
break
36
-
}
67
+
// Each lane runs the claim -> dispatch -> record loop independently until the
68
+
// queue drains or the budget runs out. `claim()` is an atomic
69
+
// `FOR UPDATE SKIP LOCKED`, so lanes never race for the same row; concurrency
70
+
// comes from lanes waiting on different jobs' network I/O at the same time.
71
+
async function lane() {
72
+
const workerId = `${process.env.VERCEL_DEPLOYMENT_ID ?? 'local'}:${crypto.randomUUID()}`
73
+
while (Date.now() < deadline) {
74
+
// eslint-disable-next-line no-await-in-loop -- each iteration processes one job to completion
75
+
const job = await claim(workerId, LEASE_MS)
76
+
if (!job) {
77
+
drained = true
78
+
return
79
+
}
80
+
81
+
try {
82
+
// eslint-disable-next-line no-await-in-loop
83
+
await withTimeout(dispatch(job), JOB_TIMEOUT_MS)
84
+
// eslint-disable-next-line no-await-in-loop
85
+
await complete(job.id)
86
+
}
87
+
catch (err) {
88
+
failed++
89
+
// eslint-disable-next-line no-await-in-loop
90
+
await fail(job.id, job.attempts, err)
91
+
}
37
92
38
-
try {
39
-
// eslint-disable-next-line no-await-in-loop
40
-
await dispatch(job)
41
-
// eslint-disable-next-line no-await-in-loop
42
-
await complete(job.id)
43
-
}
44
-
catch (err) {
45
-
// eslint-disable-next-line no-await-in-loop
46
-
await fail(job.id, job.attempts, err)
93
+
processed++
47
94
}
48
-
49
-
processed++
50
95
}
51
96
52
-
return { ok: true, processed, drained, workerId }
97
+
await Promise.all(Array.from({ length: concurrency }, () => lane()))
98
+
99
+
return { ok: true, processed, failed, drained, concurrency }
53
100
})
+1
-1
server/utils/git-wire/receive-pack.ts
+1
-1
server/utils/git-wire/receive-pack.ts
+26
-2
server/utils/queue.ts
+26
-2
server/utils/queue.ts
···
15
15
}
16
16
17
17
/**
18
+
* Hard ceiling on delivery attempts. A job that reaches this many attempts is
19
+
* marked `failed` rather than re-leased. This is the backstop for jobs whose
20
+
* work outlives the worker budget: those return without ever reaching `fail()`
21
+
* (which is what enforces the soft `maxAttempts` backoff), so the row stays
22
+
* `running` with an expired lease and would otherwise be re-claimed forever.
23
+
* `claim()` enforces this ceiling before handing a job back out.
24
+
*/
25
+
export const MAX_ATTEMPTS = 8
26
+
27
+
/**
18
28
* Push a job onto the queue. `payload` must be a JSON-serialisable object — keep
19
29
* it small (an envelope of identifiers, not a webhook body); see PLAN.md.
20
30
*/
···
42
52
// Two conditions for a job to be claimable:
43
53
// 1. status='queued' AND run_after <= now()
44
54
// 2. status='running' AND locked_until < now() (lease expired)
55
+
// Retire jobs that have burned through the attempt ceiling. Without this an
56
+
// expired-lease `running` job (its work timed out past the worker budget, so
57
+
// `fail()` never ran) would be re-claimed indefinitely, climbing to dozens of
58
+
// attempts while blocking the head of the queue. Do it in the same query as
59
+
// the claim so a single worker tick makes forward progress even when the head
60
+
// of the queue is all poison.
61
+
await db.execute(sql`
62
+
UPDATE ${job}
63
+
SET status = 'failed', locked_by = null, locked_until = null, updated_at = now()
64
+
WHERE ${job.status} = 'running'
65
+
AND ${job.lockedUntil} < now()
66
+
AND ${job.attempts} >= ${MAX_ATTEMPTS}
67
+
`)
68
+
45
69
const result = await db.execute<JobEnvelope>(sql`
46
70
UPDATE ${job}
47
71
SET
···
55
79
WHERE
56
80
(${job.status} = 'queued' AND ${job.runAfter} <= now())
57
81
OR
58
-
(${job.status} = 'running' AND ${job.lockedUntil} < now())
82
+
(${job.status} = 'running' AND ${job.lockedUntil} < now() AND ${job.attempts} < ${MAX_ATTEMPTS})
59
83
ORDER BY ${job.runAfter} ASC
60
84
FOR UPDATE SKIP LOCKED
61
85
LIMIT 1
···
84
108
* Record a failure. Re-queues with exponential backoff until `maxAttempts`,
85
109
* after which the job is marked `failed` and stays put for inspection.
86
110
*/
87
-
export async function fail(id: number, attempts: number, err: unknown, maxAttempts = 5) {
111
+
export async function fail(id: number, attempts: number, err: unknown, maxAttempts = MAX_ATTEMPTS) {
88
112
const db = useDb()
89
113
const message = err instanceof Error ? err.message : String(err)
90
114
+28
-1
test/unit/queue.spec.ts
+28
-1
test/unit/queue.spec.ts
···
2
2
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
3
3
import { job } from '../../server/db/schema'
4
4
import { clearDb, setDb, useDb } from '../../server/utils/db'
5
-
import { claim, complete, enqueue, fail } from '../../server/utils/queue'
5
+
import { claim, complete, enqueue, fail, MAX_ATTEMPTS } from '../../server/utils/queue'
6
6
import { createTestDb } from '../utils/db'
7
7
8
8
describe('queue', () => {
···
83
83
expect(rows[0]?.status).toBe('queued')
84
84
expect(rows[0]?.lastError).toBe('boom')
85
85
expect(new Date(rows[0]?.runAfter ?? 0).getTime()).toBeGreaterThan(Date.now())
86
+
})
87
+
88
+
it('retires a stuck running job once it reaches the attempt ceiling', async () => {
89
+
await enqueue('github.push', {})
90
+
const db = useDb()
91
+
92
+
// Drive the job to the ceiling by expiring its lease and re-claiming, as
93
+
// happens when a job's work times out past the worker budget without ever
94
+
// reaching fail().
95
+
let last: Awaited<ReturnType<typeof claim>> = null
96
+
for (let i = 0; i < MAX_ATTEMPTS; i++) {
97
+
// eslint-disable-next-line no-await-in-loop
98
+
last = await claim('worker', 60_000)
99
+
expect(last).not.toBeNull()
100
+
// eslint-disable-next-line no-await-in-loop
101
+
await db.execute(sql`UPDATE ${job} SET locked_until = now() - interval '1 minute'`)
102
+
}
103
+
expect(last?.attempts).toBe(MAX_ATTEMPTS)
104
+
105
+
// The next claim must not hand the poison job back out; instead it retires
106
+
// it to `failed` and returns null (queue otherwise empty).
107
+
const next = await claim('worker', 60_000)
108
+
expect(next).toBeNull()
109
+
110
+
const rows = await db.select().from(job)
111
+
expect(rows[0]?.status).toBe('failed')
112
+
expect(rows[0]?.lockedBy).toBeNull()
86
113
})
87
114
88
115
it('marks failed once attempts >= maxAttempts', async () => {