Something went wrong. Try again.
mirror your GitHub repos to tangled.org automatically
Something went wrong. Try again.
6.7 kB · 179 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180import { sql } from 'drizzle-orm'import { job } from '../db/schema'import { useDb } from './db'
export interface JobEnvelope extends Record<string, unknown> { id: number kind: string payload: unknown attempts: number}
export interface EnqueueOptions { /** Run no earlier than this time. Default: now. */ 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. */export async function enqueue(kind: string, payload: object, opts: EnqueueOptions = {}) { const db = useDb() const [row] = await db.insert(job).values({ kind, payload, runAfter: opts.runAfter ?? new Date(), }).returning({ id: job.id }) return row!}
/** * Atomically claim the oldest runnable job. Uses * `UPDATE ... WHERE id = (SELECT ... FOR UPDATE SKIP LOCKED LIMIT 1)` * so multiple workers don't race for the same row. * * Returns null when the queue is empty (or all rows are locked / not yet due). */export async function claim(workerId: string, leaseMs: number): Promise<JobEnvelope | null> { const db = useDb() const leaseUntil = new Date(Date.now() + leaseMs)
// Two conditions for a job to be claimable: // 1. status='queued' AND run_after <= now() // 2. status='running' AND locked_until < now() (lease expired) // Retire jobs that have burned through the attempt ceiling. Without this an // expired-lease `running` job (its work timed out past the worker budget, so // `fail()` never ran) would be re-claimed indefinitely, climbing to dozens of // attempts while blocking the head of the queue. Do it in the same query as // the claim so a single worker tick makes forward progress even when the head // of the queue is all poison. await db.execute(sql` UPDATE ${job} SET status = 'failed', locked_by = null, locked_until = null, updated_at = now() WHERE ${job.status} = 'running' AND ${job.lockedUntil} < now() AND ${job.attempts} >= ${MAX_ATTEMPTS} `)
const result = await db.execute<JobEnvelope>(sql` UPDATE ${job} SET status = 'running', locked_by = ${workerId}, locked_until = ${leaseUntil.toISOString()}, attempts = ${job.attempts} + 1, updated_at = now() WHERE ${job.id} = ( SELECT ${job.id} FROM ${job} WHERE (${job.status} = 'queued' AND ${job.runAfter} <= now()) OR (${job.status} = 'running' AND ${job.lockedUntil} < now() AND ${job.attempts} < ${MAX_ATTEMPTS}) ORDER BY ${job.runAfter} ASC FOR UPDATE SKIP LOCKED LIMIT 1 ) RETURNING id, kind, payload, attempts `)
// drizzle's neon-http execute returns rows on `.rows`; pglite's returns directly. // Normalise. const rows: JobEnvelope[] = Array.isArray(result) ? result : ((result as { rows?: JobEnvelope[] }).rows ?? []) const envelope = rows[0] return envelope ?? null}
/** Mark a job as completed. */export async function complete(id: number) { const db = useDb() await db.update(job) .set({ status: 'done', lockedBy: null, lockedUntil: null, lastError: null, updatedAt: new Date() }) .where(sql`${job.id} = ${id}`)}
/** Terminal jobs older than this are pruned. `failed` rows are kept longer so * there's a window to inspect them before they age out. */const DONE_RETENTION_MS = 24 * 60 * 60_000 // 1 dayconst FAILED_RETENTION_MS = 7 * 24 * 60 * 60_000 // 7 days
/** * Delete old terminal jobs so the table stays small. Unbounded growth slows * both the `claim()` scan and the dashboard's per-repo job aggregation (which * has no index on the payload's repo id). Returns the number of rows removed. * Safe to call opportunistically from the worker; it only touches rows past * their retention window. */export async function cleanupOldJobs(): Promise<number> { const db = useDb() const doneCutoff = new Date(Date.now() - DONE_RETENTION_MS) const failedCutoff = new Date(Date.now() - FAILED_RETENTION_MS) const result = await db.execute(sql` DELETE FROM ${job} WHERE (${job.status} = 'done' AND ${job.updatedAt} < ${doneCutoff.toISOString()}) OR (${job.status} = 'failed' AND ${job.updatedAt} < ${failedCutoff.toISOString()}) `) const affected = (result as { rowCount?: number, affectedRows?: number }) return affected.rowCount ?? affected.affectedRows ?? 0}
export interface FailOptions { maxAttempts?: number /** * When true, the job keeps re-queuing past `maxAttempts` instead of being * retired to `failed`. For failures that are temporary by nature (a flaky * knot SSH endpoint resetting connections), giving up permanently would * abandon legitimate work: a push to a real branch must land eventually, not * be dropped after 8 unlucky attempts. Backoff still caps at 1h, so an * indefinitely-retrying job polls at most hourly rather than hammering. */ retryForever?: boolean}
/** * Record a failure. Re-queues with exponential backoff until `maxAttempts`, * after which the job is marked `failed` and stays put for inspection, unless * `retryForever` is set (see `FailOptions`). */export async function fail(id: number, attempts: number, err: unknown, opts: FailOptions = {}) { const { maxAttempts = MAX_ATTEMPTS, retryForever = false } = opts const db = useDb() const message = err instanceof Error ? err.message : String(err)
if (!retryForever && attempts >= maxAttempts) { await db.update(job) .set({ status: 'failed', lastError: message, lockedBy: null, lockedUntil: null, updatedAt: new Date() }) .where(sql`${job.id} = ${id}`) return }
// Exponential backoff with ±20% jitter: base*2^attempts, capped. const base = 30_000 // 30s const cap = 60 * 60_000 // 1h const backoff = Math.min(base * (2 ** (attempts - 1)), cap) const jitter = backoff * (0.8 + Math.random() * 0.4) const runAfter = new Date(Date.now() + jitter)
await db.update(job) .set({ status: 'queued', lockedBy: null, lockedUntil: null, runAfter, lastError: message, updatedAt: new Date(), }) .where(sql`${job.id} = ${id}`)}