fix(knowledge): bound recovery database checks and sync staging

Author: waleedlatif1

apps/sim/background/knowledge-processing.test.ts

@@ -119,7 +119,7 @@ describe('knowledge processing worker', () => {
       }
       return value
     })
-    mockProcessDocumentAsync.mockResolvedValue(undefined)
+    mockProcessDocumentAsync.mockResolvedValue({ outcome: 'indexed' })
     mockResolveTriggerRegion.mockResolvedValue('us-east-1')
     mockTrigger.mockResolvedValue({ id: 'quota-continuation-run' })
   })
@@ -128,6 +128,27 @@ describe('knowledge processing worker', () => {
     vi.restoreAllMocks()
   })
 
+  it('reports indexed only when the document service committed the index', async () => {
+    expect(await runDocumentProcessing(WORKSPACE_PAYLOAD)).toMatchObject({
+      success: true,
+      outcome: 'indexed',
+      documentId: WORKSPACE_PAYLOAD.documentId,
+    })
+  })
+
+  it.each(['unavailable', 'not_claimed', 'superseded'] as const)(
+    'reports a harmless %s skip without turning it into a task failure or an indexed success',
+    async (reason) => {
+      mockProcessDocumentAsync.mockResolvedValue({ outcome: 'skipped', reason })
+      expect(await runDocumentProcessing(WORKSPACE_PAYLOAD)).toMatchObject({
+        success: false,
+        outcome: 'skipped',
+        reason,
+      })
+      expect(mockTrigger).not.toHaveBeenCalled()
+    }
+  )
+
   it('rejects workspace work without attribution before document processing starts', async () => {
     await expect(
       runDocumentProcessing({

apps/sim/background/knowledge-processing.ts

@@ -58,7 +58,7 @@ export async function runDocumentProcessing(
   logger.info(`[${requestId}] Starting Trigger.dev processing for document: ${docData.filename}`)
 
   try {
-    await processDocumentAsync(
+    const result = await processDocumentAsync(
       knowledgeBaseId,
       documentId,
       docData,
@@ -90,10 +90,11 @@ export async function runDocumentProcessing(
       }
     )
 
-    logger.info(`[${requestId}] Successfully processed document: ${docData.filename}`)
+    logger.info(`[${requestId}] Document processing finished`, { documentId, ...result })
 
     return {
-      success: true,
+      success: result.outcome === 'indexed',
+      ...result,
       documentId,
       filename: docData.filename,
       processingTime: Date.now() - startedAt,

apps/sim/lib/knowledge/__integration__/stored-document-recovery.integration.ts

@@ -43,7 +43,6 @@ vi.mock('@trigger.dev/sdk', async (importOriginal) => {
   const original = await importOriginal<typeof import('@trigger.dev/sdk')>()
   return {
     ...original,
-    runs: { list: fixture.listRuns },
     tasks: { ...original.tasks, batchTrigger: fixture.batchTrigger },
   }
 })
@@ -96,6 +95,7 @@ import {
   retryDocumentProcessing,
 } from '@/lib/knowledge/documents/service'
 import { MAX_PROCESSING_ATTEMPTS, QUEUED_DISPATCH_GRACE_MS } from '@/lib/knowledge/documents/types'
+import type { SyncResult } from '@/connectors/types'
 
 const fixtures: ReturnType<typeof createKnowledgeAclFixtureIds>[] = []
 const old = () => new Date(Date.now() - QUEUED_DISPATCH_GRACE_MS - 60_000)
@@ -179,7 +179,110 @@ afterAll(async () => {
   await db.$client.end()
 })
 
+async function recoverFixture(
+  ids: ReturnType<typeof createKnowledgeAclFixtureIds>,
+  mode: 'independent' | 'connector'
+) {
+  if (mode === 'independent') return recoverKnowledgeDocumentProcessing()
+  const result: SyncResult = {
+    docsAdded: 0,
+    docsUpdated: 0,
+    docsDeleted: 0,
+    docsUnchanged: 0,
+    docsSkipped: 0,
+    docsFailed: 0,
+    processingDispatch: { requested: 0, accepted: 0, failed: 0 },
+  }
+  await sweepStuckDocuments({
+    connectorId: ids.connectorId,
+    knowledgeBaseId: ids.knowledgeBaseId,
+    syncStartedAt: new Date(),
+    retryCutoff: new Date(Date.now() - 7 * 24 * 60 * 60_000),
+    billingAttribution: await resolveSystemBillingAttribution(ids.workspaceId),
+    result,
+    lease: createContentSyncLease(ids.connectorId, ids.lockId),
+  })
+  return result.processingDispatch.requested
+}
+
 describe('independent recovery of retained connector documents', () => {
+  it.each(['independent', 'connector'] as const)(
+    '%s recovery preserves a job queued beyond the grace period',
+    async (mode) => {
+      const ids = await seed()
+      const file = await failedFile(ids)
+      await db
+        .update(document)
+        .set({ processingStatus: 'pending' })
+        .where(eq(document.id, file.documentId))
+      fixture.useTrigger = true
+      fixture.listRuns.mockResolvedValue({
+        data: [{ id: 'run-queued', status: 'QUEUED' }],
+        hasNextPage: () => false,
+      })
+      expect(await recoverFixture(ids, mode)).toBe(0)
+      const [row] = await db.select().from(document).where(eq(document.id, file.documentId))
+      expect(row.processingAttempts).toBe(1)
+      expect(row.processingQueueToken).toBe('old-fixture-generation')
+      expect(row.processingRecoveryAfter).not.toBeNull()
+      expect(await eventsFor(ids)).toHaveLength(0)
+      await db
+        .update(knowledgeConnector)
+        .set({ status: 'paused' })
+        .where(eq(knowledgeConnector.id, ids.connectorId))
+    }
+  )
+
+  it.each(['independent', 'connector'] as const)(
+    '%s recovery rechecks the generation after its remote lookup',
+    async (mode) => {
+      const ids = await seed()
+      const file = await failedFile(ids)
+      fixture.useTrigger = true
+      fixture.listRuns.mockImplementation(async () => {
+        await db
+          .update(document)
+          .set({ processingQueueToken: 'replacement-generation' })
+          .where(eq(document.id, file.documentId))
+        return { data: [], hasNextPage: () => false }
+      })
+      expect(await recoverFixture(ids, mode)).toBe(0)
+      const [row] = await db.select().from(document).where(eq(document.id, file.documentId))
+      expect(row.processingAttempts).toBe(1)
+      expect(row.processingQueueToken).toBe('replacement-generation')
+      expect(await eventsFor(ids)).toHaveLength(0)
+      await db
+        .update(knowledgeConnector)
+        .set({ status: 'paused' })
+        .where(eq(knowledgeConnector.id, ids.connectorId))
+    }
+  )
+
+  it.each(['pending', 'processing'])(
+    'does not replace an aged %s outbox continuation',
+    async (status) => {
+      const ids = await seed()
+      const file = await failedFile(ids)
+      const token = generateId()
+      await db
+        .update(document)
+        .set({ processingQueueToken: token })
+        .where(eq(document.id, file.documentId))
+      await db.insert(outboxEvent).values({
+        id: token,
+        eventType: 'knowledge.document.processing.resume',
+        payload: { knowledgeBaseId: ids.knowledgeBaseId, documentId: file.documentId },
+        status,
+        availableAt: old(),
+      })
+      expect(await recoverFixture(ids, 'independent')).toBe(0)
+      expect(await recoverFixture(ids, 'connector')).toBe(0)
+      const [row] = await db.select().from(document).where(eq(document.id, file.documentId))
+      expect(row.processingAttempts).toBe(1)
+      expect(row.processingQueueToken).toBe(token)
+    }
+  )
+
   it.each(['manual', 'redelivery'])(
     'respects a concurrent liveness cooldown before %s replacement',
     async (path) => {

apps/sim/lib/knowledge/connectors/sync-primitives.ts

@@ -27,13 +27,13 @@ import {
   persistSkippedRetryHashes,
   updateDocument,
 } from '@/lib/knowledge/connectors/sync-persistence'
+import { documentProcessingRecoveryCondition } from '@/lib/knowledge/documents/processing-recovery-policy'
 import {
   DOCUMENT_LIVENESS_BATCH_SIZE,
   documentProcessingSnapshotCondition,
   findAbandonedDocumentProcessing,
   processingSnapshotColumns,
-} from '@/lib/knowledge/documents/processing-liveness'
-import { documentProcessingRecoveryCondition } from '@/lib/knowledge/documents/processing-recovery-policy'
+} from '@/lib/knowledge/documents/processing-recovery-queue'
 import { DOCUMENT_PROCESSING_STALE_THRESHOLD_MS } from '@/lib/knowledge/documents/processing-timeouts.server'
 import type { DocumentData } from '@/lib/knowledge/documents/service'
 import { isTriggerAvailable, processDocumentsWithQueue } from '@/lib/knowledge/documents/service'
@@ -1341,7 +1341,6 @@ export async function sweepStuckDocuments(input: SweepStuckDocumentsInput): Prom
       filename: document.filename,
       fileSize: document.fileSize,
       mimeType: document.mimeType,
-      uploadedAt: document.uploadedAt,
     })
     .from(document)
     .where(
@@ -1403,7 +1402,6 @@ export async function sweepStuckDocuments(input: SweepStuckDocumentsInput): Prom
           filename: document.filename,
           fileSize: document.fileSize,
           mimeType: document.mimeType,
-          uploadedAt: document.uploadedAt,
         })
         .from(document)
         .where(

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

@@ -1460,11 +1460,46 @@ describe('processDocumentAsync write guards', () => {
     expect(schedule).not.toHaveBeenCalled()
   })
 
+  it('reports an unavailable document without parsing or indexing it', async () => {
+    dbChainMockFns.limit.mockResolvedValueOnce([])
+    expect(
+      await processDocumentAsync(
+        'knowledge-base-1',
+        'document-1',
+        PERSISTED_CONTEXT,
+        {},
+        BILLING_ATTRIBUTION
+      )
+    ).toEqual({ outcome: 'skipped', reason: 'unavailable' })
+    expect(mockProcessDocument).not.toHaveBeenCalled()
+  })
+
+  it('reports discarded output when the generation changes before the index commit', async () => {
+    armProviderSource()
+    dbChainMockFns.limit.mockReset()
+    dbChainMockFns.limit
+      .mockResolvedValueOnce([PERSISTED_CONTEXT])
+      .mockResolvedValueOnce([PERSISTED_PROVENANCE_ROW])
+      .mockResolvedValueOnce([])
+    expect(
+      await processDocumentAsync(
+        'knowledge-base-1',
+        'document-1',
+        PERSISTED_CONTEXT,
+        {},
+        BILLING_ATTRIBUTION
+      )
+    ).toEqual({ outcome: 'skipped', reason: 'superseded' })
+    expect(
+      dbChainMockFns.set.mock.calls.some(([value]) => value.processingStatus === 'completed')
+    ).toBe(false)
+  })
+
   it('does not parse or reschedule a superseded provider continuation', async () => {
     armProviderSource()
     dbChainMockFns.returning.mockResolvedValueOnce([])
     const schedule = vi.fn()
-    await processDocumentAsync(
+    const result = await processDocumentAsync(
       'knowledge-base-1',
       'document-1',
       PERSISTED_CONTEXT,
@@ -1477,6 +1512,7 @@ describe('processDocumentAsync write guards', () => {
         scheduleProviderContinuation: schedule,
       }
     )
+    expect(result).toEqual({ outcome: 'skipped', reason: 'not_claimed' })
     expect(mockProcessDocument).not.toHaveBeenCalled()
     expect(schedule).not.toHaveBeenCalled()
   })

apps/sim/lib/knowledge/documents/processing-recovery-policy.ts

@@ -1,4 +1,4 @@
-import { document } from '@sim/db/schema'
+import { document, outboxEvent } from '@sim/db/schema'
 import { and, eq, gt, isNotNull, isNull, lt, lte, or, sql } from 'drizzle-orm'
 import { DOCUMENT_PROCESSING_STALE_THRESHOLD_MS } from '@/lib/knowledge/documents/processing-timeouts.server'
 import { MAX_PROCESSING_ATTEMPTS, QUEUED_DISPATCH_GRACE_MS } from '@/lib/knowledge/documents/types'
@@ -15,6 +15,11 @@ export function documentProcessingRecoveryCondition(
   return and(
     sql`${document.processingStatus} IN ('pending', 'processing', 'failed')`,
     isNotNull(document.connectorId),
+    sql`NOT EXISTS (
+      SELECT 1 FROM ${outboxEvent}
+      WHERE ${outboxEvent.id} = ${document.processingQueueToken}
+        AND ${outboxEvent.status} IN ('pending', 'processing')
+    )`,
     isNotNull(document.contentHash),
     isNotNull(document.storageKey),
     eq(document.userExcluded, false),

…ts/processing-liveness-transport.test.ts …cessing-recovery-queue-transport.test.tsapps/sim/lib/knowledge/documents/processing-liveness-transport.test.ts renamed to apps/sim/lib/knowledge/documents/processing-recovery-queue-transport.test.ts apps/sim/lib/knowledge/documents/processing-liveness-transport.test.ts renamed to apps/sim/lib/knowledge/documents/processing-recovery-queue-transport.test.ts

@@ -11,10 +11,11 @@ import { resetInsideTriggerRunForTests } from '@/lib/core/config/trigger-runtime
 import {
   type DocumentProcessingSnapshot,
   findAbandonedDocumentProcessing,
-} from '@/lib/knowledge/documents/processing-liveness'
+} from '@/lib/knowledge/documents/processing-recovery-queue'
 
 const snapshot: DocumentProcessingSnapshot = {
   id: 'doc-1',
+  uploadedAt: new Date('2026-09-01T00:00:00Z'),
   processingStatus: 'pending',
   processingQueueToken: 'generation-1',
   processingQueuedAt: new Date('2026-09-01T00:00:00Z'),
@@ -92,6 +93,9 @@ describe('document liveness SDK transport', () => {
       expect(fetch).toHaveBeenCalledOnce()
       const url = new URL(fetch.mock.calls[0][0])
       expect(url.searchParams.get('page[size]')).toBe('1')
+      expect(url.searchParams.get('filter[createdAt][from]')).toBe(
+        String(new Date('2026-08-31T20:00:00Z').getTime())
+      )
       expect(url.searchParams.get('filter[tag]')).toBe('documentId:doc-1')
     }
   )