fix(knowledge): harden connector sync and document processing

Author: waleedlatif1

apps/docs/openapi-v2-knowledge.json

@@ -3754,6 +3754,13 @@
             "maximum": 9007199254740991,
             "description": "Documents unchanged."
           },
+          "docsSkipped": {
+            "default": 0,
+            "description": "Documents intentionally skipped because they could not be indexed safely.",
+            "type": "integer",
+            "minimum": 0,
+            "maximum": 9007199254740991
+          },
           "docsFailed": {
             "type": "integer",
             "minimum": 0,
@@ -3782,6 +3789,7 @@
           "docsUpdated",
           "docsDeleted",
           "docsUnchanged",
+          "docsSkipped",
           "docsFailed",
           "errorMessage"
         ],

apps/sim/app/api/v2/knowledge/connector-utils.ts

@@ -44,6 +44,7 @@ interface KnowledgeConnectorSyncLogProjection {
   docsUpdated: number
   docsDeleted: number
   docsUnchanged: number
+  docsSkipped: number
   docsFailed: number
   errorMessage: string | null
 }

apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/connectors-section/connectors-section.test.tsx

@@ -136,6 +136,7 @@ function makeLog(overrides: Partial<SyncLogData> & Pick<SyncLogData, 'status'>):
     docsUpdated: 0,
     docsDeleted: 0,
     docsUnchanged: 0,
+    docsSkipped: 0,
     docsFailed: 0,
     errorMessage: null,
     ...overrides,

apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/connectors-section/connectors-section.tsx

@@ -716,7 +716,11 @@ export function SyncHistory({ logs, isLoading }: SyncHistoryProps) {
       {logs.map((log) => {
         const state = getSyncLogState(log, now)
         const totalChanges =
-          log.docsAdded + log.docsUpdated + log.docsDeleted + (log.docsFailed ?? 0)
+          log.docsAdded +
+          log.docsUpdated +
+          log.docsDeleted +
+          log.docsSkipped +
+          (log.docsFailed ?? 0)
 
         return (
           <div key={log.id} className='flex items-start gap-2 rounded-md px-2 py-1.5 text-xs'>
@@ -763,6 +767,16 @@ export function SyncHistory({ logs, isLoading }: SyncHistoryProps) {
                             <span className='text-[var(--text-error)]'>!{log.docsFailed}</span>
                           </>
                         )}
+                        {log.docsSkipped > 0 && (
+                          <>
+                            {(log.docsAdded > 0 ||
+                              log.docsUpdated > 0 ||
+                              log.docsDeleted > 0 ||
+                              log.docsFailed > 0) &&
+                              ' '}
+                            <span className='text-[var(--caution)]'>⊘{log.docsSkipped}</span>
+                          </>
+                        )}
                       </>
                     ) : (
                       'No changes'

apps/sim/background/knowledge-connector-sync.test.ts

@@ -9,15 +9,21 @@ const { mockAssertConnectorSyncPayload, mockExecuteSync, mockTask } = vi.hoisted
   mockTask: vi.fn((config) => config),
 }))
 
-vi.mock('@trigger.dev/sdk', () => ({ task: mockTask }))
+vi.mock('@trigger.dev/sdk', () => ({
+  task: mockTask,
+  AbortTaskRunError: class AbortTaskRunError extends Error {},
+}))
 vi.mock('@/lib/knowledge/connectors/queue', () => ({
   assertConnectorSyncPayload: mockAssertConnectorSyncPayload,
 }))
 vi.mock('@/lib/knowledge/connectors/sync-engine', () => ({
   executeSync: mockExecuteSync,
 }))
 
-import { executeConnectorSyncJob } from '@/background/knowledge-connector-sync'
+import {
+  classifyConnectorSyncResult,
+  executeConnectorSyncJob,
+} from '@/background/knowledge-connector-sync'
 
 const BILLING_ATTRIBUTION = {
   actorUserId: 'external-admin',
@@ -40,7 +46,9 @@ describe('knowledge connector sync worker', () => {
       docsUpdated: 0,
       docsDeleted: 0,
       docsUnchanged: 0,
+      docsSkipped: 0,
       docsFailed: 0,
+      processingDispatch: { requested: 0, accepted: 0, failed: 0 },
     })
   })
 
