Something went wrong. Try again.
mirror your GitHub repos to tangled.org automatically synchub.to
mirror sync tangled github git
Something went wrong. Try again.
6.3 kB · 166 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167import { sql } from 'drizzle-orm'import { afterEach, beforeEach, describe, expect, it } from 'vitest'import { job } from '../../server/db/schema'import { clearDb, setDb, useDb } from '../../server/utils/db'import { claim, cleanupOldJobs, complete, enqueue, fail, MAX_ATTEMPTS } from '../../server/utils/queue'import { createTestDb } from '../utils/db'
describe('queue', () => { beforeEach(async () => { setDb(await createTestDb()) })
afterEach(() => { clearDb() })
it('enqueues and claims a job', async () => { const queued = await enqueue('github.push', { foo: 'bar' }) expect(queued.id).toBeGreaterThan(0)
const claimed = await claim('worker-1', 60_000) expect(claimed?.kind).toBe('github.push') expect(claimed?.payload).toEqual({ foo: 'bar' }) expect(claimed?.attempts).toBe(1) })
it('returns null when the queue is empty', async () => { const claimed = await claim('worker-1', 60_000) expect(claimed).toBeNull() })
it('does not claim the same job twice', async () => { await enqueue('github.push', { foo: 'bar' })
const a = await claim('worker-1', 60_000) const b = await claim('worker-2', 60_000)
expect(a).not.toBeNull() expect(b).toBeNull() })
it('respects run_after', async () => { const future = new Date(Date.now() + 60_000) await enqueue('github.push', {}, { runAfter: future })
const claimed = await claim('worker-1', 60_000) expect(claimed).toBeNull() })
it('reclaims a job whose lease has expired', async () => { await enqueue('github.push', {})
const first = await claim('worker-1', 60_000) expect(first).not.toBeNull()
// Force the lease into the past. const db = useDb() await db.execute(sql`UPDATE ${job} SET locked_until = now() - interval '1 minute'`)
const second = await claim('worker-2', 60_000) expect(second?.id).toBe(first!.id) expect(second?.attempts).toBe(2) // attempts increments on each claim })
it('marks a job as done on complete', async () => { const queued = await enqueue('github.push', {}) const claimed = await claim('worker-1', 60_000) await complete(claimed!.id)
const db = useDb() const rows = await db.select().from(job).where(sql`${job.id} = ${queued.id}`) expect(rows[0]?.status).toBe('done') expect(rows[0]?.lockedBy).toBeNull() })
it('re-queues with backoff on fail before maxAttempts', async () => { await enqueue('github.push', {}) const claimed = await claim('worker-1', 60_000) await fail(claimed!.id, claimed!.attempts, new Error('boom'))
const db = useDb() const rows = await db.select().from(job).where(sql`${job.id} = ${claimed!.id}`) expect(rows[0]?.status).toBe('queued') expect(rows[0]?.lastError).toBe('boom') 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<ReturnType<typeof claim>> = 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) await fail(claimed!.id, 5, new Error('terminal'), { maxAttempts: 5 })
const db = useDb() const rows = await db.select().from(job).where(sql`${job.id} = ${claimed!.id}`) expect(rows[0]?.status).toBe('failed') expect(rows[0]?.lastError).toBe('terminal') })
it('keeps re-queuing past maxAttempts when retryForever is set', async () => { await enqueue('github.push', {}) const claimed = await claim('worker-1', 60_000) // Well past the ceiling, but a transient transport failure must not retire. await fail(claimed!.id, 99, new Error('ssh error: read ECONNRESET'), { retryForever: true })
const db = useDb() const rows = await db.select().from(job).where(sql`${job.id} = ${claimed!.id}`) expect(rows[0]?.status).toBe('queued') expect(new Date(rows[0]?.runAfter ?? 0).getTime()).toBeGreaterThan(Date.now()) })})
describe('cleanupOldJobs', () => { beforeEach(async () => { setDb(await createTestDb()) }) afterEach(() => clearDb())
it('deletes old done/failed jobs but keeps recent and active ones', async () => { const db = useDb() // Insert rows at controlled ages via raw updated_at. await enqueue('github.push', { a: 1 }) // stays queued const oldDone = await enqueue('github.push', { a: 2 }) const recentDone = await enqueue('github.push', { a: 3 }) const oldFailed = await enqueue('github.push', { a: 4 })
await db.execute(sql`UPDATE ${job} SET status='done', updated_at = now() - interval '2 days' WHERE ${job.id} = ${oldDone.id}`) await db.execute(sql`UPDATE ${job} SET status='done', updated_at = now() - interval '1 hour' WHERE ${job.id} = ${recentDone.id}`) await db.execute(sql`UPDATE ${job} SET status='failed', updated_at = now() - interval '8 days' WHERE ${job.id} = ${oldFailed.id}`)
const removed = await cleanupOldJobs() expect(removed).toBe(2) // oldDone + oldFailed
const remaining = await db.select().from(job) const ids = remaining.map(r => r.id) expect(ids).not.toContain(oldDone.id) expect(ids).not.toContain(oldFailed.id) expect(ids).toContain(recentDone.id) })})