fix(RAG): unbreak multi-batch ZIM ingestion (jobId dedupe)
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) <noreply@anthropic.com>
This commit is contained in:
parent
5a72560929
commit
7acc53444d
|
|
@ -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<typeof queue.add>[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}`,
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue