Fix BullMQ job stall: increase stalled interval and add periodic progress updates

Author: StephaneWamba

services/knowledge-base/src/services/processor.ts

@@ -42,16 +42,16 @@ function findSentenceBoundary(text: string, index: number): number {
 
 const logger = createLogger('knowledge-base-service:processor')
 
-export async function processDocument(documentId: string) {
+export async function processDocument(documentId: string, job?: { updateProgress: (progress: number) => Promise<void> }) {
   const timeoutPromise = new Promise<never>((_, reject) => {
     setTimeout(() => {
       reject(new Error(`Document processing timeout after ${PROCESSING_CONSTANTS.TIMEOUT_MS / 1000} seconds`))
     }, PROCESSING_CONSTANTS.TIMEOUT_MS)
   })
 
-  try {
+    try {
     await Promise.race([
-      processDocumentInternal(documentId),
+      processDocumentInternal(documentId, job),
       timeoutPromise,
     ])
   } catch (error) {
@@ -79,7 +79,7 @@ export async function processDocument(documentId: string) {
   }
 }
 
-async function processDocumentInternal(documentId: string) {
+async function processDocumentInternal(documentId: string, job?: { updateProgress: (progress: number) => Promise<void> }) {
   const supabase = getSupabase()
 
     await supabase
@@ -268,6 +268,14 @@ async function processDocumentInternal(documentId: string) {
       // Move to next window
       windowStart = windowEnd
 
+      // Update job progress periodically to prevent stall detection
+      if (job && windowNumber % 3 === 0) {
+        const progress = Math.min(10 + Math.floor((windowStart / textLength) * 80), 90)
+        await job.updateProgress(progress).catch(() => {
+          // Ignore progress update errors (job might be completed)
+        })
+      }
+      
       // Force GC every few windows for large documents
       if (global.gc && windowNumber % 5 === 0) {
         global.gc()

services/knowledge-base/src/services/queue.ts

@@ -109,7 +109,8 @@ async function initializeQueue(): Promise<void> {
 
           try {
             await job.updateProgress(10)
-            await processDocument(documentId)
+            // Pass job for progress updates during processing
+            await processDocument(documentId, job)
             await job.updateProgress(100)
           } catch (error) {
             const errorMessage = error instanceof Error ? error.message : String(error)
@@ -123,6 +124,11 @@ async function initializeQueue(): Promise<void> {
         {
           connection: redisConnection,
           concurrency: 2,
+          // Increase stalled interval to allow long-running document processing
+          // Default is 30 seconds, but large documents can take 5+ minutes
+          maxStalledCount: 0, // Don't mark as stalled (we handle timeouts in processor)
+          stalledInterval: 10 * 60 * 1000, // 10 minutes (longer than max processing time)
+          lockDuration: 10 * 60 * 1000, // 10 minutes - how long job is locked to this worker
         }
       )