fix(knowledge): reclaim stale processing attempts

Author: TheodoreSpeaks

apps/sim/lib/knowledge/documents/processing-claim.test.ts

@@ -0,0 +1,70 @@
+/**
+ * @vitest-environment node
+ */
+
+import { dbChainMockFns, resetDbChainMock } from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import {
+  KNOWLEDGE_DOCUMENT_PROCESSING_STALE_THRESHOLD_MS,
+  reclaimStaleDocumentProcessingClaim,
+} from '@/lib/knowledge/documents/processing-claim'
+
+const NOW = new Date('2026-08-11T12:00:00.000Z')
+
+describe('reclaimStaleDocumentProcessingClaim', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  it('leaves an active processing claim untouched', async () => {
+    const reclaimed = await reclaimStaleDocumentProcessingClaim({
+      knowledgeBaseId: 'knowledge-base-1',
+      documentId: 'document-1',
+      processingStartedAt: new Date(
+        NOW.getTime() - KNOWLEDGE_DOCUMENT_PROCESSING_STALE_THRESHOLD_MS
+      ),
+      now: NOW,
+    })
+
+    expect(reclaimed).toBe(false)
+    expect(dbChainMockFns.set).not.toHaveBeenCalled()
+  })
+
+  it.each([null, new Date(NOW.getTime() - KNOWLEDGE_DOCUMENT_PROCESSING_STALE_THRESHOLD_MS - 1)])(
+    'reopens an abandoned processing claim started at %s',
+    async (processingStartedAt) => {
+      dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'document-1' }])
+
+      const reclaimed = await reclaimStaleDocumentProcessingClaim({
+        knowledgeBaseId: 'knowledge-base-1',
+        documentId: 'document-1',
+        processingStartedAt,
+        now: NOW,
+      })
+
+      expect(reclaimed).toBe(true)
+      expect(dbChainMockFns.set).toHaveBeenCalledWith({
+        processingStatus: 'pending',
+        processingStartedAt: null,
+        processingCompletedAt: null,
+        processingError: null,
+      })
+    }
+  )
+
+  it('does not report success when the original claim changed before the update', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce([])
+
+    const reclaimed = await reclaimStaleDocumentProcessingClaim({
+      knowledgeBaseId: 'knowledge-base-1',
+      documentId: 'document-1',
+      processingStartedAt: new Date(
+        NOW.getTime() - KNOWLEDGE_DOCUMENT_PROCESSING_STALE_THRESHOLD_MS - 1
+      ),
+      now: NOW,
+    })
+
+    expect(reclaimed).toBe(false)
+  })
+})

apps/sim/lib/knowledge/documents/processing-claim.ts

@@ -0,0 +1,57 @@
+import { db } from '@sim/db'
+import { document } from '@sim/db/schema'
+import { and, eq, isNull } from 'drizzle-orm'
+
+export const KNOWLEDGE_DOCUMENT_PROCESSING_STALE_THRESHOLD_MS = 10 * 60 * 1000
+
+interface ReclaimStaleDocumentProcessingClaimParams {
+  knowledgeBaseId: string
+  documentId: string
+  processingStartedAt: Date | null
+  now?: Date
+}
+
+/**
+ * Reopens an abandoned processing attempt using its start time as a compare-and-set token.
+ * The former worker's timestamp-guarded writes then cannot commit after the claim is reclaimed.
+ */
+export async function reclaimStaleDocumentProcessingClaim({
+  knowledgeBaseId,
+  documentId,
+  processingStartedAt,
+  now = new Date(),
+}: ReclaimStaleDocumentProcessingClaimParams): Promise<boolean> {
+  if (
+    processingStartedAt &&
+    now.getTime() - processingStartedAt.getTime() <=
+      KNOWLEDGE_DOCUMENT_PROCESSING_STALE_THRESHOLD_MS
+  ) {
+    return false
+  }
+
+  const processingStartedAtGuard = processingStartedAt
+    ? eq(document.processingStartedAt, processingStartedAt)
+    : isNull(document.processingStartedAt)
+  const [reclaimed] = await db
+    .update(document)
+    .set({
+      processingStatus: 'pending',
+      processingStartedAt: null,
+      processingCompletedAt: null,
+      processingError: null,
+    })
+    .where(
+      and(
+        eq(document.id, documentId),
+        eq(document.knowledgeBaseId, knowledgeBaseId),
+        eq(document.processingStatus, 'processing'),
+        processingStartedAtGuard,
+        eq(document.userExcluded, false),
+        isNull(document.archivedAt),
+        isNull(document.deletedAt)
+      )
+    )
+    .returning({ id: document.id })
+
+  return Boolean(reclaimed)
+}

apps/sim/lib/knowledge/documents/processing-outbox-handler.test.ts

@@ -7,13 +7,18 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'
 const mocks = vi.hoisted(() => ({
   getKnowledgeDocument: vi.fn(),
   processDocumentsWithQueue: vi.fn(),
+  reclaimStaleDocumentProcessingClaim: vi.fn(),
 }))
 
 vi.mock('@/lib/knowledge/documents/service', () => ({
   getKnowledgeDocument: mocks.getKnowledgeDocument,
   processDocumentsWithQueue: mocks.processDocumentsWithQueue,
 }))
 
+vi.mock('@/lib/knowledge/documents/processing-claim', () => ({
+  reclaimStaleDocumentProcessingClaim: mocks.reclaimStaleDocumentProcessingClaim,
+}))
+
 import type { BillingAttributionSnapshot } from '@/lib/billing/core/billing-attribution'
 import type { OutboxEventContext } from '@/lib/core/outbox/service'
 import { KNOWLEDGE_DOCUMENT_PROCESSING_OUTBOX_EVENT } from '@/lib/knowledge/documents/processing-outbox-event'
