Merge remote-tracking branch 'origin/staging' into fix/connector-lease-transactions-bounded

# Conflicts:
#	apps/sim/lib/knowledge/connectors/member-sync-engine.ts

Author: waleedlatif1

.github/workflows/test-build.yml

@@ -102,6 +102,7 @@ jobs:
           bunx vitest run
           scripts/retired-columns.postgres.test.ts
           scripts/connector-sync-schedule-precision.postgres.test.ts
+          scripts/database-failure-classification.postgres.test.ts
 
       - name: Verify OAuth lifecycle and SCIM membership guards in PostgreSQL
         working-directory: apps/sim

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

@@ -46,6 +46,10 @@ import { MAX_PROVIDER_CONTINUATION_ATTEMPTS } from '@/lib/knowledge/documents/pr
 import { MAX_QUOTA_CONTINUATION_ATTEMPTS } from '@/lib/knowledge/documents/processing-quota-continuation'
 import type { DocumentProcessingAttemptContext } from '@/lib/knowledge/documents/service'
 import {
+  DOCUMENT_PROCESSING_RETRY_POLICY,
+  DocumentProcessingDatabaseRetryError,
+  getDocumentProcessingRetry,
+  processDocument,
   resolveQuotaContinuationDelayMs,
   runDocumentProcessing,
 } from '@/background/knowledge-processing'
@@ -607,10 +611,83 @@ describe('knowledge processing worker', () => {
     )
     expect(failure).toBeInstanceOf(Error)
     expect(failure).toMatchObject({ message: 'Database request failed (SQLSTATE 57014).' })
-    expect(failure).not.toHaveProperty('cause')
+    /** Trigger records only the name, message and stack; the cause stays for classification. */
+    expect((failure as Error).cause).toBe(error)
+    expect((failure as Error).stack).not.toContain('private')
     expect(JSON.stringify(failure)).not.toContain('private')
   })
 