@@ -103,4 +111,148 @@ describe('knowledge connector sync worker', () => {
       dispatchToken: undefined,
     })
   })
+
+  it('fails visibly without retrying an already-persisted partial sync', async () => {
+    mockAssertConnectorSyncPayload.mockReturnValue({
+      connectorId: 'connector-1',
+      requestId: 'request-1',
+      billingAttribution: BILLING_ATTRIBUTION,
+    })
+    mockExecuteSync.mockResolvedValue({
+      docsAdded: 2,
+      docsUpdated: 0,
+      docsDeleted: 0,
+      docsUnchanged: 0,
+      docsSkipped: 3,
+      docsFailed: 1,
+      processingDispatch: { requested: 2, accepted: 1, failed: 1 },
+    })
+
+    await expect(
+      executeConnectorSyncJob({
+        connectorId: 'connector-1',
+        requestId: 'request-1',
+        billingAttribution: BILLING_ATTRIBUTION,
+      })
+    ).rejects.toThrow('Connector sync partially failed')
+  })
+
+  it('does not turn intentionally skipped source files into a task failure', () => {
+    expect(
+      classifyConnectorSyncResult({
+        docsAdded: 0,
+        docsUpdated: 0,
+        docsDeleted: 0,
+        docsUnchanged: 0,
+        docsSkipped: 4,
+        docsFailed: 0,
+        processingDispatch: { requested: 0, accepted: 0, failed: 0 },
+      })
+    ).toBe('completed')
+  })
+
+  it('reports superseded and non-runnable jobs as skipped control flow', () => {
+    const baseResult = {
+      docsAdded: 0,
+      docsUpdated: 0,
+      docsDeleted: 0,
+      docsUnchanged: 0,
+      docsSkipped: 0,
+      docsFailed: 0,
+      processingDispatch: { requested: 0, accepted: 0, failed: 0 },
+    }
+
+    expect(classifyConnectorSyncResult({ ...baseResult, skipReason: 'sync_superseded' })).toBe(
+      'skipped'
+    )
+    expect(
+      classifyConnectorSyncResult({ ...baseResult, skipReason: 'connector_not_syncable' })
+    ).toBe('skipped')
+  })
+
+  it('never treats a provider error that collides with a skip reason as control flow', () => {
+    expect(
+      classifyConnectorSyncResult({
+        docsAdded: 0,
+        docsUpdated: 0,
+        docsDeleted: 0,
+        docsUnchanged: 0,
+        docsSkipped: 0,
+        docsFailed: 0,
+        processingDispatch: { requested: 0, accepted: 0, failed: 0 },
+        error: 'sync_in_progress',
+      })
+    ).toBe('failed')
+  })
+
+  it('classifies a persisted connector error as a failed task', () => {
+    expect(
+      classifyConnectorSyncResult({
+        docsAdded: 0,
+        docsUpdated: 0,
+        docsDeleted: 0,
+        docsUnchanged: 0,
+        docsSkipped: 0,
+        docsFailed: 0,
+        processingDispatch: { requested: 0, accepted: 0, failed: 0 },
+        error: 'provider unavailable',
+      })
+    ).toBe('failed')
+  })
+
+  it('fails the Trigger run for a persisted connector error without a whole-sync retry', async () => {
+    mockAssertConnectorSyncPayload.mockReturnValue({
+      connectorId: 'connector-1',
+      requestId: 'request-1',
+      billingAttribution: BILLING_ATTRIBUTION,
+    })
+    mockExecuteSync.mockResolvedValue({
+      docsAdded: 0,
+      docsUpdated: 0,
+      docsDeleted: 0,
+      docsUnchanged: 0,
+      docsSkipped: 0,
+      docsFailed: 0,
+      processingDispatch: { requested: 0, accepted: 0, failed: 0 },
+      error: 'provider unavailable',
+    })
+
+    await expect(
+      executeConnectorSyncJob({
+        connectorId: 'connector-1',
+        requestId: 'request-1',
+        billingAttribution: BILLING_ATTRIBUTION,
+      })
+    ).rejects.toThrow('Connector sync failed for connector-1: provider unavailable')
+  })
+
+  it('returns an explicit skipped outcome for a superseded task', async () => {
+    mockAssertConnectorSyncPayload.mockReturnValue({
+      connectorId: 'connector-1',
+      requestId: 'request-1',
+      billingAttribution: BILLING_ATTRIBUTION,
+    })
+    mockExecuteSync.mockResolvedValue({
+      docsAdded: 0,
+      docsUpdated: 0,
+      docsDeleted: 0,
+      docsUnchanged: 0,
+      docsSkipped: 0,
+      docsFailed: 0,
+      processingDispatch: { requested: 0, accepted: 0, failed: 0 },
+      skipReason: 'dispatch_superseded',
+    })
+
+    await expect(
+      executeConnectorSyncJob({
+        connectorId: 'connector-1',
+        requestId: 'request-1',
+        billingAttribution: BILLING_ATTRIBUTION,
+      })
+    ).resolves.toMatchObject({
+      success: false,
+      outcome: 'skipped',
+      skipReason: 'dispatch_superseded',
+    })
+  })
 })

