fix(knowledge): restore processing takeover semantics

Author: TheodoreSpeaks

apps/sim/app/api/knowledge/utils.test.ts

@@ -175,7 +175,6 @@ describe('Knowledge Utils', () => {
       ])
       /** In-transaction active-document recheck. */
       queueTableRows(schemaMock.document, [{ id: 'doc1' }])
-      dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'doc1' }])
 
       await processDocumentAsync(
         'kb1',

apps/sim/lib/knowledge/documents/document-processing-source.test.ts

@@ -138,7 +138,6 @@ describe('knowledge document processing source', () => {
       .mockResolvedValueOnce([PERSISTED_CONTEXT])
       .mockResolvedValueOnce([PERSISTED_PROVENANCE_ROW])
       .mockResolvedValueOnce([{ id: 'document-1' }])
-    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'document-1' }])
     mockCheckActorUsageLimits.mockResolvedValue({ isExceeded: false })
     mockGetFileMetadataByKeys.mockImplementation(async (_keys: string[], context: string) =>
       context === 'workspace' ? [SOURCE_BINDING] : []
@@ -230,22 +229,28 @@ describe('knowledge document processing source', () => {
     expect(mockGenerateEmbeddings).not.toHaveBeenCalled()
   })
 
-  it('keeps the carrier retryable when another processing attempt owns the document', async () => {
+  it('takes over an existing processing attempt', async () => {
+    dbChainMockFns.limit
+      .mockReset()
+      .mockResolvedValueOnce([{ ...PERSISTED_CONTEXT, processingStatus: 'processing' }])
+      .mockResolvedValueOnce([PERSISTED_PROVENANCE_ROW])
+      .mockResolvedValueOnce([{ id: 'document-1' }])
     dbChainMockFns.returning.mockReset().mockResolvedValueOnce([])
 
-    await expect(
-      processDocumentAsync('knowledge-base-1', 'document-1', {
-        filename: 'stale.pdf',
-        fileUrl: 'https://example.com/stale.pdf',
-        fileSize: 1,
-        mimeType: 'text/plain',
-      })
-    ).rejects.toThrow('processing claim is owned by another attempt')
+    await processDocumentAsync('knowledge-base-1', 'document-1', {
+      filename: 'stale.pdf',
+      fileUrl: 'https://example.com/stale.pdf',
+      fileSize: 1,
+      mimeType: 'text/plain',
+    })
 
-    expect(mockProcessDocument).not.toHaveBeenCalled()
+    expect(mockProcessDocument).toHaveBeenCalled()
     expect(mockGenerateEmbeddings).not.toHaveBeenCalled()
-    expect(dbChainMockFns.set).not.toHaveBeenCalledWith(
-      expect.objectContaining({ processingStatus: 'failed' })
+    expect(dbChainMockFns.set).toHaveBeenCalledWith(
+      expect.objectContaining({
+        processingStatus: 'processing',
+        processingStartedAt: expect.any(Date),
+      })
     )
   })
 })

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

@@ -72,7 +72,6 @@ describe('processDocumentsWithQueue billing attribution', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     resetDbChainMock()
-    setEnvFlags({ isTriggerDevEnabled: true })
     mockBatchTrigger.mockResolvedValue({ batchId: 'batch-1' })
     for (const key of Object.keys(env)) {
       delete (env as Record<string, unknown>)[key]
@@ -153,29 +152,4 @@ describe('processDocumentsWithQueue billing attribution', () => {
     })
     expect(jobs[0].payload).not.toHaveProperty('billingAttribution')
   })
-
-  it('keeps direct dispatch retryable when another worker wins the document claim', async () => {
-    setEnvFlags({ isTriggerDevEnabled: false })
-    env.TRIGGER_SECRET_KEY = undefined
-    dbChainMockFns.limit
-      .mockResolvedValueOnce([{ userId: 'legacy-owner', workspaceId: null }])
-      .mockResolvedValueOnce([
-        {
-          workspaceId: null,
-          knowledgeBaseUserId: 'legacy-owner',
-          processingStatus: 'pending',
-          filename: DOCUMENT.filename,
-          fileUrl: DOCUMENT.fileUrl,
-          fileSize: DOCUMENT.fileSize,
-          mimeType: DOCUMENT.mimeType,
-        },
-      ])
-    dbChainMockFns.returning.mockResolvedValueOnce([])
-
-    await expect(
-      processDocumentsWithQueue([DOCUMENT], 'knowledge-base-1', {}, 'request-1', undefined)
-    ).rejects.toThrow('All 1 document processing dispatches failed')
-
-    expect(mockBatchTrigger).not.toHaveBeenCalled()
-  })
 })

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

@@ -118,13 +118,6 @@ import { calculateCost } from '@/providers/utils'
 
 const logger = createLogger('DocumentService')
 
-class DocumentProcessingClaimUnavailableError extends Error {
-  constructor(documentId: string) {
-    super(`Knowledge document ${documentId} processing claim is owned by another attempt`)
-    this.name = 'DocumentProcessingClaimUnavailableError'
-  }
-}
-
 /**
  * Thrown when a knowledge-base document's `fileUrl` references an internal
  * knowledge-base storage object not owned by the target knowledge base's workspace.
@@ -824,7 +817,6 @@ export async function processDocumentAsync(
         boolean1: document.boolean1,
         boolean2: document.boolean2,
         boolean3: document.boolean3,
-        processingStatus: document.processingStatus,
       })
       .from(document)
       .innerJoin(knowledgeBase, eq(knowledgeBase.id, document.knowledgeBaseId))
@@ -866,7 +858,7 @@ export async function processDocumentAsync(
       mimeType: ctx.mimeType,
     }
 
-    const [claimedDocument] = await db
+    await db
       .update(document)
       .set({
         processingStatus: 'processing',
@@ -875,21 +867,8 @@ export async function processDocumentAsync(
         processingError: null,
       })
       .where(
-        and(
-          eq(document.id, documentId),
-          inArray(document.processingStatus, ['pending', 'failed']),
-          isNull(document.archivedAt),
-          isNull(document.deletedAt)
-        )
+        and(eq(document.id, documentId), isNull(document.archivedAt), isNull(document.deletedAt))
       )
-      .returning({ id: document.id })
-
-    if (!claimedDocument) {
-      logger.info(`[${documentId}] Skipping document processing because another attempt owns it`, {
-        processingStatus: ctx.processingStatus,
-      })
-      throw new DocumentProcessingClaimUnavailableError(documentId)
-    }
 
     logger.info(`[${documentId}] Status updated to 'processing', starting document processor`)
 
@@ -1243,10 +1222,6 @@ export async function processDocumentAsync(
       fileSize: docData.fileSize,
     })
 
-    if (error instanceof DocumentProcessingClaimUnavailableError) {
-      throw error
-    }
-
     await db
       .update(document)
       .set({