@@ -71,6 +76,7 @@ describe('knowledge document processing outbox handler', () => {
     vi.clearAllMocks()
     mocks.getKnowledgeDocument.mockResolvedValue(DOCUMENT)
     mocks.processDocumentsWithQueue.mockResolvedValue(undefined)
+    mocks.reclaimStaleDocumentProcessingClaim.mockResolvedValue(false)
   })
 
   it('dispatches the authoritative document with the stable outbox event id', async () => {
@@ -106,18 +112,58 @@ describe('knowledge document processing outbox handler', () => {
   )
 
   it('keeps the event retryable while an earlier processing attempt is active', async () => {
+    const processingStartedAt = new Date()
     mocks.getKnowledgeDocument.mockResolvedValueOnce({
       ...DOCUMENT,
       processingStatus: 'processing',
+      processingStartedAt,
     })
 
     await expect(handler()(PAYLOAD, createContext())).rejects.toThrow(
       'Knowledge document document-1 is already being processed'
     )
 
+    expect(mocks.reclaimStaleDocumentProcessingClaim).toHaveBeenCalledWith({
+      knowledgeBaseId: 'knowledge-base-1',
+      documentId: 'document-1',
+      processingStartedAt,
+    })
     expect(mocks.processDocumentsWithQueue).not.toHaveBeenCalled()
   })
 
+  it('reclaims and redispatches an abandoned processing attempt', async () => {
+    const processingStartedAt = new Date('2026-08-11T11:00:00.000Z')
+    mocks.getKnowledgeDocument.mockResolvedValueOnce({
+      ...DOCUMENT,
+      processingStatus: 'processing',
+      processingStartedAt,
+    })
+    mocks.reclaimStaleDocumentProcessingClaim.mockResolvedValueOnce(true)
+
+    await handler()(PAYLOAD, createContext('outbox-event-retry'))
+
+    expect(mocks.reclaimStaleDocumentProcessingClaim).toHaveBeenCalledWith({
+      knowledgeBaseId: 'knowledge-base-1',
+      documentId: 'document-1',
+      processingStartedAt,
+    })
+    expect(mocks.processDocumentsWithQueue).toHaveBeenCalledWith(
+      [
+        {
+          documentId: 'document-1',
+          filename: 'guide.pdf',
+          fileUrl: '/api/files/serve/kb%2Fguide.pdf?context=knowledge-base',
+          fileSize: 128,
+          mimeType: 'application/pdf',
+        },
+      ],
+      'knowledge-base-1',
+      { recipe: 'default', lang: 'en' },
+      'outbox-event-retry',
+      BILLING_ATTRIBUTION
+    )
+  })
+
   it('propagates dispatch failures so the outbox schedules a retry', async () => {
     const failure = new Error('queue unavailable')
     mocks.processDocumentsWithQueue.mockRejectedValueOnce(failure)

apps/sim/lib/knowledge/documents/processing-outbox-handler.ts

@@ -1,5 +1,6 @@
 import { assertBillingAttributionSnapshot } from '@/lib/billing/core/billing-attribution'
 import type { OutboxHandler, OutboxHandlerRegistry } from '@/lib/core/outbox/service'
+import { reclaimStaleDocumentProcessingClaim } from '@/lib/knowledge/documents/processing-claim'
 import {
   KNOWLEDGE_DOCUMENT_PROCESSING_OUTBOX_EVENT,
   type KnowledgeDocumentProcessingOutboxPayload,
@@ -60,7 +61,14 @@ const processKnowledgeDocument: OutboxHandler<unknown> = async (rawPayload, cont
   const document = await getKnowledgeDocument(payload.knowledgeBaseId, payload.documentId)
   if (!document || document.processingStatus === 'completed') return
   if (document.processingStatus === 'processing') {
-    throw new Error(`Knowledge document ${document.id} is already being processed`)
+    const reclaimed = await reclaimStaleDocumentProcessingClaim({
+      knowledgeBaseId: payload.knowledgeBaseId,
+      documentId: document.id,
+      processingStartedAt: document.processingStartedAt,
+    })
+    if (!reclaimed) {
+      throw new Error(`Knowledge document ${document.id} is already being processed`)
+    }
   }
 
   context.signal.throwIfAborted()

apps/sim/lib/knowledge/documents/service.ts

@@ -64,6 +64,7 @@ import {
   mergeDurableSecretProvenance,
 } from '@/lib/execution/durable-secret-provenance'
 import { processDocument } from '@/lib/knowledge/documents/document-processor'
+import { KNOWLEDGE_DOCUMENT_PROCESSING_STALE_THRESHOLD_MS } from '@/lib/knowledge/documents/processing-claim'
 import { enqueueKnowledgeDocumentProcessing } from '@/lib/knowledge/documents/processing-outbox-event'
 import {
   assertDocumentProcessingBillingContext,
@@ -2372,9 +2373,8 @@ export async function markDocumentAsFailedTimeout(
 ): Promise<{ success: boolean; processingDuration: number }> {
   const now = new Date()
   const processingDuration = now.getTime() - processingStartedAt.getTime()
-  const DEAD_PROCESS_THRESHOLD_MS = 600 * 1000 // 10 minutes
 
-  if (processingDuration <= DEAD_PROCESS_THRESHOLD_MS) {
+  if (processingDuration <= KNOWLEDGE_DOCUMENT_PROCESSING_STALE_THRESHOLD_MS) {
     throw new Error('Document has not been processing long enough to be considered dead')
   }