fix(jobs): bound the retry-history read, count member purges, match driver connection errors

- the connector run-history read runs under short statement and lock timeouts and falls back to this run alone
- members-mode progress counts lifecycle purges (docs_purged in the run log)
- postgres.js connection codes count only on an error the driver built, or under a query error; the driver's query signature is its four own properties, which a refused connection carries with no SQL yet
- real-error PostgreSQL test for terminated transactions, refused connections, and an ending pool, wired into CI
- knowledge-processing tests use the static task import

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

@@ -49,6 +49,7 @@ import {
   DOCUMENT_PROCESSING_RETRY_POLICY,
   DocumentProcessingDatabaseRetryError,
   getDocumentProcessingRetry,
+  processDocument,
   resolveQuotaContinuationDelayMs,
   runDocumentProcessing,
 } from '@/background/knowledge-processing'
@@ -821,14 +822,10 @@ 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 () => {
-    const { processDocument } = await import('@/background/knowledge-processing')
-
     expect(processDocument.retry?.maxAttempts).toBe(
       Math.max(
         DOCUMENT_PROCESSING_RETRY_POLICY.maxAttempts,

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

@@ -25,8 +25,10 @@ import {
   buildMemberSyncDatabaseRetryUpdate,
   buildMemberSyncFailureUpdate,
   deriveMemberActive,
+  type MemberSyncResult,
   memberFailureBackoffMs,
   memberNextAttemptAt,
+  memberRunMadeProgress,
   nextMemberSyncTime,
   resolveMemberSyncFailureUpdate,
   shouldListFully,
@@ -285,6 +287,28 @@ describe('member sync engine decisions', () => {
     })
   })
 
+  describe('memberRunMadeProgress', () => {
+    const idle = {
+      membersCompleted: 0,
+      docsAdded: 0,
+      docsUpdated: 0,
+      docsDeleted: 0,
+    } as MemberSyncResult
+
+    it('reports no progress for a run that wrote nothing', () => {
+      expect(memberRunMadeProgress(idle)).toBe(false)
+    })
+
+    it.each([
+      ['completed a member', { membersCompleted: 1 }],
+      ['added documents', { docsAdded: 2 }],
+      ['updated documents', { docsUpdated: 1 }],
+      ['purged documents in the lifecycle pass', { docsDeleted: 3 }],
+    ])('reports progress for a run that %s', (_label, writes) => {
+      expect(memberRunMadeProgress({ ...idle, ...writes })).toBe(true)
+    })
+  })
+
   describe('resolveMemberSyncFailureUpdate', () => {
     beforeEach(() => {
       resetDbChainMock()

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

@@ -350,6 +350,14 @@ export function buildMemberSyncDatabaseRetryUpdate(
   }
 }
 
+/**
+ * Whether a members-mode run moved the sync forward: it completed a member, or wrote documents.
+ * `docsDeleted` holds the document lifecycle's purges, which the run log records as `docs_purged`.
+ */
+export function memberRunMadeProgress(result: MemberSyncResult): boolean {
+  return result.membersCompleted + result.docsAdded + result.docsUpdated + result.docsDeleted > 0
+}
+
 /**
  * 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
@@ -2502,7 +2510,7 @@ export async function executeMemberSync(
           previousFailures: connector.memberSyncConsecutiveFailures,
           errorMessage,
           retryAfterMs,
-          madeProgress: result.membersCompleted + result.docsAdded + result.docsUpdated > 0,
+          madeProgress: memberRunMadeProgress(result),
         })
         const written = await db
           .update(knowledgeConnector)

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

@@ -21,6 +21,8 @@ import {
   DATABASE_FAILURE_ALERT_STREAK,
   DATABASE_RETRY_AFTER_PROGRESS_MS,
   databaseRetryDelayMs,
+  RUN_HISTORY_LOCK_TIMEOUT_MS,
+  RUN_HISTORY_STATEMENT_TIMEOUT_MS,
   resolveDatabaseRetryDelayMs,
 } from '@/lib/knowledge/connectors/sync-database-retry'
 import {
@@ -40,6 +42,7 @@ const memberRun = (status: string, writes: Record<string, number> = {}) => ({
   membersCompleted: 0,
   docsAdded: 0,
   docsUpdated: 0,
+  docsPurged: 0,
   ...writes,
 })
 
@@ -99,6 +102,7 @@ describe('countZeroProgressFailedRuns', () => {
     ['completed a member', { membersCompleted: 1 }],
     ['added documents', { docsAdded: 3 }],
     ['updated documents', { docsUpdated: 1 }],
+    ['purged documents', { docsPurged: 4 }],
   ])('ends a members-mode streak at a failed run that %s', async (_label, writes) => {
     queueTableRows(schemaMock.knowledgeConnectorMemberSyncLog, [
       memberRun('failed'),
@@ -126,6 +130,35 @@ describe('countZeroProgressFailedRuns', () => {
     )
   })
 
+  it('bounds the history read with its own statement and lock timeouts', async () => {
+    queueTableRows(schemaMock.knowledgeConnectorSyncLog, [])
+    await countZeroProgressFailedRuns('content', 'c-1', 'run-1')
+    const bound = dbChainMockFns.execute.mock.calls[0]?.[0] as {
+      toSQL: () => { sql: string; params: unknown[] }
+    }
+    const { sql, params } = bound.toSQL()
+    expect(sql).toContain("set_config('statement_timeout'")
+    expect(sql).toContain("set_config('lock_timeout'")
+    expect(params).toEqual([
+      String(RUN_HISTORY_STATEMENT_TIMEOUT_MS),
+      String(RUN_HISTORY_LOCK_TIMEOUT_MS),
+    ])
+    expect(dbChainMockFns.execute.mock.invocationCallOrder[0]).toBeLessThan(
+      dbChainMockFns.select.mock.invocationCallOrder[0]
+    )
+  })
+
+  it('falls back to this run alone when the history read times out', async () => {
+    queueTableRows(schemaMock.knowledgeConnectorSyncLog, [
+      contentRun('failed'),
+      contentRun('failed'),
+    ])
+    dbChainMockFns.limit.mockRejectedValueOnce(
+      Object.assign(new Error('canceling statement due to statement timeout'), { code: '57014' })
+    )
+    expect(await countZeroProgressFailedRuns('content', 'c-1', 'run-1')).toBe(1)
+  })
+
   it('falls back to this run alone when the history cannot be read', async () => {
     dbChainMockFns.limit.mockRejectedValueOnce(new Error('canceling statement'))
     expect(await countZeroProgressFailedRuns('content', 'c-1', 'run-1')).toBe(1)

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

@@ -3,7 +3,8 @@ import { knowledgeConnectorMemberSyncLog, knowledgeConnectorSyncLog } from '@sim
 import { createLogger } from '@sim/logger'
 import { describeError } from '@sim/utils/errors'
 import { randomInt } from '@sim/utils/random'
-import { and, desc, eq, ne } from 'drizzle-orm'
+import { and, desc, eq, ne, sql } from 'drizzle-orm'
+import type { DbTransaction } from '@/lib/db/types'
 import {
   CONNECTOR_FAILURE_BACKOFF_CAP_MINUTES,
   CONNECTOR_FAILURE_BACKOFF_STEP_MINUTES,
@@ -33,6 +34,14 @@ const LADDER_RUNGS = Math.ceil(
   CONNECTOR_FAILURE_BACKOFF_CAP_MINUTES / CONNECTOR_FAILURE_BACKOFF_STEP_MINUTES
 )
 
+/**
+ * Limits on the run-history read. It runs on the failure path, usually while the database is the
+ * thing failing, so it must give up fast rather than queue behind the slow window; a read that
+ * times out falls back to counting this run alone.
+ */
+export const RUN_HISTORY_STATEMENT_TIMEOUT_MS = 2_000
+export const RUN_HISTORY_LOCK_TIMEOUT_MS = 500
+
 /** Which run log a sync writes: the content sync's, or the members-mode run's. */
 export type SyncRunLogKind = 'content' | 'member'
 
@@ -43,14 +52,15 @@ interface LoggedRun {
 
 /** Earlier runs of this connector, newest first, and whether each one moved the sync forward. */
 async function readEarlierRuns(
+  tx: DbTransaction,
   kind: SyncRunLogKind,
   connectorId: string,
   runId: string,
   limit: number
 ): Promise<LoggedRun[]> {
   if (kind === 'content') {
     const log = knowledgeConnectorSyncLog
-    const rows = await db
+    const rows = await tx
       .select({
         status: log.status,
         docsAdded: log.docsAdded,
@@ -67,20 +77,21 @@ async function readEarlierRuns(
     }))
   }
   const log = knowledgeConnectorMemberSyncLog
-  const rows = await db
+  const rows = await tx
     .select({
       status: log.status,
       membersCompleted: log.membersCompleted,
       docsAdded: log.docsAdded,
       docsUpdated: log.docsUpdated,
+      docsPurged: log.docsPurged,
     })
     .from(log)
     .where(and(eq(log.connectorId, connectorId), ne(log.id, runId)))
     .orderBy(desc(log.startedAt))
     .limit(limit)
   return rows.map((row) => ({
     status: row.status,
-    progressed: row.membersCompleted + row.docsAdded + row.docsUpdated > 0,
+    progressed: row.membersCompleted + row.docsAdded + row.docsUpdated + row.docsPurged > 0,
   }))
 }
 
@@ -93,17 +104,24 @@ async function readEarlierRuns(
  * cannot spend the breaker that disables connectors for persistent source failures. The run log
  * already records every attempt and what it wrote, so it measures the streak instead: a statement
  * that fails every run without progress still backs off rung by rung, while a run that added,
- * updated, or deleted documents (or, in members mode, completed a member) before failing ends it.
- * The read uses the log's `(connector_id, started_at DESC)` index. If it fails too, the streak
- * counts only this run and the caller's own floor applies.
+ * updated, or deleted documents (in members mode, purged by the document lifecycle and logged as
+ * `docs_purged`), or completed a member, before failing ends it.
+ * The read uses the log's `(connector_id, started_at DESC)` index and is bounded by
+ * {@link RUN_HISTORY_STATEMENT_TIMEOUT_MS}. If it fails or times out, the streak counts only this
+ * run and the caller's own floor applies.
  */
 export async function countZeroProgressFailedRuns(
   kind: SyncRunLogKind,
   connectorId: string,
   runId: string
 ): Promise<number> {
   try {
-    const earlier = await readEarlierRuns(kind, connectorId, runId, LADDER_RUNGS - 1)
+    const earlier = await db.transaction(async (tx) => {
+      await tx.execute(
+        sql`SELECT set_config('statement_timeout', ${String(RUN_HISTORY_STATEMENT_TIMEOUT_MS)}, true), set_config('lock_timeout', ${String(RUN_HISTORY_LOCK_TIMEOUT_MS)}, true)`
+      )
+      return readEarlierRuns(tx, kind, connectorId, runId, LADDER_RUNGS - 1)
+    })
     const streakEnd = earlier.findIndex((run) => run.status !== 'failed' || run.progressed)
     return 1 + (streakEnd === -1 ? earlier.length : streakEnd)
   } catch (error) {

packages/db/script-migrations/0021_embedding_search_connector.test.ts

@@ -20,8 +20,19 @@ const SERVER_MESSAGES: Record<string, string> = {
   '40P01': 'deadlock detected',
 }
 
-/** A driver error carrying a SQLSTATE, the shape `postgres` throws: the server's message as is. */
+/**
+ * The error `postgres` throws for `code`: a SQLSTATE carries the server's message as is, and a
+ * lost connection is the driver's own connection error (`write <code> <host:port>`).
+ */
 function postgresError(code: string, message = SERVER_MESSAGES[code] ?? 'failed'): Error {
+  if (code.startsWith('CONNECTION_')) {
+    return Object.assign(new Error(`write ${code} localhost:5432`), {
+      code,
+      errno: code,
+      address: 'localhost',
+      port: 5432,
+    })
+  }
   return Object.assign(new Error(message), { code })
 }
 

packages/db/scripts/database-failure-classification.postgres.test.ts

@@ -0,0 +1,80 @@
+import { migrationTestDatabaseUrl } from '@sim/db/scripts/migration-fixture'
+import { classifyDatabaseFailure } from '@sim/utils/errors'
+import { sql as statement } from 'drizzle-orm'
+import { drizzle } from 'drizzle-orm/postgres-js'
+import postgres from 'postgres'
+import { describe, expect, it } from 'vitest'
+
+/** A port on the database host with nothing listening, so a connection is refused at once. */
+function closedPortUrl(): string {
+  const url = new URL(migrationTestDatabaseUrl!)
+  url.port = '1'
+  return url.toString()
+}
+
+async function rejectionOf(work: () => Promise<unknown>): Promise<unknown> {
+  try {
+    await work()
+  } catch (error) {
+    return error
+  }
+  throw new Error('Expected the work to fail')
+}
+
+/**
+ * Classifies the errors postgres.js and Drizzle actually raise, rather than hand-built shapes:
+ * a transaction that loses its connection is rejected with the driver's bare connection error,
+ * with no query attached and no Drizzle wrapper, and must still read as a connection failure.
+ */
+describe.skipIf(!migrationTestDatabaseUrl)('database failure classification', () => {
+  it('reads a connection terminated mid-transaction as a connection failure', async () => {
+    const admin = postgres(migrationTestDatabaseUrl!, { max: 1 })
+    const client = postgres(migrationTestDatabaseUrl!, { max: 1 })
+    try {
+      const error = await rejectionOf(() =>
+        drizzle(client).transaction(async (tx) => {
+          const [row] = await tx.execute<{ pid: number }>(statement`SELECT pg_backend_pid() AS pid`)
+          await admin`SELECT pg_terminate_backend(${row.pid})`
+          await tx.execute(statement`SELECT pg_sleep(0.2)`)
+        })
+      )
+      expect(classifyDatabaseFailure(error)).toBe('connection')
+    } finally {
+      await client.end({ timeout: 1 }).catch(() => {})
+      await admin.end()
+    }
+  })
+
+  it('reads a refused connection as a connection failure, in a query and in a transaction', async () => {
+    const client = postgres(closedPortUrl(), { max: 1, connect_timeout: 2 })
+    try {
+      const db = drizzle(client)
+      expect(
+        classifyDatabaseFailure(await rejectionOf(() => db.execute(statement`SELECT 1`)))
+      ).toBe('connection')
+      expect(
+        classifyDatabaseFailure(
+          await rejectionOf(() => db.transaction((tx) => tx.execute(statement`SELECT 1`)))
+        )
+      ).toBe('connection')
+    } finally {
+      await client.end({ timeout: 1 }).catch(() => {})
+    }
+  })
+
+  it('reads a transaction begun on an ending pool as a connection failure', async () => {
+    const client = postgres(migrationTestDatabaseUrl!, { max: 1 })
+    await client`SELECT 1`
+    const ending = client.end({ timeout: 5 })
+    const error = await rejectionOf(() =>
+      drizzle(client).transaction((tx) => tx.execute(statement`SELECT 1`))
+    )
+    await ending
+    expect(classifyDatabaseFailure(error)).toBe('connection')
+  })
+
+  it('does not read a refused connection from another client as a database failure', async () => {
+    const error = await rejectionOf(() => fetch(`http://${new URL(closedPortUrl()).host}/`))
+    expect(classifyDatabaseFailure(error)).toBe('permanent')
+  })
+})