fix(knowledge): make indexing and connector recovery durable (#7618)

* fix(knowledge): make indexing and connector recovery durable

* fix(helm): bump chart version for indexing job limits

* fix(knowledge): protect connector uploads during attachment

Author: icecrasher321

apps/sim/.env.example

@@ -231,3 +231,13 @@ CRON_SECRET=your_cron_secret # Use `openssl rand -hex 32` to generate. Authentic
 # Agent tool-call loop (Optional). Model round trips one Agent block takes before it
 # must answer. Defaults to 20; raise it for agents that chain many tool calls.
 # MAX_TOOL_ITERATIONS=20
+
+# Mistral OCR capacity (regular KBs, Sim Search and Mistral tools share these limits)
+# Use operating ceilings below the organization's actual quota, allowing for other clients.
+# KB_CONFIG_OCR_REQUESTS_PER_MINUTE=60
+# KB_CONFIG_MISTRAL_OCR_PAGES_PER_MINUTE=1000
+# KB_CONFIG_MISTRAL_OCR_PAGES_PER_REQUEST=30
+# KB_CONFIG_MISTRAL_OCR_MAX_CONCURRENT=2
+# Hosted MISTRAL_API_KEY requests share capacity across key rotation. Map any additional
+# keys in the same organization to one group using SHA-256 fingerprints, never raw keys.
+# MISTRAL_OCR_QUOTA_GROUPS={"<64-character lowercase key fingerprint>":"organization-id"}

apps/sim/app/api/v2/knowledge/connector-utils.ts

@@ -89,9 +89,18 @@ export function toV2KnowledgeConnectorDetail(
     ...toV2KnowledgeConnector(connector),
     syncLogs: connector.syncLogs.map((log) =>
       v2KnowledgeConnectorSyncLogSchema.parse({
-        ...log,
+        id: log.id,
+        connectorId: log.connectorId,
+        status: log.status,
         startedAt: serializeDate(log.startedAt),
         completedAt: serializeNullableDate(log.completedAt),
+        docsAdded: log.docsAdded,
+        docsUpdated: log.docsUpdated,
+        docsDeleted: log.docsDeleted,
+        docsUnchanged: log.docsUnchanged,
+        docsSkipped: log.docsSkipped,
+        docsFailed: log.docsFailed,
+        errorMessage: log.errorMessage,
       })
     ),
   })

apps/sim/app/api/webhooks/outbox/process/route.ts

@@ -24,7 +24,7 @@ import { reapStaleBackgroundWork } from '@/ee/workspace-forking/lib/background-w
 const logger = createLogger('OutboxProcessorAPI')
 
 export const dynamic = 'force-dynamic'
