fix(search): preserve connector failure diagnostics (#7958)

Author: waleedlatif1

apps/sim/connectors/google-workspace/api-errors.test.ts

@@ -21,6 +21,19 @@ afterEach(() => {
 })
 
 describe('Google API diagnostics', () => {
+  it('retains the Calendar account-status reason without retaining the response message', async () => {
+    const error = await readGoogleApiError(failure(403, 'notACalendarUser'), 'calendar.events.list')
+    expect(error.reasonsComplete).toBe(true)
+    expect(error.rateLimited).toBe(false)
+    expect(getConnectorFailureDiagnostic(error)).toMatchObject({
+      status: 403,
+      operation: 'calendar.events.list',
+      reasons: ['notACalendarUser'],
+      reasonState: 'present',
+    })
+    expect(JSON.stringify(error)).not.toContain(RESPONSE_SECRET)
+  })
+
   it.each([
     [400, 'badRequest', 'request_rejected'],
     [403, 'forbidden', 'authorization'],
@@ -227,6 +240,7 @@ describe('Google API retries', () => {
   it.each([
     [400, 'failedPrecondition'],
     [403, 'forbidden'],
+    [403, 'notACalendarUser'],
     [404, 'notFound'],
   ] as const)('does not retry or suppress %s %s', async (status, reason) => {
     const fetch = vi.fn().mockResolvedValueOnce(failure(status, reason))

apps/sim/connectors/google-workspace/api-errors.ts

@@ -32,6 +32,7 @@ const SAFE_REASONS = new Set([
   'internalError',
   'invalid',
   'invalidArgument',
+  'notACalendarUser',
   'notFound',
   'quotaExceeded',
   'rateLimitExceeded',

apps/sim/lib/knowledge/connectors/connector-error.test.ts

@@ -73,6 +73,24 @@ describe('connector failure diagnostics', () => {
     })
   })
 
+  it.each([
+    ['canceling statement due to statement timeout', 'statement_timeout'],
+    ['canceling statement due to user request', 'user_cancel'],
+  ])('distinguishes cancellations with the same SQLSTATE: %s', (message, databaseReason) => {
+    const error = new DrizzleQueryError(
+      'select private_column from private_source',
+      ['private-value'],
+      Object.assign(new Error(message), { code: '57014', detail: 'private driver detail' })
+    )
+    expect(getConnectorFailureDiagnostic(error)).toEqual({
+      category: 'database',
+      code: '57014',
+      databaseReason,
+      message: 'Database request failed (SQLSTATE 57014).',
+    })
+    expect(JSON.stringify(getConnectorFailureDiagnostic(error))).not.toContain('private')
+  })
+
   it('suppresses query text even when the driver provides no error code', () => {
     expect(
       getConnectorFailureDiagnostic(new DrizzleQueryError('select private', ['private'], null))

apps/sim/lib/knowledge/connectors/connector-error.ts

@@ -1,4 +1,9 @@
-import { findCause, getPostgresErrorCode } from '@sim/utils/errors'
+import {
+  findCause,
+  getPostgresCancellationReason,
+  getPostgresErrorCode,
+  type PostgresCancellationReason,
+} from '@sim/utils/errors'
 import { DrizzleQueryError } from 'drizzle-orm/errors'
 import { getEmbeddingAPIError } from '@/lib/embeddings/api-error'
 import {
@@ -13,6 +18,7 @@ export interface ConnectorFailureDiagnostic {
   message: string
   status?: number
   code?: string
+  databaseReason?: PostgresCancellationReason
   operation?: string
   reasons?: readonly string[]
   reasonState?: ConnectorSourceReasonState
@@ -57,9 +63,11 @@ function classifyFailure(error: unknown): ConnectorFailureDiagnostic | null {
     }
   }
   if (code && /^(?:[0-9][0-9A-Z]|F0|HV|P0|XX)[0-9A-Z]{3}$/.test(code)) {
+    const databaseReason = getPostgresCancellationReason(error)
     return {
       category: 'database',
       code,
+      ...(databaseReason ? { databaseReason } : {}),
       message: `Database request failed (SQLSTATE ${code}).`,
     }
   }

apps/sim/lib/knowledge/connectors/google-company-scheduler.test.ts

@@ -222,21 +222,25 @@ describe('durable Google company user scheduling', () => {
     expect(f.saved().cursor!.length).toBeLessThan(2048)
   })
 
-  it.each(['google_calendar', 'google_drive'])(
-    'retains an unresolved %s user and continues other users',
-    async (provider) => {
+  it.each([
+    ['google_calendar', []],
+    ['google_calendar', ['notACalendarUser']],
+    ['google_drive', []],
+  ] as const)(
+    'retains a failed %s user (%j) and continues other users',
+    async (provider, reasons) => {
       mocks.directory.mockResolvedValue({ users: [user('a'), user('z')] })
       const f = fixture(provider)
       f.list.mockRejectedValueOnce(
         provider === 'google_drive'
           ? new GoogleDriveApiError(403, [], 'drive.files.list', false)
-          : new GoogleApiError('calendar.events.list', 403, [], false)
+          : new GoogleApiError('calendar.events.list', 403, reasons, reasons.length > 0)
       )
       await f.step(4)
       expect(f.rows.get('a:content')).toMatchObject({
         complete: false,
         attempts: 1,
-        failure: { status: 403, reasons: [] },
+        failure: { status: 403, reasons },
       })
       expect(f.rows.get('z:content')?.complete).toBe(true)
       expect(f.saved()).toMatchObject({

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

@@ -13,6 +13,7 @@ import {
   schemaMock,
 } from '@sim/testing'
 import { generateShortId } from '@sim/utils/id'
+import { DrizzleQueryError } from 'drizzle-orm/errors'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import * as connectorTokens from '@/lib/knowledge/connectors/access-token'
 import { executeSync, isConnectorRunnableStatus } from '@/lib/knowledge/connectors/sync-engine'
@@ -83,6 +84,12 @@ const { mockGetDocument, mockMapTags, mockListDocuments } = vi.hoisted(() => ({
   mockListDocuments: vi.fn(),
 }))
 
+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 }) }
+})
+
 vi.mock('@/lib/billing/core/billing-attribution', () => ({
   assertBillingAttributionOwner: vi.fn(),
   assertBillingAttributionSnapshot: (snapshot: unknown) => snapshot,
@@ -2840,6 +2847,46 @@ describe('executeSync heartbeats during the listing phase', () => {
     dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'c-1', accessMode: 'workspace' }])
   }
 
+  it.each(['workspace', 'admin'] as const)(
+    'diagnoses a %s tombstone query failure without continuing the crawl',
+    async (accessMode) => {
+      primeSyncUpToListing()
+      dbChainMockFns.returning.mockReset()
+      dbChainMockFns.returning.mockResolvedValueOnce([{ ...CONNECTOR, accessMode }])
+      const error = new DrizzleQueryError(
+        'select private SQL',
+        ['private bound content'],
+        Object.assign(new Error('canceling statement due to statement timeout'), { code: '57014' })
+      )
+      dbChainMockFns.limit
+        .mockResolvedValueOnce([CONNECTOR])
+        .mockResolvedValueOnce([{ userId: 'u-1', workspaceId: 'ws-1' }])
+        .mockRejectedValueOnce(error)
+
+      const result = await executeSync('c-1', {
+        billingAttribution: { workspaceId: 'ws-1' } as never,
+      })
+
+      expect(result.error).toBe('Database request failed (SQLSTATE 57014).')
+      expect(mockLogError).toHaveBeenCalledWith('Connector tombstone check failed', {
+        connectorId: 'c-1',
+        operation: 'document.tombstone-check',
+        elapsedMs: expect.any(Number),
+        diagnostic: {
+          category: 'database',
+          code: '57014',
+          databaseReason: 'statement_timeout',
+          message: 'Database request failed (SQLSTATE 57014).',
+        },
+      })
+      expect(mockListDocuments).not.toHaveBeenCalled()
+      expect(JSON.stringify(mockLogError.mock.calls)).not.toContain('private')
+      expect(dbChainMockFns.set).toHaveBeenCalledWith(
+        expect.objectContaining({ status: 'error', consecutiveFailures: 1 })
+      )
+    }
+  )
+
   it.each(['workspace', 'admin'] as const)(
     'uses the locked source mode %s when resolving its token',
     async (accessMode) => {

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

@@ -1050,6 +1050,7 @@ export async function executeSync(
        * that would delete a document we have no positive evidence is actually
        * gone, reintroducing the exact risk this whole design exists to avoid.
        */
+      const tombstoneCheckStartedAt = Date.now()
       const hasTombstonedDocs = await db
         .select({ id: document.id })
         .from(document)
@@ -1065,6 +1066,15 @@ export async function executeSync(
         )
         .limit(1)
         .then((rows) => rows.length > 0)
+        .catch((error: unknown) => {
+          logger.error('Connector tombstone check failed', {
+            connectorId,
+            operation: 'document.tombstone-check',
+            elapsedMs: Date.now() - tombstoneCheckStartedAt,
+            diagnostic: getConnectorFailureDiagnostic(error),
+          })
+          throw error
+        })
 
       /**
        * Determine if this sync should be incremental. A `rehydrate` request forces a

apps/sim/lib/knowledge/documents/document-processing-source.test.ts

@@ -794,6 +794,7 @@ describe('processDocumentAsync write guards', () => {
       diagnostic: {
         category: 'database',
         code: '57014',
+        databaseReason: 'statement_timeout',
         message: 'Database request failed (SQLSTATE 57014).',
       },
     })

apps/sim/lib/workspace-files/search/dispatcher.test.ts

@@ -141,6 +141,28 @@ describe('workspace file search dispatch deadlines', () => {
     })
   })
 
+  it('records the driver cancellation reason and preserves the original failure', async () => {
+    const error = new Error('Failed query\nparams: private-content', {
+      cause: Object.assign(new Error('canceling statement due to statement timeout'), {
+        code: '57014',
+        detail: 'private driver detail',
+      }),
+    })
+    dbChainMockFns.execute.mockResolvedValueOnce([]).mockResolvedValueOnce([{ acquired: true }])
+    dbChainMockFns.onConflictDoNothing.mockRejectedValueOnce(error)
+
+    await expect(dispatchWorkspaceFileSearchIndexJobs()).rejects.toBe(error)
+    expect(mocks.error).toHaveBeenCalledWith('Workspace file search dispatch phase failed', {
+      phase: 'backfill',
+      durationMs: expect.any(Number),
+      code: '57014',
+      databaseReason: 'statement_timeout',
+      error: 'Failed query',
+    })
+    expect(mocks.batchTrigger).not.toHaveBeenCalled()
+    expect(JSON.stringify(mocks.error.mock.calls)).not.toContain('private')
+  })
+
   it.each([false, true])(
     'preserves enqueue failures when claim release fails: %s',
     async (releaseFails) => {

apps/sim/lib/workspace-files/search/dispatcher.ts

@@ -6,7 +6,11 @@ import {
   workspaceFiles,
 } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { getErrorMessage, getPostgresErrorCode } from '@sim/utils/errors'
+import {
+  getErrorMessage,
+  getPostgresCancellationReason,
+  getPostgresErrorCode,
+} from '@sim/utils/errors'
 import { truncate } from '@sim/utils/string'
 import {
   and,
@@ -65,10 +69,12 @@ async function runDispatchPhase<T>(phase: string, operation: () => Promise<T>):
     })
     return result
   } catch (error) {
+    const databaseReason = getPostgresCancellationReason(error)
     logger.error('Workspace file search dispatch phase failed', {
       phase,
       durationMs: Date.now() - startedAt,
       code: getPostgresErrorCode(error),
+      ...(databaseReason ? { databaseReason } : {}),
       error: truncate(getErrorMessage(error).split('\nparams: ')[0], 500),
     })
     throw error