workers: requeue stale active jobs on start, dedupe scheduled crawls, kick resolver at boot
2 changed files +8 −0
modified
workers/crawler/scheduler.ts
+4 −0
@@ -17,7 +17,11 @@ export async function scheduleDueCrawls(queue: Queue): Promise<number> { | |||||
| 17 | .from(connectorsTable) | 17 | .from(connectorsTable) | ||
| 18 | .where(and(eq(connectorsTable.status, 'active'), or(isNull(connectorsTable.nextRunAt), lte(connectorsTable.nextRunAt, now)))); | 18 | .where(and(eq(connectorsTable.status, 'active'), or(isNull(connectorsTable.nextRunAt), lte(connectorsTable.nextRunAt, now)))); | ||
| 19 | let n = 0; | 19 | let n = 0; | ||
| 20 | + // pg-boss singleton keys do not dedupe across states reliably; skip connectors that already have a queued/active crawl. | ||||
| 21 | + const queued = (await db().execute(sql`select distinct data->>'connectorId' as id from pgboss.job where name = ${JOBS.crawlRun} and state in ('created','retry','active')`)) as unknown as Array<{ id: string }>; | ||||
| 22 | + const busy = new Set(queued.map((q) => q.id)); | ||||
| 20 | for (const c of due) { | 23 | for (const c of due) { | ||
| 24 | + if (busy.has(c.id)) continue; | ||||
| 21 | const id = await queue.send(JOBS.crawlRun, { connectorId: c.id, mode: 'incremental', trigger: 'schedule' }, { singletonKey: `crawl:${c.id}`, priority: PRIORITY[c.priority] ?? 5, expireInSeconds: 6 * 3600, retryLimit: 1 }); | 25 | const id = await queue.send(JOBS.crawlRun, { connectorId: c.id, mode: 'incremental', trigger: 'schedule' }, { singletonKey: `crawl:${c.id}`, priority: PRIORITY[c.priority] ?? 5, expireInSeconds: 6 * 3600, retryLimit: 1 }); | ||
| 22 | if (id) n++; | 26 | if (id) n++; | ||
| 23 | } | 27 | } | ||
modified
workers/main.ts
+4 −0
@@ -26,6 +26,9 @@ export async function startWorker(): Promise<() => Promise<void>> { | |||||
| 26 | const log = logger.child({ component: 'worker' }); | 26 | const log = logger.child({ component: 'worker' }); | ||
| 27 | const queue = createQueue(); | 27 | const queue = createQueue(); | ||
| 28 | await queue.start(); | 28 | await queue.start(); | ||
| 29 | + // Single-worker deployment: jobs left 'active' by a previous process (restart/crash) would wait for | ||||
| 30 | + // their expiry (hours). Re-queue them so crawls resume immediately. | ||||
| 31 | + await db().execute(sql`update pgboss.job set state = 'retry', started_on = null where state = 'active' and started_on < now() - interval '90 seconds'`); | ||||
| 29 | const concurrency = env().WORKER_CONCURRENCY; | 32 | const concurrency = env().WORKER_CONCURRENCY; | ||
| 30 | 33 | ||||
| 31 | // ---- handlers ---- | 34 | // ---- handlers ---- | ||
@@ -114,6 +117,7 @@ export async function startWorker(): Promise<() => Promise<void>> { | |||||
| 114 | } | 117 | } | ||
| 115 | }; | 118 | }; | ||
| 116 | await tick(); | 119 | await tick(); | ||
| 120 | + if ((await pendingCount()) > 0) for (let k = 0; k < RESOLVE_WORKERS; k++) await queue.send(JOBS.resolveBatch, {}, { singletonKey: `resolve:${k}`, singletonSeconds: 30 }); | ||||
| 117 | const timer = setInterval(tick, 60_000); | 121 | const timer = setInterval(tick, 60_000); | ||
| 118 | // hourly incremental valuation for assets with fresh evidence | 122 | // hourly incremental valuation for assets with fresh evidence | ||
| 119 | const hourly = setInterval(() => void queue.send(JOBS.valuationRebuild, { all: false }, { singletonKey: 'valuation:hourly', singletonSeconds: 3000 }), 3600_000); | 123 | const hourly = setInterval(() => void queue.send(JOBS.valuationRebuild, { all: false }, { singletonKey: 'valuation:hourly', singletonSeconds: 3000 }), 3600_000); | ||
| 120 | 124 | ||||