fix(jobs): progress-aware connector database retries, stuck-sync signal, strict query shapes

- the database-failure streak counts only failed runs that made no progress; a run that wrote documents (or completed a member) retries in minutes
- ten zero-progress database failures in a row log an alertable error; the connector is never disabled for them
- the classifier's query check matches only Drizzle's query error and the postgres.js query error shapes, so a client error carrying its own query stays a source failure

Author: waleedlatif1

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

@@ -278,14 +278,10 @@ describe('member sync engine decisions', () => {
       })
     })
 
-    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))
+    it('schedules the next run after the resolved retry delay', () => {
+      expect(
+        buildMemberSyncDatabaseRetryUpdate(now, 0, 'db timeout', 120 * 60 * 1000).nextMemberSyncAt
+      ).toEqual(new Date(minutesAfter(120)))
     })
   })
 
@@ -299,13 +295,23 @@ describe('member sync engine decisions', () => {
       runId: 'run-1',
       previousFailures: MAX_CONSECUTIVE_FAILURES - 1,
       errorMessage: 'failed',
+      madeProgress: false,
     }
+    const run = (status: string, membersCompleted = 0) => ({
+      status,
+      membersCompleted,
+      docsAdded: 0,
+      docsUpdated: 0,
+    })
+    const deadlock = () =>
+      new DrizzleQueryError(
+        'update private SQL',
+        ['private'],
+        Object.assign(new Error('deadlock detected'), { code: '40P01' })
+      )
 
     it('does not disable a connector one failure from the breaker over a database timeout', async () => {
-      queueTableRows(schemaMock.knowledgeConnectorMemberSyncLog, [
-        { status: 'failed' },
-        { status: 'completed' },
-      ])
+      queueTableRows(schemaMock.knowledgeConnectorMemberSyncLog, [run('failed'), run('completed')])
       const timeout = new DrizzleQueryError(
         'select private SQL',
         ['private'],
@@ -323,23 +329,29 @@ describe('member sync engine decisions', () => {
 
     it('reads the members-mode run log for the streak', async () => {
       queueTableRows(schemaMock.knowledgeConnectorMemberSyncLog, [
-        { status: 'failed' },
-        { status: 'failed' },
-        { status: 'completed' },
+        run('failed'),
+        run('failed'),
+        run('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, {
+      const update = await resolveMemberSyncFailureUpdate(deadlock(), {
         ...failure,
         previousFailures: 0,
       })
       expect(update.nextMemberSyncAt!.getTime() - before).toBeGreaterThanOrEqual(90 * 60 * 1000)
     })
 
+    it('retries within minutes after a run that completed members before the database failed', async () => {
+      queueTableRows(schemaMock.knowledgeConnectorMemberSyncLog, [run('failed'), run('failed')])
+      const before = Date.now()
+      const update = await resolveMemberSyncFailureUpdate(deadlock(), {
+        ...failure,
+        madeProgress: true,
+      })
+      expect(update.nextMemberSyncAt!.getTime() - before).toBeLessThanOrEqual(5 * 60 * 1000)
+      expect(update.memberSyncConsecutiveFailures).toBe(MAX_CONSECUTIVE_FAILURES - 1)
+    })
+
     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({

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

@@ -65,10 +65,7 @@ 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 { resolveDatabaseRetryDelayMs } from '@/lib/knowledge/connectors/sync-database-retry'
 import {
   deferConnectorSync,
   getConnectorSyncDeferral,
@@ -334,20 +331,18 @@ 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.
+ * failures already counted, and the retry waits the delay `resolveDatabaseRetryDelayMs` chose.
  */
 export function buildMemberSyncDatabaseRetryUpdate(
   now: Date,
   previousFailures: number | null | undefined,
   errorMessage: string,
-  failedRunStreak: number
+  retryDelayMs: number
 ) {
   return {
     memberSyncStatus: 'error' as const,
     lastMemberSyncError: errorMessage,
-    nextMemberSyncAt: new Date(
-      now.getTime() + databaseRetryDelayMs(failedRunStreak, previousFailures)
-    ),
+    nextMemberSyncAt: new Date(now.getTime() + retryDelayMs),
     memberSyncConsecutiveFailures: previousFailures ?? 0,
     memberSyncLockToken: null,
     memberSyncLockLeaseAt: null,
@@ -368,6 +363,8 @@ export async function resolveMemberSyncFailureUpdate(
     previousFailures: number
     errorMessage: string
     retryAfterMs?: number
+    /** Whether the run completed a member or wrote documents before it failed. */
+    madeProgress: boolean
   }
 ) {
   const now = new Date()
@@ -387,7 +384,13 @@ export async function resolveMemberSyncFailureUpdate(
       now,
       failure.previousFailures,
       failure.errorMessage,
-      await countFailedRunStreak('member', failure.connectorId, failure.runId)
+      await resolveDatabaseRetryDelayMs({
+        kind: 'member',
+        connectorId: failure.connectorId,
+        runId: failure.runId,
+        previousFailures: failure.previousFailures,
+        madeProgress: failure.madeProgress,
+      })
     )
   }
   return buildMemberSyncFailureUpdate(
@@ -2499,6 +2502,7 @@ export async function executeMemberSync(
           previousFailures: connector.memberSyncConsecutiveFailures,
           errorMessage,
           retryAfterMs,
+          madeProgress: result.membersCompleted + result.docsAdded + result.docsUpdated > 0,
         })
         const written = await db
           .update(knowledgeConnector)

apps/sim/lib/knowledge/connectors/sync-database-retry.test.ts

@@ -8,50 +8,109 @@ import {
   resetDbChainMock,
   schemaMock,
 } from '@sim/testing'
-import { beforeEach, describe, expect, it } from 'vitest'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockLogError } = vi.hoisted(() => ({ mockLogError: vi.fn() }))
+vi.mock('@sim/logger', async () => {
+  const { createMockLogger } = await import('@sim/testing/mocks/logger.mock')
+  return { createLogger: () => ({ ...createMockLogger(), error: mockLogError }) }
+})
+
 import {
-  countFailedRunStreak,
+  countZeroProgressFailedRuns,
+  DATABASE_FAILURE_ALERT_STREAK,
+  DATABASE_RETRY_AFTER_PROGRESS_MS,
   databaseRetryDelayMs,
+  resolveDatabaseRetryDelayMs,
 } from '@/lib/knowledge/connectors/sync-database-retry'
 import {
   CONNECTOR_FAILURE_BACKOFF_CAP_MINUTES,
   MAX_CONSECUTIVE_FAILURES,
 } from '@/lib/knowledge/connectors/sync-limits'
 
 const MINUTE = 60 * 1000
-const runs = (...statuses: string[]) => statuses.map((status) => ({ status }))
+const NO_WRITES = { docsAdded: 0, docsUpdated: 0, docsDeleted: 0 }
+const contentRun = (status: string, writes: Partial<typeof NO_WRITES> = {}) => ({
+  status,
+  ...NO_WRITES,
+  ...writes,
+})
+const memberRun = (status: string, writes: Record<string, number> = {}) => ({
+  status,
+  membersCompleted: 0,
+  docsAdded: 0,
+  docsUpdated: 0,
+  ...writes,
+})
 
-describe('countFailedRunStreak', () => {
+describe('countZeroProgressFailedRuns', () => {
   beforeEach(() => {
     resetDbChainMock()
   })
 
   it('counts this run plus the failed runs before it, up to the last one that did not fail', async () => {
-    queueTableRows(
-      schemaMock.knowledgeConnectorSyncLog,
-      runs('failed', 'failed', 'completed', 'failed')
-    )
-    expect(await countFailedRunStreak('content', 'c-1', 'run-1')).toBe(3)
+    queueTableRows(schemaMock.knowledgeConnectorSyncLog, [
+      contentRun('failed'),
+      contentRun('failed'),
+      contentRun('completed'),
+      contentRun('failed'),
+    ])
+    expect(await countZeroProgressFailedRuns('content', 'c-1', 'run-1')).toBe(3)
+  })
+
+  it.each([
+    ['added', { docsAdded: 5 }],
+    ['updated', { docsUpdated: 1 }],
+    ['deleted', { docsDeleted: 2 }],
+  ])('ends the streak at a failed run that %s documents', async (_label, writes) => {
+    queueTableRows(schemaMock.knowledgeConnectorSyncLog, [
+      contentRun('failed'),
+      contentRun('failed', writes),
+      contentRun('failed'),
+    ])
+    expect(await countZeroProgressFailedRuns('content', 'c-1', 'run-1')).toBe(2)
   })
 
   it('counts only this run after a success', async () => {
-    queueTableRows(schemaMock.knowledgeConnectorSyncLog, runs('completed', 'failed'))
-    expect(await countFailedRunStreak('content', 'c-1', 'run-1')).toBe(1)
+    queueTableRows(schemaMock.knowledgeConnectorSyncLog, [
+      contentRun('completed'),
+      contentRun('failed'),
+    ])
+    expect(await countZeroProgressFailedRuns('content', 'c-1', 'run-1')).toBe(1)
   })
 
   it('counts an unbroken history in full', async () => {
-    queueTableRows(schemaMock.knowledgeConnectorSyncLog, runs('failed', 'failed'))
-    expect(await countFailedRunStreak('content', 'c-1', 'run-1')).toBe(3)
+    queueTableRows(schemaMock.knowledgeConnectorSyncLog, [
+      contentRun('failed'),
+      contentRun('failed'),
+    ])
+    expect(await countZeroProgressFailedRuns('content', 'c-1', 'run-1')).toBe(3)
   })
 
   it('reads the members-mode run log for a members-mode run', async () => {
-    queueTableRows(schemaMock.knowledgeConnectorMemberSyncLog, runs('failed', 'started'))
-    expect(await countFailedRunStreak('member', 'c-1', 'run-1')).toBe(2)
+    queueTableRows(schemaMock.knowledgeConnectorMemberSyncLog, [
+      memberRun('failed'),
+      memberRun('started'),
+    ])
+    expect(await countZeroProgressFailedRuns('member', 'c-1', 'run-1')).toBe(2)
+  })
+
+  it.each([
+    ['completed a member', { membersCompleted: 1 }],
+    ['added documents', { docsAdded: 3 }],
+    ['updated documents', { docsUpdated: 1 }],
+  ])('ends a members-mode streak at a failed run that %s', async (_label, writes) => {
+    queueTableRows(schemaMock.knowledgeConnectorMemberSyncLog, [
+      memberRun('failed'),
+      memberRun('failed', writes),
+      memberRun('failed'),
+    ])
+    expect(await countZeroProgressFailedRuns('member', 'c-1', 'run-1')).toBe(2)
   })
 
   it('excludes the current run and reads only as far back as the ladder climbs', async () => {
     queueTableRows(schemaMock.knowledgeConnectorSyncLog, [])
-    await countFailedRunStreak('content', 'c-1', 'run-1')
+    await countZeroProgressFailedRuns('content', 'c-1', 'run-1')
     const where = dbChainMockFns.where.mock.calls.at(-1)?.[0]
     expect(
       hasMockCondition(
@@ -69,7 +128,7 @@ describe('countFailedRunStreak', () => {
 
   it('falls back to this run alone when the history cannot be read', async () => {
     dbChainMockFns.limit.mockRejectedValueOnce(new Error('canceling statement'))
-    expect(await countFailedRunStreak('content', 'c-1', 'run-1')).toBe(1)
+    expect(await countZeroProgressFailedRuns('content', 'c-1', 'run-1')).toBe(1)
   })
 })
 
@@ -96,3 +155,60 @@ describe('databaseRetryDelayMs', () => {
     expect(delay).toBeGreaterThanOrEqual(MAX_CONSECUTIVE_FAILURES * 30 * MINUTE)
   })
 })
+
+describe('resolveDatabaseRetryDelayMs', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  const retry = {
+    kind: 'content' as const,
+    connectorId: 'c-1',
+    runId: 'run-1',
+    previousFailures: 0,
+  }
+
+  it('retries shortly after a run that made progress, without reading the streak', async () => {
+    queueTableRows(
+      schemaMock.knowledgeConnectorSyncLog,
+      Array.from({ length: 20 }, () => contentRun('failed'))
+    )
+    const delay = await resolveDatabaseRetryDelayMs({ ...retry, madeProgress: true })
+    expect(delay).toBeGreaterThanOrEqual(DATABASE_RETRY_AFTER_PROGRESS_MS)
+    expect(delay).toBeLessThanOrEqual(DATABASE_RETRY_AFTER_PROGRESS_MS + MINUTE)
+    expect(dbChainMockFns.select).not.toHaveBeenCalled()
+  })
+
+  it('climbs the ladder by the zero-progress streak for a run that made none', async () => {
+    queueTableRows(schemaMock.knowledgeConnectorSyncLog, [
+      contentRun('failed'),
+      contentRun('failed'),
+      contentRun('completed'),
+    ])
+    const delay = await resolveDatabaseRetryDelayMs({ ...retry, madeProgress: false })
+    expect(delay).toBeGreaterThanOrEqual(90 * MINUTE)
+    expect(delay).toBeLessThanOrEqual(91 * MINUTE)
+  })
+
+  it('reports a streak that reaches the alert threshold at error level, without disabling', async () => {
+    queueTableRows(
+      schemaMock.knowledgeConnectorSyncLog,
+      Array.from({ length: DATABASE_FAILURE_ALERT_STREAK - 1 }, () => contentRun('failed'))
+    )
+    await resolveDatabaseRetryDelayMs({ ...retry, madeProgress: false })
+    expect(mockLogError).toHaveBeenCalledWith(
+      'Connector sync keeps failing on the database without progress',
+      { connectorId: 'c-1', kind: 'content', zeroProgressFailedRuns: DATABASE_FAILURE_ALERT_STREAK }
+    )
+  })
+
+  it('stays quiet below the alert threshold', async () => {
+    queueTableRows(
+      schemaMock.knowledgeConnectorSyncLog,
+      Array.from({ length: DATABASE_FAILURE_ALERT_STREAK - 2 }, () => contentRun('failed'))
+    )
+    await resolveDatabaseRetryDelayMs({ ...retry, madeProgress: false })
+    expect(mockLogError).not.toHaveBeenCalled()
+  })
+})