From d28eb9be59b9c283f5c8d28e848f050ac4613ee1 Mon Sep 17 00:00:00 2001 From: Chris Sherwood Date: Wed, 13 May 2026 12:43:24 -0700 Subject: [PATCH] fix(RAG): report ZIM ingestion progress in overall-file frame MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Before this change, the Active Downloads / Processing Queue UI showed the ingestion progress gauge jumping wildly during multi-batch ZIM ingestion (e.g. 5% → 88% → 27% → 5% → 56% → 36% over ~60 seconds for cooking SE). Each continuation batch is a separate BullMQ job, and `EmbedFileJob.handle()` reported `job.progress` in two different reference frames depending on where it was in the batch lifecycle: - During-batch (via the onProgress callback): 5% → 95% scaled across "% through this batch's chunks" - End-of-batch (just before dispatching the next): overwritten to `(nextOffset / totalArticles) * 100` — % through the whole file - Next continuation batch starts with progress = 5% explicitly, then climbs through the per-batch range again `listActiveJobs()` returns the latest active BullMQ job's progress. With GPU-accelerated ingestion completing a batch every ~4 seconds, the UI saw the jobId rotate constantly and the gauge whipsaw between the two reference frames. `totalArticles` was already wired through the EmbedFileJob params shape and used end-of-batch — but RagService never actually populated it, so any frame-scaling that depended on it silently fell back to the per-batch range. Two fixes together: 1. `ZIMExtractionService.extractZIMContent()` now returns `{ chunks: ZIMContentChunk[]; totalArticles: number }` instead of a raw chunks array, surfacing `archive.articleCount` to the caller. Single caller (rag_service) updated to destructure. 2. `RagService.processZimFile()` includes `totalArticles` in its result so `EmbedFileJob.dispatch()` can propagate it to the continuation batch (which the existing code already does via `totalArticles: totalArticles || result.totalArticles`). 3. `EmbedFileJob`'s onProgress callback scales the service-reported per-batch percent into the overall-file frame when `totalArticles` is known: `((batchOffset + (percent/100) * ZIM_BATCH_SIZE) / totalArticles) * 100`. Capped at 99% to leave room for the explicit 100% set at file completion. Falls back to the original 5-95% range for single-batch files (uploaded PDFs/txts) where totalArticles is undefined — the gauge then represents % through the only batch, which is what the UI expects for one-shot files. Validated on NOMAD8 (RX 6800, ROCm-accelerated nomic): - devdocs python (small, ~1500 articles): batch progressions seen monotonically across continuation jobIds: 1501@30% → 1510@33% → 1514@43% → 1518@52%. - ifixit (huge, ~100k articles): stays near 3% for the first many batches at offset 0..3000 — correct, the file is enormous. - wikipedia_en_medicine (large, ~70k articles): stays near 0-1% for the first batches — also correct. - Brief 0-5% blip on continuation handoff (the explicit `safeUpdateProgress(job, 5)` at batch start, before the first onProgress callback fires) — visible but quickly resolves to the overall-frame value. No more 5% ↔ 88% chaos. --- admin/app/jobs/embed_file_job.ts | 21 ++++++++++++++++++-- admin/app/services/rag_service.ts | 11 +++++----- admin/app/services/zim_extraction_service.ts | 7 +++++-- 3 files changed, 30 insertions(+), 9 deletions(-) diff --git a/admin/app/jobs/embed_file_job.ts b/admin/app/jobs/embed_file_job.ts index 426608a..5a3e682 100644 --- a/admin/app/jobs/embed_file_job.ts +++ b/admin/app/jobs/embed_file_job.ts @@ -7,6 +7,7 @@ import { OllamaService } from '#services/ollama_service' import { createHash } from 'crypto' import logger from '@adonisjs/core/services/logger' import fs from 'node:fs/promises' +import { ZIM_BATCH_SIZE } from '../../constants/zim_extraction.js' export interface EmbedFileJobParams { filePath: string @@ -93,9 +94,25 @@ export class EmbedFileJob { logger.info(`[EmbedFileJob] Processing file: ${filePath}`) - // Progress callback: maps service-reported 0-100% into the 5-95% job range + // Progress callback. For multi-batch ZIM ingestions, scale the service-reported + // 0-100% (which is % through the current batch's chunks) into the overall-file + // frame so the UI gauge climbs monotonically across the many continuation jobs + // BullMQ creates per file. Without this, every new continuation jobId resets the + // gauge to ~5% and the user sees ingestion progress "jumping around" between + // each batch's local frame and the end-of-batch overall-file overwrite below. + // + // For single-batch files (uploaded PDFs, txts) totalArticles is undefined and + // we fall back to the original 5-95% per-job range, which is what the UI expects + // for a one-shot file with no continuations. const onProgress = async (percent: number) => { - await this.safeUpdateProgress(job, Math.min(95, Math.round(5 + percent * 0.9))) + const useOverallFrame = totalArticles && totalArticles > 0 + if (useOverallFrame) { + const articlesDone = (batchOffset || 0) + (percent / 100) * ZIM_BATCH_SIZE + const overallPercent = Math.min(99, Math.round((articlesDone / totalArticles) * 100)) + await this.safeUpdateProgress(job, overallPercent) + } else { + await this.safeUpdateProgress(job, Math.min(95, Math.round(5 + percent * 0.9))) + } } // Process and embed the file diff --git a/admin/app/services/rag_service.ts b/admin/app/services/rag_service.ts index bd5371d..2a31bd1 100644 --- a/admin/app/services/rag_service.ts +++ b/admin/app/services/rag_service.ts @@ -500,13 +500,13 @@ export class RagService { `[RAG] Extracting ZIM content (batch: offset=${startOffset}, size=${ZIM_BATCH_SIZE})` ) - const zimChunks = await zimExtractionService.extractZIMContent(filepath, { - startOffset, - batchSize: ZIM_BATCH_SIZE, - }) + const { chunks: zimChunks, totalArticles } = await zimExtractionService.extractZIMContent( + filepath, + { startOffset, batchSize: ZIM_BATCH_SIZE } + ) logger.info( - `[RAG] Extracted ${zimChunks.length} chunks from ZIM file with enhanced metadata` + `[RAG] Extracted ${zimChunks.length} chunks from ZIM file with enhanced metadata (file totalArticles=${totalArticles})` ) // Process each chunk individually with its metadata @@ -582,6 +582,7 @@ export class RagService { chunks: totalChunks, hasMoreBatches, articlesProcessed: articlesInBatch, + totalArticles, } } diff --git a/admin/app/services/zim_extraction_service.ts b/admin/app/services/zim_extraction_service.ts index c48b72d..196add9 100644 --- a/admin/app/services/zim_extraction_service.ts +++ b/admin/app/services/zim_extraction_service.ts @@ -40,7 +40,10 @@ export class ZIMExtractionService { * @param filePath - Path to the ZIM file * @param opts - Options including maxArticles, strategy, onProgress, startOffset, and batchSize */ - async extractZIMContent(filePath: string, opts: ExtractZIMContentOptions = {}): Promise { + async extractZIMContent( + filePath: string, + opts: ExtractZIMContentOptions = {} + ): Promise<{ chunks: ZIMContentChunk[]; totalArticles: number }> { try { logger.info(`[ZIMExtractionService]: Processing ZIM file at path: ${filePath}`) @@ -161,7 +164,7 @@ export class ZIMExtractionService { textPreview: c.text.substring(0, 100) }))) logger.debug("Total structured sections extracted:", toReturn.length) - return toReturn + return { chunks: toReturn, totalArticles: archive.articleCount } } catch (error) { logger.error('Error processing ZIM file:', error) throw error