-export const maxDuration = 120
+export const maxDuration = 800
 
 const handlers = {
   ...adminInvitationOperationOutboxHandlers,
@@ -53,7 +53,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
 
     const result = await processOutboxEvents(handlers, {
       batchSize: 20,
-      maxRuntimeMs: 110_000,
+      maxRuntimeMs: 790_000,
       minRemainingMs: 95_000,
     })
 

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

@@ -26,13 +26,17 @@ vi.mock('@/lib/knowledge/documents/service', () => ({
   processDocumentAsync: mockProcessDocumentAsync,
 }))
 
+import { ProviderCapacityDeferredError } from '@/lib/core/rate-limiter/provider-capacity-error'
 import { EmbeddingAPIError, EmbeddingQuotaExhaustedError } from '@/lib/embeddings/client'
 import { EMBEDDING_QUOTA_CIRCUIT_TTL_MS } from '@/lib/embeddings/quota-circuit'
 import {
+  OcrRequestRejectedError,
   PermanentDocumentProcessingError,
   UsageLimitDocumentProcessingError,
 } from '@/lib/knowledge/documents/document-processing-error'
+import { MAX_PROVIDER_CONTINUATION_ATTEMPTS } from '@/lib/knowledge/documents/processing-provider-continuation'
 import { MAX_QUOTA_CONTINUATION_ATTEMPTS } from '@/lib/knowledge/documents/processing-quota-continuation'
+import type { DocumentProcessingAttemptContext } from '@/lib/knowledge/documents/service'
 import {
   resolveQuotaContinuationDelayMs,
   runDocumentProcessing,
@@ -91,7 +95,7 @@ const ORGANIZATION_PAYLOAD = {
 function mockQuotaExhaustion(error: EmbeddingQuotaExhaustedError): void {
   mockProcessDocumentAsync.mockImplementation(async (...args: unknown[]) => {
     const attemptContext = args[6] as {
-      scheduleQuotaContinuation?: () => Promise<Date>
+      scheduleQuotaContinuation?: () => Promise<unknown>
     }
     await attemptContext.scheduleQuotaContinuation?.()
     throw error
@@ -321,6 +325,39 @@ describe('knowledge processing worker', () => {
     )
   })
 
+  it('carries the actual parent admission flag when Trigger attempt two hands off quickly', async () => {
+    const error = new ProviderCapacityDeferredError('rate_limit')
+    mockProcessDocumentAsync.mockImplementation(async (...args: unknown[]) => {
+      const context = args[6] as {
+        scheduleProviderContinuation: (error: ProviderCapacityDeferredError) => Promise<unknown>
+      }
+      await context.scheduleProviderContinuation(error)
+      throw error
+    })
+    await runDocumentProcessing(
+      { ...WORKSPACE_PAYLOAD, processingQueueToken: 'request-1', chargedAtDispatch: true },
+      2
+    )
+    expect(mockTrigger.mock.calls[0][1]).toMatchObject({
+      processingPredecessorToken: 'request-1',
+      processingPredecessorCharged: false,
+    })
+  })
+
+  it('does not refund the original dispatch again when a healthy processing slice resumes', async () => {
+    await runDocumentProcessing({
+      ...WORKSPACE_PAYLOAD,
+      processingQueueToken: 'knowledge-slice-document-1-request-1-1',
+      processingSliceCount: 1,
+      providerRetryStartedAt: new Date().toISOString(),
+      chargedAtDispatch: true,
+    })
+    expect(mockProcessDocumentAsync.mock.calls[0][6]).toMatchObject({
+      chargedAtDispatch: false,
+      processingQueueToken: 'knowledge-slice-document-1-request-1-1',
+    })
+  })
+
   it('reports elapsed processing time rather than an epoch timestamp', async () => {
     vi.spyOn(Date, 'now').mockReturnValueOnce(1_000).mockReturnValueOnce(1_125)
 
@@ -357,6 +394,29 @@ describe('knowledge processing worker', () => {
     })
   })
 
+  it('completes provider-rejected OCR runs without requesting futile Trigger retries', async () => {
+    mockProcessDocumentAsync.mockRejectedValue(
+      new Error('OCR chunk batch failed', {
+        cause: new AggregateError([
+          new OcrRequestRejectedError(400),
+          new ProviderCapacityDeferredError('rate_limit'),
+        ]),
+      })
+    )
+    await expect(
+      runDocumentProcessing({
+        ...BASE_PAYLOAD,
+        billingScope: 'non-workspace',
+        actorUserId: 'legacy-owner',
+        workspaceId: null,
+      })
+    ).resolves.toMatchObject({
+      success: false,
+      outcome: 'provider_request_rejected',
+      code: 'ocr_request_rejected',
+    })
+  })
+
   it('reports a mutable usage-limit outcome without requesting an immediate retry', async () => {
     mockProcessDocumentAsync.mockRejectedValue(
       new UsageLimitDocumentProcessingError('Usage limit exceeded. Upgrade to continue.')
@@ -430,7 +490,8 @@ describe('knowledge processing worker', () => {
       expect.objectContaining({
         documentId: 'document-1',
         requestId: 'request-1',
-        processingQueuedAt: BASE_PAYLOAD.processingQueuedAt,
+        processingQueueToken: 'knowledge-quota-document-1-request-1-1',
+        processingQueuedAt: expect.any(String),
         quotaRetryCount: 1,
       }),
       expect.objectContaining({
@@ -444,6 +505,78 @@ describe('knowledge processing worker', () => {
     expect(delay.getTime()).toBeLessThanOrEqual(1_000 + EMBEDDING_QUOTA_CIRCUIT_TTL_MS * 1.2)
   })
 
+  it('continues provider pressure beyond the task retry budget without admitting another pass', async () => {
+    const error = new ProviderCapacityDeferredError('rate_limit', { retryAfterMs: 600_000 })
+    mockProcessDocumentAsync.mockImplementation(async (...args: unknown[]) => {
+      await (args[6] as DocumentProcessingAttemptContext).scheduleProviderContinuation!(error)
+      throw error
+    })
+    const now = Date.now()
+    await expect(
+      runDocumentProcessing(
+        {
+          ...WORKSPACE_PAYLOAD,
+          processingQueueToken: 'request-1',
+          providerRetryCount: 3,
+          providerRetryStartedAt: new Date(now).toISOString(),
+        },
+        3
+      )
+    ).resolves.toMatchObject({ outcome: 'provider_deferred' })
+    expect(mockProcessDocumentAsync).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.anything(),
+      expect.anything(),
+      expect.anything(),
+      expect.anything(),
+      'request-1',
+      expect.objectContaining({ chargedAtDispatch: false, processingQueueToken: 'request-1' })
+    )
+    expect(mockTrigger).toHaveBeenCalledWith(
+      'knowledge-process-document',
+      expect.objectContaining({
+        requestId: 'request-1',
+        processingQueueToken: 'knowledge-provider-document-1-request-1-4',
+        providerRetryCount: 4,
+        billingAttribution: BILLING_ATTRIBUTION,
+      }),
+      expect.objectContaining({ idempotencyKey: 'knowledge-provider-document-1-request-1-4' })
+    )
+    expect((mockTrigger.mock.calls[0][2].delay as Date).getTime()).toBeGreaterThanOrEqual(
+      now + 600_000
+    )
+  })
+
+  it('reports provider recovery exhaustion as an actionable terminal outcome', async () => {
+    mockProcessDocumentAsync.mockImplementation(async (...args: unknown[]) => {
+      await (args[6] as DocumentProcessingAttemptContext).scheduleProviderContinuation!(
+        new ProviderCapacityDeferredError('rate_limit')
+      )
+    })
+    await expect(
+      runDocumentProcessing({
+        ...WORKSPACE_PAYLOAD,
+        providerRetryCount: MAX_PROVIDER_CONTINUATION_ATTEMPTS,
+        providerRetryStartedAt: new Date().toISOString(),
+      })
+    ).resolves.toMatchObject({
+      outcome: 'provider_exhausted',
+      error: expect.stringContaining('then retry this document'),
+    })
+    expect(mockTrigger).not.toHaveBeenCalled()
+  })
+
+  it('retries failed provider continuation dispatch instead of reporting a successful deferral', async () => {
+    const error = new Error('Trigger dispatch unavailable')
+    mockTrigger.mockRejectedValue(error)
+    mockProcessDocumentAsync.mockImplementation(async (...args: unknown[]) => {
+      await (args[6] as DocumentProcessingAttemptContext).scheduleProviderContinuation!(
+        new ProviderCapacityDeferredError('rate_limit')
+      )
+    })
+    await expect(runDocumentProcessing(WORKSPACE_PAYLOAD)).rejects.toBe(error)
+  })
+
   it('ends a quota chain after the bounded continuation horizon', async () => {
     mockQuotaExhaustion(new EmbeddingQuotaExhaustedError('openai'))
 

apps/sim/background/knowledge-processing.ts

@@ -8,14 +8,21 @@ import {
   isEmbeddingQuotaExhaustion,
 } from '@/lib/embeddings'
 import {
+  getOcrRequestRejection,
   isPermanentDocumentProcessingError,
   isUsageLimitDocumentProcessingError,
 } from '@/lib/knowledge/documents/document-processing-error'
 import {
   assertDocumentProcessingBillingContext,
   assertDocumentProcessingPayload,
   type DocumentProcessingPayload,
+  shouldRefundDocumentProcessingPredecessor,
 } from '@/lib/knowledge/documents/processing-payload'
+import { scheduleDocumentProcessingProviderContinuation } from '@/lib/knowledge/documents/processing-provider-continuation'
+import {
+  getProviderCapacityDeferral,
+  ProviderCapacityContinuationExhaustedError,
+} from '@/lib/knowledge/documents/processing-provider-deferral'
 import {
   canScheduleDocumentProcessingQuotaContinuation,
   MAX_QUOTA_CONTINUATION_ATTEMPTS,
@@ -36,6 +43,12 @@ export async function runDocumentProcessing(
   const { knowledgeBaseId, documentId, docData, processingOptions, requestId } = payload
   const billingContext = assertDocumentProcessingBillingContext(payload)
   const canScheduleQuotaContinuation = canScheduleDocumentProcessingQuotaContinuation(payload)
+  const chargedAtDispatch =
+    (payload.chargedAtDispatch ?? payload.processingQueuedAt !== undefined) &&
+    attemptNumber === 1 &&
+    payload.quotaRetryCount === undefined &&
+    payload.providerRetryCount === undefined &&
+    payload.processingSliceCount === undefined
 
   logger.info(`[${requestId}] Starting Trigger.dev processing for document: ${docData.filename}`)
 
@@ -48,10 +61,13 @@ export async function runDocumentProcessing(
       billingContext,
       requestId,
       {
-        chargedAtDispatch:
-          (payload.chargedAtDispatch ?? payload.processingQueuedAt !== undefined) &&
-          attemptNumber === 1 &&
-          payload.quotaRetryCount === undefined,
+        chargedAtDispatch,
+        ...(payload.processingPredecessorToken
+          ? {
+              processingPredecessorToken: payload.processingPredecessorToken,
+              refundPredecessorAdmission: shouldRefundDocumentProcessingPredecessor(payload),
+            }
+          : {}),
         ...(payload.processingQueueToken
           ? { processingQueueToken: payload.processingQueueToken }
           : {}),
@@ -60,9 +76,12 @@ export async function runDocumentProcessing(
           : {}),
         ...(canScheduleQuotaContinuation
           ? {
-              scheduleQuotaContinuation: () => scheduleDocumentProcessingQuotaContinuation(payload),
+              scheduleQuotaContinuation: () =>
+                scheduleDocumentProcessingQuotaContinuation(payload, true, chargedAtDispatch),
             }
           : { quotaContinuationExhausted: true }),
+        scheduleProviderContinuation: (error) =>
+          scheduleDocumentProcessingProviderContinuation(payload, error, true, chargedAtDispatch),
       }
     )
 
@@ -75,6 +94,30 @@ export async function runDocumentProcessing(
       processingTime: Date.now() - startedAt,
     }
   } catch (error) {
+    const providerDeferral = getProviderCapacityDeferral(error)
+    if (providerDeferral || error instanceof ProviderCapacityContinuationExhaustedError) {
+      const outcome =
+        error instanceof ProviderCapacityContinuationExhaustedError
+          ? 'provider_exhausted'
+          : 'provider_deferred'
+      logger.warn(`[${requestId}] Document processing is waiting for provider recovery`, {
+        documentId,
+        providerRetryCount: payload.providerRetryCount ?? 0,
+        reason: providerDeferral?.reason,
+        outcome,
+      })
+      return {
+        success: false,
+        outcome,
+        documentId,
+        filename: docData.filename,
+        error:
+          error instanceof ProviderCapacityContinuationExhaustedError
+            ? error.message
+            : providerDeferral!.message,
+        processingTime: Date.now() - startedAt,
+      }
+    }
     if (isUsageLimitDocumentProcessingError(error)) {
       logger.warn(`[${requestId}] Document processing is blocked by the current usage limit`, {
         filename: docData.filename,
@@ -120,6 +163,22 @@ export async function runDocumentProcessing(
         processingTime: Date.now() - startedAt,
       }
     }
+    const ocrRejection = getOcrRequestRejection(error)
+    if (ocrRejection) {
+      logger.warn(`[${requestId}] OCR request requires remediation before retrying`, {
+        status: ocrRejection.status,
+        code: ocrRejection.code,
+      })
+      return {
+        success: false,
+        outcome: 'provider_request_rejected' as const,
+        documentId,
+        filename: docData.filename,
+        code: ocrRejection.code,
+        error: ocrRejection.message,
+        processingTime: Date.now() - startedAt,
+      }
+    }
     if (isPermanentDocumentProcessingError(error)) {
       logger.warn(`[${requestId}] Document cannot be processed without changing its content`, {
         code: error.code,