diff --git a/admin/app/utils/queue_stall_options.ts b/admin/app/utils/queue_stall_options.ts new file mode 100644 index 00000000..2c64041a --- /dev/null +++ b/admin/app/utils/queue_stall_options.ts @@ -0,0 +1,68 @@ +/** + * Per-queue BullMQ stall-recovery options. + * + * The pure, dependency-free core of `getStallOptionsForQueue` in + * `commands/queue/work.ts` (mirrors `decideScanAction` in + * `kb_ingest_decision.ts`). + * + * Two questions decide a queue's entry, and they pull in opposite directions: + * + * 1. **How long does one job legitimately hold the lock?** If that can exceed + * `lockDuration`, BullMQ calls a perfectly healthy job stalled. + * 2. **Is re-running a job safe?** On a stall BullMQ increments the job's `stc` + * counter and re-queues it, failing it only once the counter passes + * `maxStalledCount` (`moveStalledJobsToWait`). It does **not** check whether + * the original is still running. For an idempotent job that recovery is what + * we want; for a job that writes non-idempotent data it is worse than the + * failure it is preventing. + * + * Note that `maxStalledCount: 0` is meaningful and is not the same as leaving + * it unset: the comparison is `stalledCount > maxStalledCount` after an + * increment, so 0 fails the job on the FIRST stall rather than re-queueing it, + * while the BullMQ default of 1 re-queues once. + */ + +export const DEFAULT_LOCK_DURATION = 300_000 +export const LONG_LOCK_DURATION = 1_800_000 + +export interface StallOptions { + lockDuration: number + /** Left undefined to keep BullMQ's default of 1. */ + maxStalledCount?: number +} + +export interface StallQueueNames { + /** ZIM/document embedding — a self-continuing batch chain. */ + embedFile: string + /** Content downloads — resumable. */ + download: string + drugDownload: string + drugIngest: string +} + +export function stallOptionsForQueue(queueName: string, queues: StallQueueNames): StallOptions { + // Long single streams (a ~150 MB resumable HTTP pull, then an unzip + + // JSON-stream ingest at concurrency 1). Re-running a part is safe: the + // download resumes and the ingest upserts by key. + if (queueName === queues.drugDownload || queueName === queues.drugIngest) { + return { lockDuration: LONG_LOCK_DURATION, maxStalledCount: 3 } + } + + // NEVER auto-recover a stalled job on this queue (#1269). Embedding is a + // self-continuing chain, so re-queueing while the original is alive produces + // two chains walking the same file, both writing to Qdrant with freshly + // minted point ids. A failed chain is visible and re-indexable; a forked one + // is silent and corrupts the index. + if (queueName === queues.embedFile) { + return { lockDuration: LONG_LOCK_DURATION, maxStalledCount: 0 } + } + + // The opposite call to embeddings (#1203). A download is resumable and + // re-running one is harmless, but the default maxStalledCount of 1 fails the + // job on a second stall and bypasses its own `attempts: 10` retry policy. + if (queueName === queues.download) { + return { lockDuration: LONG_LOCK_DURATION, maxStalledCount: 3 } + } + + return { lockDuration: DEFAULT_LOCK_DURATION } +} diff --git a/admin/commands/queue/work.ts b/admin/commands/queue/work.ts index dddd1589..bcfc58ff 100644 --- a/admin/commands/queue/work.ts +++ b/admin/commands/queue/work.ts @@ -14,6 +14,7 @@ import { AppAutoUpdateJob } from '#jobs/app_auto_update_job' import { ContentAutoUpdateJob } from '#jobs/content_auto_update_job' import { DownloadDrugDataJob } from '#jobs/download_drug_data_job' import { IngestDrugDataJob } from '#jobs/ingest_drug_data_job' +import { stallOptionsForQueue, type StallOptions } from '../../app/utils/queue_stall_options.js' export default class QueueWork extends BaseCommand { static commandName = 'queue:work' @@ -68,11 +69,10 @@ export default class QueueWork extends BaseCommand { { connection: queueConfig.connection, concurrency: this.getConcurrencyForQueue(queueName), - // lockDuration/maxStalledCount are per-queue. Non-drug queues keep - // the existing default (300000, BullMQ's default maxStalledCount). - // The drug download/ingest queues are NEW per-queue overrides - // (1_800_000 / 3) — see getStallOptionsForQueue — not a change to - // the default applied to every other queue. + // lockDuration/maxStalledCount are per-queue — see + // getStallOptionsForQueue for which queues override the default + // (300000, BullMQ's default maxStalledCount) and why. Queues without + // an entry there are unaffected. lockDuration: stall.lockDuration, ...(stall.maxStalledCount !== undefined ? { maxStalledCount: stall.maxStalledCount } @@ -203,26 +203,17 @@ export default class QueueWork extends BaseCommand { } /** - * Per-queue BullMQ stall-recovery options. - * - * Every queue except the two drug queues keeps the branch default - * (lockDuration 300000, and BullMQ's default maxStalledCount of 1 — left - * unset). The drug download/ingest queues are the ONLY per-queue override: - * each part is a long single stream (a ~150 MB resumable HTTP pull, then an - * unzip + JSON-stream ingest at concurrency 1), so a longer lock plus a - * higher stalled tolerance keeps a transient lock-renewal miss from killing - * the continuation chain ("job stalled more than allowable limit"). + * Per-queue BullMQ stall-recovery options. See `queue_stall_options.ts` for + * the reasoning behind each queue's entry; the decision itself is kept pure + * and unit-tested there. */ - private getStallOptionsForQueue( - queueName: string - ): { lockDuration: number; maxStalledCount?: number } { - if ( - queueName === DownloadDrugDataJob.queue || - queueName === IngestDrugDataJob.queue - ) { - return { lockDuration: 1_800_000, maxStalledCount: 3 } - } - return { lockDuration: 300000 } + private getStallOptionsForQueue(queueName: string): StallOptions { + return stallOptionsForQueue(queueName, { + embedFile: EmbedFileJob.queue, + download: RunDownloadJob.queue, + drugDownload: DownloadDrugDataJob.queue, + drugIngest: IngestDrugDataJob.queue, + }) } /** diff --git a/admin/tests/unit/queue_stall_options.spec.ts b/admin/tests/unit/queue_stall_options.spec.ts new file mode 100644 index 00000000..65d55a07 --- /dev/null +++ b/admin/tests/unit/queue_stall_options.spec.ts @@ -0,0 +1,81 @@ +import * as assert from 'node:assert/strict' +import { test } from 'node:test' + +import { + DEFAULT_LOCK_DURATION, + LONG_LOCK_DURATION, + stallOptionsForQueue, +} from '../../app/utils/queue_stall_options.js' + +const QUEUES = { + embedFile: 'file-embeddings', + download: 'downloads', + drugDownload: 'drug-download', + drugIngest: 'drug-ingest', +} + +const optionsFor = (queue: string) => stallOptionsForQueue(queue, QUEUES) + +test('queues without an entry keep the default lock and BullMQ default maxStalledCount', () => { + for (const queue of ['benchmarks', 'model-downloads', 'update-checks', 'pmtiles-extracts']) { + assert.deepEqual(optionsFor(queue), { lockDuration: DEFAULT_LOCK_DURATION }) + } +}) + +test('an unlisted queue leaves maxStalledCount unset rather than setting it to a value', () => { + // The caller spreads this key only when it is not undefined, so an explicit + // value here would silently change every other queue's behaviour. + assert.equal('maxStalledCount' in optionsFor('benchmarks'), false) +}) + +/** + * Regression: #1269. Embedding is a self-continuing chain, so a stalled-recovery + * re-queue runs a second chain alongside a live one, and both write fresh Qdrant + * points. Failing on the first stall is the only safe option. + */ +test('file-embeddings never auto-recovers a stalled job', () => { + const opts = optionsFor(QUEUES.embedFile) + assert.equal(opts.maxStalledCount, 0, 'must fail on the first stall, not re-queue') + assert.equal(opts.lockDuration, LONG_LOCK_DURATION) +}) + +test('maxStalledCount 0 is distinguishable from unset', () => { + // `stalledCount > maxStalledCount` after an increment: 0 fails on the first + // stall, BullMQ's default of 1 re-queues once first. Guarding against a + // refactor that treats a falsy 0 as "not configured". + const embed = optionsFor(QUEUES.embedFile) + assert.equal(embed.maxStalledCount, 0) + assert.notEqual(embed.maxStalledCount, undefined) +}) + +/** + * Regression: #1203. A download is resumable, so recovery is safe, and the + * default maxStalledCount of 1 was failing jobs outright and bypassing their + * own `attempts: 10` policy. + */ +test('downloads tolerate stalls instead of failing outright', () => { + assert.deepEqual(optionsFor(QUEUES.download), { + lockDuration: LONG_LOCK_DURATION, + maxStalledCount: 3, + }) +}) + +test('the two long-running queues take opposite stall policies', () => { + const embed = optionsFor(QUEUES.embedFile) + const download = optionsFor(QUEUES.download) + assert.equal(embed.lockDuration, download.lockDuration) + assert.notEqual( + embed.maxStalledCount, + download.maxStalledCount, + 'non-idempotent chain and resumable download must not share a recovery policy' + ) +}) + +test('the drug queues are unchanged', () => { + for (const queue of [QUEUES.drugDownload, QUEUES.drugIngest]) { + assert.deepEqual(optionsFor(queue), { + lockDuration: LONG_LOCK_DURATION, + maxStalledCount: 3, + }) + } +})