Show accurate progress for scanning job

Author: Veikkosuhonen

dalai/worker.ts

@@ -117,6 +117,25 @@ const worker = new Worker(
   async (job) => {
     const { s3Bucket, s3Key, outputBucket } = job.data || {}
 
+    /**
+     *  Full progress from 0 to 100
+     */
+    let _progress = 0
+    /**
+     *
+     * @param progress fraction (0-1) of progress of the current section
+     * @param sectionSize the size of the section as a percentage of the whole job (0-100). All section sizes should add up to 100.
+     */
+    const incrementProgress = (progress: number, sectionSize: number) => {
+      _progress += progress * sectionSize
+      job
+        .updateProgress({
+          ragFileId: job.data.ragFileId,
+          progress: _progress,
+        })
+        .catch(() => {})
+    }
+
     console.log(`Processing job ${job.id}`)
 
     if (!s3Bucket || !s3Key) {
@@ -146,6 +165,8 @@ const worker = new Worker(
       await fs.mkdir(outputTextDir, { recursive: true })
       await fs.mkdir(outputImagesDir, { recursive: true })
 
+      incrementProgress(1, 1) // 1% - Setup directories
+
       /**
        * Download the pdf
        */
@@ -155,6 +176,8 @@ const worker = new Worker(
         throw new Error(`Failed to download s3://${s3Bucket}/${s3Key}: ${err.message || err}`)
       }
 
+      incrementProgress(1, 1) // 1% - Download PDF
+
       /**
        * Convert PDF pages to text
        */
@@ -202,6 +225,8 @@ const worker = new Worker(
         throw new Error('PDF to text conversion failed')
       }
 
+      incrementProgress(1, 2) // 2% - PDF to text
+
       /**
        * Convert PDF pages to PNG images
        */
@@ -216,6 +241,8 @@ const worker = new Worker(
         throw new Error('PDF to PNG conversion failed')
       }
 
+      incrementProgress(1, 6) // 6% - PDF to PNGs. Total so far: 10%
+
       /**
        * Transcription & Markdown Generation (with Ollama health/retry, fallback to PDF text)
        */
@@ -268,6 +295,10 @@ const worker = new Worker(
             const txt = data?.response || ''
             await fs.writeFile(existingTxtPath, txt, 'utf-8')
             console.log(`Job ${job.id}: transcription complete for page ${pngPage.pageNumber}/${pngPages.length}`)
+
+            const pageProgress = 0.5 / pngPages.length // Halfway through the page processing
+            incrementProgress(pageProgress, 87) // 87% - VLM & Markdown
+
             return txt
           }, RETRY_COUNT)
           finalText = await retryOllamaCall(async () => {
@@ -311,6 +342,7 @@ const worker = new Worker(
             }
             await fs.writeFile(existingMdPath, text, 'utf-8')
             console.log(`Job ${job.id}: markdown generation complete for page ${pngPage.pageNumber}/${pngPages.length}`)
+
             return text
           }, RETRY_COUNT)
         } catch (error) {
@@ -320,6 +352,9 @@ const worker = new Worker(
           await fs.writeFile(existingMdPath, finalText, 'utf-8')
         }
 
+        const pageProgress = 0.5 / pngPages.length // Second half of the page processing done
+        incrementProgress(pageProgress, 87) // 87% - VLM & Markdown. Total so far: 97%
+
         resultingMarkdown += `\n\n${finalText}`
       }
 
@@ -336,6 +371,8 @@ const worker = new Worker(
         throw new Error(`Failed uploading outputs to s3://${outputBucket}: ${err.message || err}`)
       }
 
+      incrementProgress(1, 3) // 97 + 3 = 100% - Upload results
+
       return {
         input: { bucket: s3Bucket, key: s3Key },
         output: { bucket: outputBucket },

src/client/components/Rag/RagFile.tsx

@@ -5,7 +5,6 @@ import type { RagFileAttributes } from '../../../shared/types'
 import { Box, Button, Container, Link, Typography } from '@mui/material'
 import { RagFileInfo } from './RagFileDetails'
 import type { RagIndexAttributes } from '../../../server/db/models/ragIndex'
-import { Chunk } from './Chunk'
 import { useDeleteRagFileMutation, useDeleteRagFileTextMutation } from './api'
 import { useTranslation } from 'react-i18next'
 import Markdown from 'react-markdown'

src/client/components/Rag/RagFileDetails.tsx

@@ -54,6 +54,7 @@ export const RagFileInfo: React.FC<{
             <TableCell>
               <Box display="flex" alignItems="end" gap={1}>
                 {IngestionPipelineStages[file.pipelineStage]}
+                {file.pipelineStage === 'parsing' ? file.progress && ` (${file.progress.toFixed()}%)` : null}
                 {ProgressIcon[file.pipelineStage]}
               </Box>
             </TableCell>

src/server/db/migrations/20250925_00_add_progress_to_ragfile.ts

@@ -0,0 +1,14 @@
+import { DataTypes } from 'sequelize'
+
+import type { Migration } from '../connection'
+
+export const up: Migration = async ({ context: queryInterface }) => {
+  await queryInterface.addColumn('rag_files', 'progress', {
+    type: DataTypes.FLOAT,
+    allowNull: true,
+  })
+}
+
+export const down: Migration = async ({ context: queryInterface }) => {
+  await queryInterface.removeColumn('rag_files', 'progress')
+}

src/server/db/models/ragFile.ts

@@ -11,6 +11,8 @@ class RagFile extends Model<InferAttributes<RagFile>, InferCreationAttributes<Ra
 
   declare pipelineStage: IngestionPipelineStageKey
 
+  declare progress: CreationOptional<number | null>
+
   declare error: CreationOptional<string | null>
 
   declare filename: string
@@ -58,6 +60,10 @@ RagFile.init(
       type: DataTypes.STRING,
       allowNull: false,
     },
+    progress: {
+      type: DataTypes.FLOAT,
+      allowNull: true,
+    },
     error: {
       type: DataTypes.STRING,
       allowNull: true,

src/server/routes/rag/ragIndex.ts

@@ -13,6 +13,7 @@ import { SearchSchema } from '../../../shared/rag'
 import { S3_BUCKET } from '../../util/config'
 import { s3Client } from '../../util/s3client'
 import { ingestRagFiles } from '../../services/rag/ingestion'
+import { getPdfParsingJobId, queue } from 'src/server/services/jobs/pdfParsing.job'
 
 const ragIndexRouter = Router()
 

src/server/services/jobs/pdfParsing.job.ts

@@ -1,6 +1,6 @@
 import IORedis from 'ioredis'
 import { BMQ_REDIS_CA, BMQ_REDIS_CERT, BMQ_REDIS_HOST, BMQ_REDIS_KEY, BMQ_REDIS_PORT, S3_BUCKET } from '../../util/config'
-import { Queue, QueueEvents } from 'bullmq'
+import { Job, Queue, QueueEvents } from 'bullmq'
 import { FileStore } from '../rag/fileStore'
 import { RagFile } from '../../db/models'
 
@@ -24,30 +24,45 @@ if (BMQ_REDIS_CA !== 'none') {
 
 const connection = new IORedis(creds)
 
-const queue = new Queue('llama-scan-queue', {
+export const queue = new Queue('llama-scan-queue', {
   connection,
 })
 
+export const getPdfParsingJobId = (ragFile: RagFile) => {
+  const s3Key = FileStore.getRagFileKey(ragFile)
+  return `scan:${S3_BUCKET}/${s3Key}`
+}
+
 export const pdfQueueEvents = new QueueEvents('llama-scan-queue', { connection })
 
+pdfQueueEvents.on('progress', async (progressEvent) => {
+  const data = progressEvent.data as { ragFileId: number; progress: number }
+  await RagFile.update({ progress: data.progress }, { where: { id: data.ragFileId } })
+})
+
+type PDFJobData = {
+  s3Bucket: string
+  s3Key: string
+  outputBucket: string
+  ragFileId: number
+}
+
 /**
  * Adds a pdf parsing job to the queue. The file must be uploaded to S3 beforehand. The jobId is based on the ragFile - resubmitting with the same jobId while the previous job is running has no effect.
  * @param ragFile
  * @returns the job
  */
 export const submitPdfParsingJob = async (ragFile: RagFile) => {
   const s3Key = FileStore.getRagFileKey(ragFile)
-  const jobId = `scan:${S3_BUCKET}/${s3Key}`
+  const jobId = getPdfParsingJobId(ragFile)
   console.log(`Submitting PDF parsing job ${jobId}`)
-  const job = await queue.add(
-    jobId,
-    {
-      s3Bucket: S3_BUCKET,
-      s3Key,
-      outputBucket: S3_BUCKET,
-    },
-    { jobId, removeOnComplete: true, removeOnFail: true },
-  )
+  const jobData: PDFJobData = {
+    s3Bucket: S3_BUCKET,
+    s3Key,
+    outputBucket: S3_BUCKET,
+    ragFileId: ragFile.id,
+  }
+  const job = await queue.add(jobId, jobData, { jobId, removeOnComplete: true, removeOnFail: true })
 
   return job
 }

src/shared/ingestion.ts

@@ -2,7 +2,7 @@ export const IngestionPipelineStages = {
   uploading: 'Uploading',
   parsing: 'Scanning',
   embedding: 'Creating embeddings',
-  storing: 'Storing',
+  storing: 'Waiting',
   completed: 'Completed',
   error: 'Error',
 } as const

src/shared/types.ts

@@ -24,6 +24,7 @@ export type RagFileAttributes = {
   updatedAt: string
   ragIndexId: number
   pipelineStage: IngestionPipelineStageKey
+  progress: number | null
   fileType: string
   fileSize: number
   numChunks: number | null