felixs.dev
forked from
danielroe.dev/synchub.to
mirror your GitHub repos to tangled.org automatically
perf: prune old terminal jobs in the worker
author
Daniel Roe
date
(Jul 10, 2026, 1:07 PM +0200)
commit
d93ff76b
d93ff76be96937eb005e637fa23f20494633a013
parent
b0ac91b9
b0ac91b9e3be2e6e8e7f40e61ba654c3a8e9cf7b
+62
-3
+7
-2
server/api/jobs/run.get.ts
+7
-2
server/api/jobs/run.get.ts
···
1
1
import crypto from 'node:crypto'
2
2
import { dispatch } from '#server/utils/job-handlers'
3
-
import { claim, complete, fail } from '#server/utils/queue'
3
+
import { claim, cleanupOldJobs, complete, fail } from '#server/utils/queue'
4
4
5
5
const LEASE_MS = 5 * 60_000 // 5 min — generous for a sync job
6
6
···
60
60
const concurrency = Number(config.workerConcurrency) || DEFAULT_CONCURRENCY
61
61
const deadline = Date.now() + budgetMs
62
62
63
+
// Opportunistic pruning of old terminal jobs. Cheap, and keeps the table
64
+
// from growing without bound (which would slow claim() and the dashboard's
65
+
// per-repo job aggregation). Failures here must not block draining.
66
+
const pruned = await cleanupOldJobs().catch(() => 0)
67
+
63
68
let processed = 0
64
69
let failed = 0
65
70
let drained = false
···
96
101
97
102
await Promise.all(Array.from({ length: concurrency }, () => lane()))
98
103
99
-
return { ok: true, processed, failed, drained, concurrency }
104
+
return { ok: true, processed, failed, pruned, drained, concurrency }
100
105
})
+25
server/utils/queue.ts
+25
server/utils/queue.ts
···
104
104
.where(sql`${job.id} = ${id}`)
105
105
}
106
106
107
+
/** Terminal jobs older than this are pruned. `failed` rows are kept longer so
108
+
* there's a window to inspect them before they age out. */
109
+
const DONE_RETENTION_MS = 24 * 60 * 60_000 // 1 day
110
+
const FAILED_RETENTION_MS = 7 * 24 * 60 * 60_000 // 7 days
111
+
112
+
/**
113
+
* Delete old terminal jobs so the table stays small. Unbounded growth slows
114
+
* both the `claim()` scan and the dashboard's per-repo job aggregation (which
115
+
* has no index on the payload's repo id). Returns the number of rows removed.
116
+
* Safe to call opportunistically from the worker; it only touches rows past
117
+
* their retention window.
118
+
*/
119
+
export async function cleanupOldJobs(): Promise<number> {
120
+
const db = useDb()
121
+
const doneCutoff = new Date(Date.now() - DONE_RETENTION_MS)
122
+
const failedCutoff = new Date(Date.now() - FAILED_RETENTION_MS)
123
+
const result = await db.execute(sql`
124
+
DELETE FROM ${job}
125
+
WHERE (${job.status} = 'done' AND ${job.updatedAt} < ${doneCutoff.toISOString()})
126
+
OR (${job.status} = 'failed' AND ${job.updatedAt} < ${failedCutoff.toISOString()})
127
+
`)
128
+
const affected = (result as { rowCount?: number, affectedRows?: number })
129
+
return affected.rowCount ?? affected.affectedRows ?? 0
130
+
}
131
+
107
132
/**
108
133
* Record a failure. Re-queues with exponential backoff until `maxAttempts`,
109
134
* after which the job is marked `failed` and stays put for inspection.
+30
-1
test/unit/queue.spec.ts
+30
-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, MAX_ATTEMPTS } from '../../server/utils/queue'
5
+
import { claim, cleanupOldJobs, complete, enqueue, fail, MAX_ATTEMPTS } from '../../server/utils/queue'
6
6
import { createTestDb } from '../utils/db'
7
7
8
8
describe('queue', () => {
···
123
123
expect(rows[0]?.lastError).toBe('terminal')
124
124
})
125
125
})
126
+
127
+
describe('cleanupOldJobs', () => {
128
+
beforeEach(async () => {
129
+
setDb(await createTestDb())
130
+
})
131
+
afterEach(() => clearDb())
132
+
133
+
it('deletes old done/failed jobs but keeps recent and active ones', async () => {
134
+
const db = useDb()
135
+
// Insert rows at controlled ages via raw updated_at.
136
+
await enqueue('github.push', { a: 1 }) // stays queued
137
+
const oldDone = await enqueue('github.push', { a: 2 })
138
+
const recentDone = await enqueue('github.push', { a: 3 })
139
+
const oldFailed = await enqueue('github.push', { a: 4 })
140
+
141
+
await db.execute(sql`UPDATE ${job} SET status='done', updated_at = now() - interval '2 days' WHERE ${job.id} = ${oldDone.id}`)
142
+
await db.execute(sql`UPDATE ${job} SET status='done', updated_at = now() - interval '1 hour' WHERE ${job.id} = ${recentDone.id}`)
143
+
await db.execute(sql`UPDATE ${job} SET status='failed', updated_at = now() - interval '8 days' WHERE ${job.id} = ${oldFailed.id}`)
144
+
145
+
const removed = await cleanupOldJobs()
146
+
expect(removed).toBe(2) // oldDone + oldFailed
147
+
148
+
const remaining = await db.select().from(job)
149
+
const ids = remaining.map(r => r.id)
150
+
expect(ids).not.toContain(oldDone.id)
151
+
expect(ids).not.toContain(oldFailed.id)
152
+
expect(ids).toContain(recentDone.id)
153
+
})
154
+
})