fix(jobs): escalate database retries without spending the breaker, cover members mode

- connector and members-mode syncs climb the failure ladder by the failed-run streak read from their run logs; database failures never advance the auto-disable counter
- classify 25P03 as capacity and ECONNREFUSED/EHOSTUNREACH/ENOTFOUND/EAI_AGAIN (with a query) as connection
- in-process document processing keeps recording database failures as failed, documented
- offer Retry for a pending document whose dispatch or deferred retry is past the retry API's grace
- correct the processing task's retry-ceiling comment

Author: waleedlatif1

apps/sim/app/workspace/[workspaceId]/knowledge/[id]/base.tsx

@@ -49,6 +49,7 @@ import {
   KNOWLEDGE_DOCUMENT_PROCESSING_STALE_THRESHOLD_MS,
 } from '@/lib/knowledge/constants'
 import {
+  canRetryDocumentProcessing,
   type DocumentSortField,
   getDocumentIndexingStatus,
   type SortOrder,
@@ -1569,7 +1570,7 @@ export function KnowledgeBase({
         }
         onRetry={
           contextMenuDocument &&
-          getDocumentIndexingStatus(contextMenuDocument) === 'failed' &&
+          canRetryDocumentProcessing(contextMenuDocument) &&
           selectedDocumentCount === 1 &&
           userPermissions.canEdit
             ? () => handleRetryDocument(contextMenuDocument.id)

apps/sim/background/knowledge-processing.ts

@@ -327,9 +327,10 @@ export const processDocument = task({
   machine: 'medium-2x',
   retry: {
     /**
-     * The ceiling for database retries; `catchError` stops other failures at
-     * `KB_CONFIG_MAX_ATTEMPTS`. An out-of-memory kill never reaches `catchError`,
-     * so it may use the full ceiling.
+     * 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),

apps/sim/lib/api/contracts/knowledge/documents.ts

@@ -293,6 +293,8 @@ export const documentDataSchema = z
     processingOutcome: z.literal('skipped').nullable().default(null),
     /** When indexing was last dispatched to a worker, which precedes a worker starting it. */
     processingQueuedAt: nullableWireDateSchema.optional(),
+    /** When a deferred retry of a `pending` document is due. */
+    processingDeferredUntil: nullableWireDateSchema.optional(),
     processingStartedAt: nullableWireDateSchema.optional(),
     processingCompletedAt: nullableWireDateSchema.optional(),
     processingError: z.string().nullable().optional(),

apps/sim/lib/knowledge/api/internal-route.test.ts

@@ -13,6 +13,7 @@ import {
   internalKnowledgeProvenanceUserId,
   resolveInternalKnowledgeBillingAttribution,
   toInternalKnowledgeConnector,
+  toInternalKnowledgeDocument,
 } from '@/lib/knowledge/api/internal-route'
 import { resolveKnowledgeAttributedUserId } from '@/lib/knowledge/application/billing'
 
@@ -133,6 +134,30 @@ describe('internal Knowledge execution attribution', () => {
   })
 })
 
+describe('toInternalKnowledgeDocument', () => {
+  it('presents when a deferred retry is due, so the Retry action can match the API', () => {
+    const deferredUntil = new Date('2026-09-01T12:00:00.000Z')
+    const presented = toInternalKnowledgeDocument({
+      id: 'doc-1',
+      knowledgeBaseId: 'kb-1',
+      filename: 'a.txt',
+      fileUrl: 'https://example.com/a.txt',
+      fileSize: 1,
+      mimeType: 'text/plain',
+      chunkCount: 0,
+      tokenCount: 0,
+      characterCount: 0,
+      processingStatus: 'pending',
+      enabled: true,
+      uploadedAt: new Date('2026-08-01T00:00:00.000Z'),
+      processingQueuedAt: new Date('2026-08-01T00:00:00.000Z'),
+      processingDeferredUntil: deferredUntil,
+    })
+    expect(presented.processingDeferredUntil).toBe(deferredUntil.toISOString())
+    expect(presented.processingQueuedAt).toBe('2026-08-01T00:00:00.000Z')
+  })
+})
+
 describe('toInternalKnowledgeConnector', () => {
   const row = {
     id: 'connector-1',

apps/sim/lib/knowledge/api/internal-route.ts

@@ -89,6 +89,7 @@ export function toInternalKnowledgeDocument<
   T extends {
     uploadedAt: Date | string
     processingQueuedAt?: Date | string | null
+    processingDeferredUntil?: Date | string | null
     processingStartedAt?: Date | string | null
     processingCompletedAt?: Date | string | null
     date1?: Date | string | null
@@ -99,6 +100,7 @@ export function toInternalKnowledgeDocument<
     ...document,
     uploadedAt: serializeDate(document.uploadedAt),
     processingQueuedAt: serializeNullableDate(document.processingQueuedAt ?? null),
+    processingDeferredUntil: serializeNullableDate(document.processingDeferredUntil ?? null),
     processingStartedAt: serializeNullableDate(document.processingStartedAt ?? null),
     processingCompletedAt: serializeNullableDate(document.processingCompletedAt ?? null),
     date1: serializeNullableDate(document.date1 ?? null),

apps/sim/lib/knowledge/connectors/member-sync-engine.test.ts

@@ -1,7 +1,9 @@
 /**
  * @vitest-environment node
  */
-import { describe, expect, it, vi } from 'vitest'
+import { queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing'
+import { DrizzleQueryError } from 'drizzle-orm/errors'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
 
 vi.mock('@/connectors/registry.server', () => ({ CONNECTOR_REGISTRY: {} }))
 vi.mock('@/lib/knowledge/documents/service', () => ({
@@ -20,11 +22,13 @@ vi.mock('@/lib/billing/core/workspace-access', () => ({
 vi.mock('@/lib/credential-groups/availability', () => ({ isCredentialGroupsAvailable: vi.fn() }))
 
 import {
+  buildMemberSyncDatabaseRetryUpdate,
   buildMemberSyncFailureUpdate,
   deriveMemberActive,
   memberFailureBackoffMs,
   memberNextAttemptAt,
   nextMemberSyncTime,
+  resolveMemberSyncFailureUpdate,
   shouldListFully,
 } from '@/lib/knowledge/connectors/member-sync-engine'
 import {
@@ -254,6 +258,97 @@ describe('member sync engine decisions', () => {
     })
   })
 
+  describe('buildMemberSyncDatabaseRetryUpdate', () => {
+    const now = new Date('2026-09-01T12:00:00Z')
+    const minutesAfter = (mins: number) => now.getTime() + mins * 60 * 1000
+
+    it('keeps the error visible without advancing the breaker', () => {
+      const update = buildMemberSyncDatabaseRetryUpdate(
+        now,
+        MAX_CONSECUTIVE_FAILURES - 1,
+        'db timeout',
+        40
+      )
+      expect(update).toMatchObject({
+        memberSyncStatus: 'error',
+        lastMemberSyncError: 'db timeout',
+        memberSyncConsecutiveFailures: MAX_CONSECUTIVE_FAILURES - 1,
+        memberSyncLockToken: null,
+        memberSyncLockLeaseAt: null,
+      })
+    })
+
+    it('climbs the failure ladder with the failed-run streak, up to its ceiling', () => {
+      const at = (streak: number) =>
+        buildMemberSyncDatabaseRetryUpdate(now, 0, 'db timeout', streak).nextMemberSyncAt.getTime()
+      expect(at(1)).toBeGreaterThanOrEqual(minutesAfter(30))
+      expect(at(1)).toBeLessThanOrEqual(minutesAfter(31))
+      expect(at(4)).toBeGreaterThanOrEqual(minutesAfter(120))
+      expect(at(4)).toBeLessThanOrEqual(minutesAfter(121))
+      expect(at(500)).toBeLessThanOrEqual(minutesAfter(24 * 60 + 1))
+    })
+  })
+
+  describe('resolveMemberSyncFailureUpdate', () => {
+    beforeEach(() => {
+      resetDbChainMock()
+    })
+
+    const failure = {
+      connectorId: 'c-1',
+      runId: 'run-1',
+      previousFailures: MAX_CONSECUTIVE_FAILURES - 1,
+      errorMessage: 'failed',
+    }
+
+    it('does not disable a connector one failure from the breaker over a database timeout', async () => {
+      queueTableRows(schemaMock.knowledgeConnectorMemberSyncLog, [
+        { status: 'failed' },
+        { status: 'completed' },
+      ])
+      const timeout = new DrizzleQueryError(
+        'select private SQL',
+        ['private'],
+        Object.assign(new Error('canceling statement due to statement timeout'), {
+          code: '57014',
+        })
+      )
+      const update = await resolveMemberSyncFailureUpdate(timeout, failure)
+      expect(update).toMatchObject({
+        memberSyncStatus: 'error',
+        memberSyncConsecutiveFailures: MAX_CONSECUTIVE_FAILURES - 1,
+      })
+      expect(update.nextMemberSyncAt).not.toBeNull()
+    })
+
+    it('reads the members-mode run log for the streak', async () => {
+      queueTableRows(schemaMock.knowledgeConnectorMemberSyncLog, [
+        { status: 'failed' },
+        { status: 'failed' },
+        { status: 'completed' },
+      ])
+      const deadlock = new DrizzleQueryError(
+        'update private SQL',
+        ['private'],
+        Object.assign(new Error('deadlock detected'), { code: '40P01' })
+      )
+      const before = Date.now()
+      const update = await resolveMemberSyncFailureUpdate(deadlock, {
+        ...failure,
+        previousFailures: 0,
+      })
+      expect(update.nextMemberSyncAt!.getTime() - before).toBeGreaterThanOrEqual(90 * 60 * 1000)
+    })
+
+    it('still disables at the breaker for a failure the database did not cause', async () => {
+      const update = await resolveMemberSyncFailureUpdate(new Error('source broke'), failure)
+      expect(update).toMatchObject({
+        memberSyncStatus: 'disabled',
+        memberSyncConsecutiveFailures: MAX_CONSECUTIVE_FAILURES,
+      })
+    })
+  })
+
   describe('nextMemberSyncTime', () => {
     const now = new Date('2026-09-01T12:00:00Z')
 

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

@@ -11,7 +11,7 @@ import {
   knowledgeDocumentObservation,
 } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { getErrorMessage, toError } from '@sim/utils/errors'
+import { getErrorMessage, getTransientDatabaseFailure, toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { randomInt } from '@sim/utils/random'
 import { and, asc, eq, gt, inArray, isNull, lte, notExists, sql } from 'drizzle-orm'
@@ -65,6 +65,10 @@ import {
 } from '@/lib/knowledge/connectors/member-observations'
 import { inviteWorkspaceMembersToCredentialGroup } from '@/lib/knowledge/connectors/member-provisioning'
 import { runConnectorContentPass } from '@/lib/knowledge/connectors/sync-content-pass'
+import {
+  countFailedRunStreak,
+  databaseRetryDelayMs,
+} from '@/lib/knowledge/connectors/sync-database-retry'
 import {
   deferConnectorSync,
   getConnectorSyncDeferral,
@@ -327,6 +331,73 @@ export function buildMemberSyncFailureUpdate(
   }
 }
 
+/**
+ * The connector row written after the database, not the source, failed a members-mode run. The
+ * members-mode counterpart of `buildSyncDatabaseRetryUpdate`: the breaker keeps only the source
+ * failures already counted, while the retry climbs the failure ladder by the failed-run streak.
+ */
+export function buildMemberSyncDatabaseRetryUpdate(
+  now: Date,
+  previousFailures: number | null | undefined,
+  errorMessage: string,
+  failedRunStreak: number
+) {
+  return {
+    memberSyncStatus: 'error' as const,
+    lastMemberSyncError: errorMessage,
+    nextMemberSyncAt: new Date(
+      now.getTime() + databaseRetryDelayMs(failedRunStreak, previousFailures)
+    ),
+    memberSyncConsecutiveFailures: previousFailures ?? 0,
+    memberSyncLockToken: null,
+    memberSyncLockLeaseAt: null,
+    updatedAt: now,
+  }
+}
+
+/**
+ * The connector row a failed members-mode run writes. A deterministic capacity rejection waits for
+ * an operator, a transient database failure retries without touching the breaker, and anything
+ * else climbs the ladder toward auto-disable.
+ */
+export async function resolveMemberSyncFailureUpdate(
+  error: unknown,
+  failure: {
+    connectorId: string
+    runId: string
+    previousFailures: number
+    errorMessage: string
+    retryAfterMs?: number
+  }
+) {
+  const now = new Date()
+  if (error instanceof ConnectorSyncCapacityError) {
+    return {
+      memberSyncStatus: 'error' as const,
+      lastMemberSyncError: failure.errorMessage,
+      nextMemberSyncAt: null,
+      memberSyncConsecutiveFailures: failure.previousFailures,
+      memberSyncLockToken: null,
+      memberSyncLockLeaseAt: null,
+      updatedAt: now,
+    }
+  }
+  if (getTransientDatabaseFailure(error)) {
+    return buildMemberSyncDatabaseRetryUpdate(
+      now,
+      failure.previousFailures,
+      failure.errorMessage,
+      await countFailedRunStreak('member', failure.connectorId, failure.runId)
+    )
+  }
+  return buildMemberSyncFailureUpdate(
+    now,
+    failure.previousFailures,
+    failure.errorMessage,
+    failure.retryAfterMs
+  )
+}
+
 /**
  * When a member who completed is next due: exactly one interval on, with no
  * jitter, so they are due whenever the connector's own (jittered) run lands.
@@ -2422,23 +2493,13 @@ export async function executeMemberSync(
       logger.error('Member sync failed', { connectorId, runId, error: errorMessage, diagnostic })
       try {
         await failMemberSyncLog(runId, result, errorMessage)
-        const failureUpdate =
-          error instanceof ConnectorSyncCapacityError
-            ? {
-                memberSyncStatus: 'error' as const,
-                lastMemberSyncError: errorMessage,
-                nextMemberSyncAt: null,
-                memberSyncConsecutiveFailures: connector.memberSyncConsecutiveFailures,
-                memberSyncLockToken: null,
-                memberSyncLockLeaseAt: null,
-                updatedAt: new Date(),
-              }
-            : buildMemberSyncFailureUpdate(
-                new Date(),
-                connector.memberSyncConsecutiveFailures,
-                errorMessage,
-                retryAfterMs
-              )
+        const failureUpdate = await resolveMemberSyncFailureUpdate(error, {
+          connectorId,
+          runId,
+          previousFailures: connector.memberSyncConsecutiveFailures,
+          errorMessage,
+          retryAfterMs,
+        })
         const written = await db
           .update(knowledgeConnector)
           .set(failureUpdate)