+  describe('transient database failures', () => {
+    const MINUTE = 60 * 1000
+    const statementTimeout = () =>
+      new DrizzleQueryError(
+        'insert private SQL',
+        ['private bound content'],
+        Object.assign(new Error('canceling statement due to statement timeout'), {
+          code: '57014',
+        })
+      )
+
+    /** A service failure that asks the worker whether to schedule a database retry, as the service does. */
+    function failProcessingWith(error: Error): { scheduled: Array<Date | null> } {
+      const scheduled: Array<Date | null> = []
+      mockProcessDocumentAsync.mockImplementation(async (...args: unknown[]) => {
+        const context = args[6] as DocumentProcessingAttemptContext
+        scheduled.push(context.scheduleDatabaseRetry?.(error) ?? null)
+        throw error
+      })
+      return { scheduled }
+    }
+
+    it('schedules a minute-scale retry and hands Trigger the same time the document records', async () => {
+      const error = statementTimeout()
+      const { scheduled } = failProcessingWith(error)
+      const startedAt = Date.now()
+
+      const failure = await runDocumentProcessing(WORKSPACE_PAYLOAD, 1).catch(
+        (caught: unknown) => caught
+      )
+
+      expect(failure).toBeInstanceOf(DocumentProcessingDatabaseRetryError)
+      expect(failure).toMatchObject({ message: 'Database request failed (SQLSTATE 57014).' })
+      expect((failure as Error).cause).toBe(error)
+      expect((failure as Error).stack).not.toContain('private')
+      const retryAt = scheduled[0]
+      expect(retryAt).toBeInstanceOf(Date)
+      expect(retryAt!.getTime() - startedAt).toBeGreaterThanOrEqual(2 * MINUTE * 0.8)
+      expect(retryAt!.getTime() - startedAt).toBeLessThanOrEqual(2 * MINUTE * 1.2 + 1000)
+      expect(getDocumentProcessingRetry(failure, 1)).toEqual({ retryAt })
+    })
+
+    it('records the failure and stops once the database attempts are spent', async () => {
+      const error = statementTimeout()
+      const { scheduled } = failProcessingWith(error)
+      const lastAttempt = DOCUMENT_PROCESSING_RETRY_POLICY.database.maxAttempts
+
+      const failure = await runDocumentProcessing(WORKSPACE_PAYLOAD, lastAttempt).catch(
+        (caught: unknown) => caught
+      )
+
+      expect(scheduled).toEqual([null])
+      expect(failure).not.toBeInstanceOf(DocumentProcessingDatabaseRetryError)
+      expect((failure as Error).cause).toBe(error)
+      expect(getDocumentProcessingRetry(failure, lastAttempt)).toEqual({ skipRetrying: true })
+    })
+
+    it('leaves other failures on the task retry settings and attempt count', async () => {
+      const error = new Error('Storage request timed out')
+      const { scheduled } = failProcessingWith(error)
+
+      await expect(runDocumentProcessing(WORKSPACE_PAYLOAD, 1)).rejects.toBe(error)
+
+      expect(scheduled).toEqual([null])
+      expect(getDocumentProcessingRetry(error, 1)).toBeUndefined()
+      expect(
+        getDocumentProcessingRetry(error, DOCUMENT_PROCESSING_RETRY_POLICY.maxAttempts)
+      ).toEqual({ skipRetrying: true })
+    })
+  })
+
   it('retries failed provider continuation dispatch instead of reporting a successful deferral', async () => {
     const error = new Error('Trigger dispatch unavailable')
     mockTrigger.mockRejectedValue(error)
@@ -745,11 +822,29 @@ describe('knowledge-process-document task configuration', () => {
    * `attempt_count = 1`, so each was left `failed` having never been retried.
    */
   it('escalates to a larger machine on an out-of-memory kill', async () => {
-    const { processDocument } = await import('@/background/knowledge-processing')
-
     expect(processDocument.retry?.outOfMemory?.machine).toBe('large-2x')
   })
 
+  it('declares enough attempts for database retries and routes failures through catchError', async () => {
+    expect(processDocument.retry?.maxAttempts).toBe(
+      Math.max(
+        DOCUMENT_PROCESSING_RETRY_POLICY.maxAttempts,
+        DOCUMENT_PROCESSING_RETRY_POLICY.database.maxAttempts
+      )
+    )
+    const retryAt = new Date('2026-01-01T00:02:00.000Z')
+    const scheduled = new DocumentProcessingDatabaseRetryError(
+      'Database request failed.',
+      retryAt,
+      {
+        cause: new Error('private'),
+      }
+    )
+    await expect(
+      processDocument.catchError?.({ error: scheduled, ctx: { attempt: { number: 1 } } } as never)
+    ).resolves.toEqual({ retryAt })
+  })
+
   it('backs durable quota continuations off to a bounded polling interval', () => {
     const first = resolveQuotaContinuationDelayMs(1)
     const second = resolveQuotaContinuationDelayMs(2)

apps/sim/background/knowledge-processing.ts

@@ -1,6 +1,14 @@
 import { createLogger } from '@sim/logger'
+import { findCause } from '@sim/utils/errors'
 import { queue, task } from '@trigger.dev/sdk'
 import { env, envNumber } from '@/lib/core/config/env'
+import {
+  type BackgroundRetryDecision,
+  type BackgroundRetryPolicy,
+  backgroundRetryAttemptCeiling,
+  getBackgroundRetryDecision,
+  getDatabaseRetryAt,
+} from '@/lib/core/errors/background-retry'
 import {
   BYOK_EMBEDDING_CREDENTIAL_REJECTION_MESSAGE,
   EMBEDDING_QUOTA_EXHAUSTED_MESSAGE,
@@ -39,6 +47,46 @@ import { processDocumentAsync } from '@/lib/knowledge/documents/service'
 const logger = createLogger('TriggerKnowledgeProcessing')
 export { resolveQuotaContinuationDelayMs }
 
+/**
+ * Ordinary failures keep the configured short retries. A transient database failure backs off for
+ * minutes, about an hour in total, so a slow database window does not exhaust every attempt inside
+ * it and leave an uploaded document failed for good.
+ */
+export const DOCUMENT_PROCESSING_RETRY_POLICY: BackgroundRetryPolicy = {
+  maxAttempts: envNumber(env.KB_CONFIG_MAX_ATTEMPTS, 3),
+  database: { maxAttempts: 6, baseDelayMs: 2 * 60 * 1000, maxDelayMs: 30 * 60 * 1000 },
+}
+
+/**
+ * A database failure whose next attempt is already scheduled, and recorded on the document as
+ * `pending` until {@link retryAt}. The message names only the database code; the driver error
+ * stays in `cause`, which the task runner does not record.
+ */
+export class DocumentProcessingDatabaseRetryError extends Error {
+  constructor(
+    message: string,
+    readonly retryAt: Date,
+    options: { cause: unknown }
+  ) {
+    super(message, options)
+    this.name = 'DocumentProcessingDatabaseRetryError'
+  }
+}
+
+/** The `catchError` decision for `knowledge-process-document` after `attempt` (1-based) failed. */
+export function getDocumentProcessingRetry(
+  error: unknown,
+  attempt: number
+): BackgroundRetryDecision {
+  const scheduled = findCause(
+    error,
+    (value): value is DocumentProcessingDatabaseRetryError =>
+      value instanceof DocumentProcessingDatabaseRetryError
+  )
+  if (scheduled) return { retryAt: scheduled.retryAt }
+  return getBackgroundRetryDecision(error, attempt, DOCUMENT_PROCESSING_RETRY_POLICY)
+}
+
 export async function runDocumentProcessing(
   rawPayload: DocumentProcessingPayload,
   attemptNumber = 1
@@ -56,6 +104,8 @@ export async function runDocumentProcessing(
     payload.processingSliceCount === undefined
 
   logger.info(`[${requestId}] Starting Trigger.dev processing for document: ${docData.filename}`)
+  /** Set from the service's callback, so control-flow narrowing cannot see it change. */
+  let databaseRetryAt = null as Date | null
 
   try {
     const result = await processDocumentAsync(
@@ -87,6 +137,14 @@ export async function runDocumentProcessing(
           : { quotaContinuationExhausted: true }),
         scheduleProviderContinuation: (error) =>
           scheduleDocumentProcessingProviderContinuation(payload, error, true, chargedAtDispatch),
+        scheduleDatabaseRetry: (error) => {
+          databaseRetryAt = getDatabaseRetryAt(
+            error,
+            attemptNumber,
+            DOCUMENT_PROCESSING_RETRY_POLICY
+          )
+          return databaseRetryAt
+        },
       }
     )
 
@@ -100,6 +158,20 @@ export async function runDocumentProcessing(
       processingTime: Date.now() - startedAt,
     }
   } catch (error) {
+    if (databaseRetryAt) {
+      const diagnostic = getConnectorFailureDiagnostic(error)
+      logger.warn(`[${requestId}] Document processing will retry after a database failure`, {
+        documentId,
+        diagnostic,
+        attempt: attemptNumber,
+        retryAt: databaseRetryAt.toISOString(),
+      })
+      throw new DocumentProcessingDatabaseRetryError(
+        diagnostic?.message ?? 'Database request failed.',
+        databaseRetryAt,
+        { cause: error }
+      )
+    }
     const providerDeferral = getProviderCapacityDeferral(error)
     if (providerDeferral || error instanceof ProviderCapacityContinuationExhaustedError) {
       const outcome =
@@ -205,7 +277,8 @@ export async function runDocumentProcessing(
       `[${requestId}] Failed to process document: ${docData.filename}`,
       diagnostic ?? error
     )
-    if (diagnostic?.category === 'database') throw new Error(diagnostic.message)
+    /** Trigger records the thrown message and stack, never `cause`; Drizzle's message carries SQL. */
+    if (diagnostic?.category === 'database') throw new Error(diagnostic.message, { cause: error })
     throw error
   }
 }
@@ -253,7 +326,13 @@ export const processDocument = task({
    */
   machine: 'medium-2x',
   retry: {
-    maxAttempts: envNumber(env.KB_CONFIG_MAX_ATTEMPTS, 3),
+    /**
+     * The ceiling for thrown errors: database retries use all of it, and
+     * `catchError` stops every other thrown error at `KB_CONFIG_MAX_ATTEMPTS`.
+     * A crashed or timed-out run is not retried; an out-of-memory kill is
+     * retried once, on the `outOfMemory` machine below.
+     */
+    maxAttempts: backgroundRetryAttemptCeiling(DOCUMENT_PROCESSING_RETRY_POLICY),
     factor: envNumber(env.KB_CONFIG_RETRY_FACTOR, 2),
     minTimeoutInMs: envNumber(env.KB_CONFIG_MIN_TIMEOUT, 1000),
     maxTimeoutInMs: envNumber(env.KB_CONFIG_MAX_TIMEOUT, 10000),
@@ -272,4 +351,5 @@ export const processDocument = task({
   queue: interactiveProcessingQueue,
   run: (payload: DocumentProcessingPayload, { ctx }) =>
     runDocumentProcessing(payload, ctx.attempt.number),
+  catchError: async ({ error, ctx }) => getDocumentProcessingRetry(error, ctx.attempt.number),
 })

apps/sim/lib/core/errors/background-retry.test.ts

@@ -0,0 +1,95 @@
+/**
+ * @vitest-environment node
+ */
+import { DrizzleQueryError } from 'drizzle-orm/errors'
+import { describe, expect, it } from 'vitest'
+import {
+  type BackgroundRetryDecision,
+  type BackgroundRetryPolicy,
+  backgroundRetryAttemptCeiling,
+  getBackgroundRetryDecision,
+  getDatabaseRetryAt,
+} from '@/lib/core/errors/background-retry'
+
+const MINUTE = 60 * 1000
+const POLICY: BackgroundRetryPolicy = {
+  maxAttempts: 3,
+  database: { maxAttempts: 6, baseDelayMs: 2 * MINUTE, maxDelayMs: 30 * MINUTE },
+}
+const NOW = Date.parse('2026-01-01T00:00:00.000Z')
+
+function failedQuery(code: string, message = 'private driver detail'): DrizzleQueryError {
+  return new DrizzleQueryError(
+    'private SQL',
+    ['private'],
+    Object.assign(new Error(message), { code })
+  )
+}
+
+function delayOf(decision: BackgroundRetryDecision): number {
+  if (!decision || !('retryAt' in decision)) throw new Error('expected a scheduled retry')
+  return decision.retryAt.getTime() - NOW
+}
+
+describe('getBackgroundRetryDecision', () => {
+  it.each([
+    ['capacity', failedQuery('57014', 'canceling statement due to statement timeout')],
+    ['conflict', failedQuery('40P01', 'deadlock detected')],
+    ['connection', failedQuery('CONNECTION_CLOSED')],
+  ])('waits minutes after a %s failure', (_label, error) => {
+    const delay = delayOf(getBackgroundRetryDecision(error, 1, POLICY, NOW))
+    expect(delay).toBeGreaterThanOrEqual(2 * MINUTE * 0.8)
+    expect(delay).toBeLessThanOrEqual(2 * MINUTE * 1.2)
+  })
+
+  it('doubles the delay per attempt up to the ceiling', () => {
+    const error = failedQuery('55P03')
+    const delays = [1, 2, 3, 4, 5].map((attempt) =>
+      delayOf(getBackgroundRetryDecision(error, attempt, POLICY, NOW))
+    )
+    const bases = [2, 4, 8, 16, 30].map((minutes) => minutes * MINUTE)
+    delays.forEach((delay, index) => {
+      expect(delay).toBeGreaterThanOrEqual(bases[index] * 0.8)
+      expect(delay).toBeLessThanOrEqual(bases[index] * 1.2)
+    })
+    const longPolicy = { ...POLICY, database: { ...POLICY.database, maxAttempts: 20 } }
+    expect(delayOf(getBackgroundRetryDecision(error, 12, longPolicy, NOW))).toBeLessThanOrEqual(
+      30 * MINUTE * 1.2
+    )
+  })
+
+  it('stops database retries at their own attempt ceiling', () => {
+    const error = failedQuery('53300')
+    expect(getBackgroundRetryDecision(error, 5, POLICY, NOW)).toHaveProperty('retryAt')
+    expect(getBackgroundRetryDecision(error, 6, POLICY, NOW)).toEqual({ skipRetrying: true })
+  })
+
+  it('keeps the task default for other failures until their attempt ceiling', () => {
+    const error = failedQuery('23505')
+    expect(getBackgroundRetryDecision(error, 1, POLICY, NOW)).toBeUndefined()
+    expect(getBackgroundRetryDecision(error, 2, POLICY, NOW)).toBeUndefined()
+    expect(getBackgroundRetryDecision(error, 3, POLICY, NOW)).toEqual({ skipRetrying: true })
+  })
+
+  it('does not stretch an explicit cancellation onto the database pacing', () => {
+    const cancelled = failedQuery('57014', 'canceling statement due to user request')
+    expect(getBackgroundRetryDecision(cancelled, 1, POLICY, NOW)).toBeUndefined()
+  })
+})
+
+describe('getDatabaseRetryAt', () => {
+  it('returns null for a failure that is not a transient database failure', () => {
+    expect(getDatabaseRetryAt(new Error('parser failed'), 1, POLICY, NOW)).toBeNull()
+  })
+
+  it('returns null once the database attempts are spent', () => {
+    expect(getDatabaseRetryAt(failedQuery('40001'), 6, POLICY, NOW)).toBeNull()
+  })
+})
+
+describe('backgroundRetryAttemptCeiling', () => {
+  it('covers whichever kind of failure retries longer', () => {
+    expect(backgroundRetryAttemptCeiling(POLICY)).toBe(6)
+    expect(backgroundRetryAttemptCeiling({ ...POLICY, maxAttempts: 8 })).toBe(8)
+  })
+})