From 74cef75153528e13b60bfd00255f2dbc2f5a24c6 Mon Sep 17 00:00:00 2001 From: Chris Sherwood Date: Tue, 12 May 2026 18:05:18 -0700 Subject: [PATCH] fix(RAG): unbreak multi-batch ZIM ingestion (jobId dedupe) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit EmbedFileJob.dispatch() uses a deterministic per-file jobId (sha256(filePath).slice(0,16)) for every batch. The parent batch's handle() calls EmbedFileJob.dispatch({ batchOffset }) before returning, so the parent is still in `active` state and locked when the continuation tries to enqueue. BullMQ silently returns the locked parent instead of creating a new job — and in newer BullMQ versions it does so without throwing, so the existing `catch (error.message.includes('job already exists'))` branch never fires. After the parent completes, its entry stays in the `completed` ZSET (held by `removeOnComplete: { count: 50 }`), continuing to trip jobId dedupe for any subsequent re-dispatch attempts. Result: every NOMAD install since 2026-02-08 (feat: zim content embedding) with a multi-batch ZIM (wikipedia, cooking SE, ifixit, lrnselfreliance, etc.) has only the first 50 articles indexed in qdrant. The RAG feature has been silently degraded for ~3 months — the user sees the file appear in their KB, qdrant accumulates ~50 articles' worth of vectors, and pagination quietly halts. No error surfaces anywhere. Fix: dispatch() skips the deterministic jobId for continuation batches (batchOffset > 0), letting BullMQ auto-generate a unique one so each batch stacks as an independent queue entry. Initial dispatches keep the deterministic jobId so re-triggering an install (UI re-click, sync rescan) remains idempotent. The existing 'job already exists' branch is now gated on !isContinuation, since by construction continuation batches will never hit dedupe. Validated on NOMAD8 (RX 6800 / Threadripper 3960X, rc.3 + this patch): devdocs_en_python (~1,500 chunks across multiple batches) correctly paginates end-to-end. admin.log shows the expected sequence of "Dispatched embedding job for file: X (continuation @ offset N)" followed by "Starting embedding process for: X (batch offset: N)" for each batch. Co-Authored-By: Claude Opus 4.7 (1M context) --- admin/app/jobs/embed_file_job.ts | 54 ++++++++++++++++++++++---------- 1 file changed, 38 insertions(+), 16 deletions(-) diff --git a/admin/app/jobs/embed_file_job.ts b/admin/app/jobs/embed_file_job.ts index eef45a8..c7566fa 100644 --- a/admin/app/jobs/embed_file_job.ts +++ b/admin/app/jobs/embed_file_job.ts @@ -207,36 +207,58 @@ export class EmbedFileJob { static async dispatch(params: EmbedFileJobParams) { const queueService = new QueueService() const queue = queueService.getQueue(this.queue) - const jobId = this.getJobId(params.filePath) + + // Continuation batches (batchOffset > 0) must NOT reuse the deterministic + // per-file jobId. Two BullMQ dedupe paths would otherwise silently swallow them: + // 1) The parent batch's handle() calls dispatch() before returning, so the + // parent job is still `active` and locked — queue.add() with the same + // jobId returns the locked parent rather than enqueueing the new batch. + // 2) After the parent completes, its entry stays in `completed` (held by + // `removeOnComplete: { count: 50 }`), still tripping jobId dedupe. + // Letting BullMQ auto-generate a unique jobId for continuation batches stacks + // them as independent queue entries that each process via handle(). + // Initial dispatches keep the deterministic jobId so re-triggering an install + // (UI re-click, sync rescan, etc.) is still idempotent. + const isContinuation = !!(params.batchOffset && params.batchOffset > 0) + const initialJobId = this.getJobId(params.filePath) + + const jobOptions: Parameters[2] = { + attempts: 30, + backoff: { + type: 'fixed', + delay: 60000, // Check every 60 seconds for service readiness + }, + removeOnComplete: { count: 50 }, // Keep last 50 completed jobs for history + removeOnFail: { count: 20 }, // Keep last 20 failed jobs for debugging + } + if (!isContinuation) { + jobOptions.jobId = initialJobId + } try { - const job = await queue.add(this.key, params, { - jobId, - attempts: 30, - backoff: { - type: 'fixed', - delay: 60000, // Check every 60 seconds for service readiness - }, - removeOnComplete: { count: 50 }, // Keep last 50 completed jobs for history - removeOnFail: { count: 20 } // Keep last 20 failed jobs for debugging - }) + const job = await queue.add(this.key, params, jobOptions) - logger.info(`[EmbedFileJob] Dispatched embedding job for file: ${params.fileName}`) + const continuationLabel = isContinuation + ? ` (continuation @ offset ${params.batchOffset})` + : '' + logger.info( + `[EmbedFileJob] Dispatched embedding job for file: ${params.fileName}${continuationLabel}` + ) return { job, created: true, - jobId, + jobId: job.id ?? initialJobId, message: `File queued for embedding: ${params.fileName}`, } } catch (error) { - if (error.message && error.message.includes('job already exists')) { - const existing = await queue.getJob(jobId) + if (!isContinuation && error.message && error.message.includes('job already exists')) { + const existing = await queue.getJob(initialJobId) logger.info(`[EmbedFileJob] Job already exists for file: ${params.fileName}`) return { job: existing, created: false, - jobId, + jobId: initialJobId, message: `Embedding job already exists for: ${params.fileName}`, } }