fix(queue): stop stalled-job recovery forking the embedding chain

Fixes #1269. Fixes #1203.

`file-embeddings` and `downloads` both routinely outlive the default
5 minute lock, so BullMQ calls healthy jobs stalled on both. They need
opposite recovery policies, which is why neither was covered by the
existing drug-queue override.

Embedding is a self-continuing chain, so re-queueing a stalled job while
the original is alive runs two chains over the same file. Both write to
Qdrant with freshly minted point ids, so the second cannot overwrite the
first, and the queue's concurrency of 2 means the fork never contends
for a slot. maxStalledCount 0 fails on the first stall instead.

A download is resumable, so recovery is safe there; the default of 1 was
failing jobs outright and bypassing their own attempts: 10 policy.

The decision moves to a pure, unit-tested util alongside the other
decision helpers.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LtGWCVMPYn6qPWVYiycK74

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 }
+}

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,
+    })
   }
 
   /**

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,
+    })
+  }
+})