From cd8c9d438dbf9c3e450085eb9454dd8a2f2b2465 Mon Sep 17 00:00:00 2001 From: Daniel Roe Date: Fri, 10 Jul 2026 12:18:50 +0200 Subject: [PATCH] fix: drain job queue concurrently and retire poison jobs --- .env.example | 9 ++- nuxt.config.ts | 1 + server/api/jobs/run.get.ts | 99 ++++++++++++++++++++------- server/utils/git-wire/receive-pack.ts | 2 +- server/utils/queue.ts | 28 +++++++- test/unit/queue.spec.ts | 29 +++++++- 6 files changed, 136 insertions(+), 32 deletions(-) diff --git a/.env.example b/.env.example index 375d28e..b95474c 100644 --- a/.env.example +++ b/.env.example @@ -83,6 +83,11 @@ NUXT_GITHUB_APP_INSTALL_URL=https://github.com/apps/synchub-to/installations/new CRON_SECRET= # Optional: per-invocation worker time budget in milliseconds. -# Default 25_000. Set lower in dev so `pnpm jobs:tick` returns sooner when -# the queue is empty. +# Default 270_000 (just under the 300s Vercel maxDuration). Set lower in dev +# so `pnpm jobs:tick` returns sooner when the queue is empty. # NUXT_WORKER_BUDGET_MS=5000 + +# Optional: how many jobs the worker runs concurrently per invocation. +# Default 6. Jobs are network-bound (SSH to the knot, HTTPS to GitHub), so +# concurrency raises throughput without CPU contention. +# NUXT_WORKER_CONCURRENCY=6 diff --git a/nuxt.config.ts b/nuxt.config.ts index ab8ffb2..34e7525 100644 --- a/nuxt.config.ts +++ b/nuxt.config.ts @@ -26,6 +26,7 @@ export default defineNuxtConfig({ githubAppClientSecret: '', githubWebhookSecret: '', workerBudgetMs: '', + workerConcurrency: '', maxPackBytes: '', encryptionKey: '', atprotoPrivateJwk: '', diff --git a/server/api/jobs/run.get.ts b/server/api/jobs/run.get.ts index 6cebedb..9537e39 100644 --- a/server/api/jobs/run.get.ts +++ b/server/api/jobs/run.get.ts @@ -3,7 +3,46 @@ import { dispatch } from '#server/utils/job-handlers' import { claim, complete, fail } from '#server/utils/queue' const LEASE_MS = 5 * 60_000 // 5 min — generous for a sync job -const DEFAULT_BUDGET_MS = 25_000 // leave headroom under Vercel's 10s default; pro tiers can override + +// Use most of Vercel's 300s `maxDuration` (see nuxt.config `functions`). The +// old 25s budget drained only a handful of jobs per minute, so the queue grew +// unboundedly under real push volume. Leave headroom for the response and for +// in-flight jobs to settle. +const DEFAULT_BUDGET_MS = 270_000 + +// How many jobs to run at once inside one invocation. Each job is mostly +// network wait (SSH to the knot, HTTPS to GitHub), so concurrency buys real +// throughput without CPU contention. A single slow/hung job no longer blocks +// the others for the whole budget. +const DEFAULT_CONCURRENCY = 6 + +// Hard cap on a single job's wall-clock time. A knot SSH connection can hang +// past its `readyTimeout`; without this the slot stays busy until the budget +// ends. On timeout the job is recorded as a failure so backoff and the attempt +// ceiling apply, and the slot is freed for the next job. +const JOB_TIMEOUT_MS = 45_000 + +class JobTimeoutError extends Error { + constructor(ms: number) { + super(`job exceeded ${ms}ms wall-clock cap`) + this.name = 'JobTimeoutError' + } +} + +async function withTimeout(p: Promise, ms: number): Promise { + let timer: ReturnType | undefined + try { + return await Promise.race([ + p, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new JobTimeoutError(ms)), ms) + }), + ]) + } + finally { + if (timer) clearTimeout(timer) + } +} export default defineEventHandler(async event => { const cronSecret = process.env.CRON_SECRET @@ -16,38 +55,46 @@ export default defineEventHandler(async event => { throw createError({ statusCode: 401, statusMessage: 'unauthorized' }) } - const workerId = `${process.env.VERCEL_DEPLOYMENT_ID ?? 'local'}:${crypto.randomUUID()}` - const budgetMs = Number(useRuntimeConfig().workerBudgetMs) || DEFAULT_BUDGET_MS + const config = useRuntimeConfig() + const budgetMs = Number(config.workerBudgetMs) || DEFAULT_BUDGET_MS + const concurrency = Number(config.workerConcurrency) || DEFAULT_CONCURRENCY const deadline = Date.now() + budgetMs let processed = 0 + let failed = 0 let drained = false - // Sequential by design: each iteration claims one job, runs it, records the - // outcome. We don't parallelise because each Vercel invocation is a single - // small worker; concurrency comes from cron firing multiple invocations. - // eslint-disable-next-line no-await-in-loop - while (Date.now() < deadline) { - // eslint-disable-next-line no-await-in-loop - const job = await claim(workerId, LEASE_MS) - if (!job) { - drained = true - break - } + // Each lane runs the claim -> dispatch -> record loop independently until the + // queue drains or the budget runs out. `claim()` is an atomic + // `FOR UPDATE SKIP LOCKED`, so lanes never race for the same row; concurrency + // comes from lanes waiting on different jobs' network I/O at the same time. + async function lane() { + const workerId = `${process.env.VERCEL_DEPLOYMENT_ID ?? 'local'}:${crypto.randomUUID()}` + while (Date.now() < deadline) { + // eslint-disable-next-line no-await-in-loop -- each iteration processes one job to completion + const job = await claim(workerId, LEASE_MS) + if (!job) { + drained = true + return + } - try { - // eslint-disable-next-line no-await-in-loop - await dispatch(job) - // eslint-disable-next-line no-await-in-loop - await complete(job.id) - } - catch (err) { - // eslint-disable-next-line no-await-in-loop - await fail(job.id, job.attempts, err) - } + try { + // eslint-disable-next-line no-await-in-loop + await withTimeout(dispatch(job), JOB_TIMEOUT_MS) + // eslint-disable-next-line no-await-in-loop + await complete(job.id) + } + catch (err) { + failed++ + // eslint-disable-next-line no-await-in-loop + await fail(job.id, job.attempts, err) + } - processed++ + processed++ + } } - return { ok: true, processed, drained, workerId } + await Promise.all(Array.from({ length: concurrency }, () => lane())) + + return { ok: true, processed, failed, drained, concurrency } }) diff --git a/server/utils/git-wire/receive-pack.ts b/server/utils/git-wire/receive-pack.ts index fd4b081..5cfa977 100644 --- a/server/utils/git-wire/receive-pack.ts +++ b/server/utils/git-wire/receive-pack.ts @@ -127,7 +127,7 @@ export function ssh2ReceivePackFactory(target: SshTarget): ReceivePackFactory { port: target.port ?? 22, username: 'git', privateKey: target.privateKey, - readyTimeout: 15_000, + readyTimeout: 8_000, hostVerifier: () => true, }) diff --git a/server/utils/queue.ts b/server/utils/queue.ts index 662c4e4..cd60a63 100644 --- a/server/utils/queue.ts +++ b/server/utils/queue.ts @@ -14,6 +14,16 @@ export interface EnqueueOptions { runAfter?: Date } +/** + * Hard ceiling on delivery attempts. A job that reaches this many attempts is + * marked `failed` rather than re-leased. This is the backstop for jobs whose + * work outlives the worker budget: those return without ever reaching `fail()` + * (which is what enforces the soft `maxAttempts` backoff), so the row stays + * `running` with an expired lease and would otherwise be re-claimed forever. + * `claim()` enforces this ceiling before handing a job back out. + */ +export const MAX_ATTEMPTS = 8 + /** * Push a job onto the queue. `payload` must be a JSON-serialisable object — keep * it small (an envelope of identifiers, not a webhook body); see PLAN.md. @@ -42,6 +52,20 @@ export async function claim(workerId: string, leaseMs: number): Promise= ${MAX_ATTEMPTS} + `) + const result = await db.execute(sql` UPDATE ${job} SET @@ -55,7 +79,7 @@ export async function claim(workerId: string, leaseMs: number): Promise { @@ -85,6 +85,33 @@ describe('queue', () => { expect(new Date(rows[0]?.runAfter ?? 0).getTime()).toBeGreaterThan(Date.now()) }) + it('retires a stuck running job once it reaches the attempt ceiling', async () => { + await enqueue('github.push', {}) + const db = useDb() + + // Drive the job to the ceiling by expiring its lease and re-claiming, as + // happens when a job's work times out past the worker budget without ever + // reaching fail(). + let last: Awaited> = null + for (let i = 0; i < MAX_ATTEMPTS; i++) { + // eslint-disable-next-line no-await-in-loop + last = await claim('worker', 60_000) + expect(last).not.toBeNull() + // eslint-disable-next-line no-await-in-loop + await db.execute(sql`UPDATE ${job} SET locked_until = now() - interval '1 minute'`) + } + expect(last?.attempts).toBe(MAX_ATTEMPTS) + + // The next claim must not hand the poison job back out; instead it retires + // it to `failed` and returns null (queue otherwise empty). + const next = await claim('worker', 60_000) + expect(next).toBeNull() + + const rows = await db.select().from(job) + expect(rows[0]?.status).toBe('failed') + expect(rows[0]?.lockedBy).toBeNull() + }) + it('marks failed once attempts >= maxAttempts', async () => { await enqueue('github.push', {}) const claimed = await claim('worker-1', 60_000) -- 2.51.2