workers: close orphaned runs, dedupe queued crawls, one crawl per connector at a time
1 changed file +10 −0
modified
workers/main.ts
+10 −0
@@ -29,10 +29,20 @@ export async function startWorker(): Promise<() => Promise<void>> { | |||||
| 29 | // Single-worker deployment: jobs left 'active' by a previous process (restart/crash) would wait for | 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. | 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'`); | 31 | await db().execute(sql`update pgboss.job set state = 'retry', started_on = null where state = 'active' and started_on < now() - interval '90 seconds'`); | ||
| 32 | + // Runs left 'running' by a dead process can never finish: close them so the per-connector guard below works. | ||||
| 33 | + await db().execute(sql`update connector_runs set status = 'failed', finished_at = now(), error = 'worker restarted' where status = 'running'`); | ||||
| 34 | + // Collapse duplicate queued crawls per connector (keep the oldest). | ||||
| 35 | + await db().execute(sql`delete from pgboss.job j using pgboss.job k where j.name = 'crawl.run' and k.name = 'crawl.run' and j.state in ('created','retry') and k.state in ('created','retry') and j.data->>'connectorId' = k.data->>'connectorId' and j.created_on > k.created_on`); | ||||
| 32 | const concurrency = env().WORKER_CONCURRENCY; | 36 | const concurrency = env().WORKER_CONCURRENCY; | ||
| 33 | 37 | ||||
| 34 | // ---- handlers ---- | 38 | // ---- handlers ---- | ||
| 35 | await queue.work<{ connectorId: string; mode?: 'incremental' | 'backfill' | 'probe'; limit?: number; trigger?: string }>(JOBS.crawlRun, { concurrency: Math.max(1, Math.floor(concurrency / 2)), pollingIntervalSeconds: 5 }, async (data) => { | 39 | await queue.work<{ connectorId: string; mode?: 'incremental' | 'backfill' | 'probe'; limit?: number; trigger?: string }>(JOBS.crawlRun, { concurrency: Math.max(1, Math.floor(concurrency / 2)), pollingIntervalSeconds: 5 }, async (data) => { | ||
| 40 | + // One crawl per connector at a time (duplicates waste engine credits and crawl slots). | ||||
| 41 | + const [running] = (await db().execute(sql`select 1 from connector_runs where connector_id = ${data.connectorId} and status = 'running' and started_at > now() - interval '6 hours' limit 1`)) as unknown as unknown[]; | ||||
| 42 | + if (running) { | ||||
| 43 | + log.info({ connector: data.connectorId }, 'crawl already running; skipping duplicate job'); | ||||
| 44 | + return; | ||||
| 45 | + } | ||||
| 36 | const res = await runCrawl(data.connectorId, { mode: data.mode ?? 'incremental', limit: data.limit, trigger: data.trigger ?? 'schedule' }); | 46 | const res = await runCrawl(data.connectorId, { mode: data.mode ?? 'incremental', limit: data.limit, trigger: data.trigger ?? 'schedule' }); | ||
| 37 | if (res.recordsRaw > 0) await queue.send(JOBS.normalizeBatch, { connectorId: data.connectorId }, { singletonKey: `normalize:${data.connectorId}`, singletonSeconds: 30 }); | 47 | if (res.recordsRaw > 0) await queue.send(JOBS.normalizeBatch, { connectorId: data.connectorId }, { singletonKey: `normalize:${data.connectorId}`, singletonSeconds: 30 }); | ||
| 38 | }); | 48 | }); | ||
| 39 | 49 | ||||