apps/sim/background/knowledge-connector-sync.ts

@@ -1,14 +1,40 @@
 import { createLogger } from '@sim/logger'
-import { task } from '@trigger.dev/sdk'
+import { AbortTaskRunError, task } from '@trigger.dev/sdk'
 import {
   assertConnectorSyncPayload,
   type ConnectorSyncPayload,
 } from '@/lib/knowledge/connectors/queue'
 import { executeSync } from '@/lib/knowledge/connectors/sync-engine'
 import { CONNECTOR_SYNC_MAX_DURATION_SECONDS } from '@/lib/knowledge/connectors/sync-limits'
+import type { SyncResult } from '@/connectors/types'
 
 const logger = createLogger('TriggerKnowledgeConnectorSync')
 
+export type ConnectorSyncTaskOutcome = 'completed' | 'partial' | 'skipped' | 'failed'
+
+/**
+ * Separates source-sync failures from expected queue/lock no-ops. Intentional
+ * source skips do not make a run partial; actual hydration, persistence, or
+ * processing-dispatch failures do.
+ */
+export function classifyConnectorSyncResult(result: SyncResult): ConnectorSyncTaskOutcome {
+  if (result.skipReason) return 'skipped'
+  if (result.error) return 'failed'
+  if (result.docsFailed > 0 || result.processingDispatch.failed > 0) return 'partial'
+  return 'completed'
+}
+
+function formatConnectorSyncFailure(
+  connectorId: string,
+  result: SyncResult,
+  outcome: Extract<ConnectorSyncTaskOutcome, 'partial' | 'failed'>
+): string {
+  if (outcome === 'failed') {
+    return `Connector sync failed for ${connectorId}: ${result.error}`
+  }
+  return `Connector sync partially failed for ${connectorId}: ${result.docsFailed} source failures, ${result.processingDispatch.failed} dispatch failures`
+}
+
 export async function executeConnectorSyncJob(payload: unknown) {
   const {
     connectorId,
@@ -33,15 +59,33 @@ export async function executeConnectorSyncJob(payload: unknown) {
 
     logger.info(`[${requestId}] Connector sync completed`, {
       connectorId,
+      outcome: classifyConnectorSyncResult(result),
       added: result.docsAdded,
       updated: result.docsUpdated,
       deleted: result.docsDeleted,
       unchanged: result.docsUnchanged,
+      skipped: result.docsSkipped,
       failed: result.docsFailed,
+      processingRequested: result.processingDispatch.requested,
+      processingAccepted: result.processingDispatch.accepted,
+      processingDispatchFailed: result.processingDispatch.failed,
     })
 
+    const outcome = classifyConnectorSyncResult(result)
+    if (outcome === 'failed' || outcome === 'partial') {
+      /**
+       * `executeSync` has already persisted its terminal state. Source failures
+       * preserve the previous incremental watermark so the next connector pass
+       * replays them; dispatch failures remain eligible for the stuck-document
+       * sweep. Retrying this whole task immediately would duplicate a large
+       * fan-out, so fail visibly without retrying the completed transaction.
+       */
+      throw new AbortTaskRunError(formatConnectorSyncFailure(connectorId, result, outcome))
+    }
+
     return {
-      success: !result.error,
+      success: outcome === 'completed',
+      outcome,
       connectorId,
       ...result,
     }