Fix connector sync pause race

Author: TheodoreSpeaks

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

@@ -60,6 +60,7 @@ describe('knowledge connector sync worker', () => {
       connectorId: 'connector-1',
       requestId: 'request-1',
       fullSync: true,
+      requireRunnable: true,
       billingAttribution: BILLING_ATTRIBUTION,
     })
 
@@ -72,6 +73,7 @@ describe('knowledge connector sync worker', () => {
     expect(mockExecuteSync).toHaveBeenCalledWith('connector-1', {
       billingAttribution: BILLING_ATTRIBUTION,
       fullSync: true,
+      requireRunnable: true,
       rehydrate: undefined,
     })
   })
@@ -94,6 +96,7 @@ describe('knowledge connector sync worker', () => {
     expect(mockExecuteSync).toHaveBeenCalledWith('connector-1', {
       billingAttribution: BILLING_ATTRIBUTION,
       fullSync: undefined,
+      requireRunnable: undefined,
       rehydrate: true,
     })
   })

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

@@ -10,13 +10,18 @@ import { CONNECTOR_SYNC_MAX_DURATION_SECONDS } from '@/lib/knowledge/connectors/
 const logger = createLogger('TriggerKnowledgeConnectorSync')
 
 export async function executeConnectorSyncJob(payload: unknown) {
-  const { connectorId, fullSync, rehydrate, requestId, billingAttribution } =
+  const { connectorId, fullSync, requireRunnable, rehydrate, requestId, billingAttribution } =
     assertConnectorSyncPayload(payload)
 
   logger.info(`[${requestId}] Starting connector sync: ${connectorId}`)
 
   try {
-    const result = await executeSync(connectorId, { billingAttribution, fullSync, rehydrate })
+    const result = await executeSync(connectorId, {
+      billingAttribution,
+      fullSync,
+      requireRunnable,
+      rehydrate,
+    })
 
     logger.info(`[${requestId}] Connector sync completed`, {
       connectorId,

apps/sim/lib/knowledge/connectors/queue.test.ts

@@ -21,6 +21,7 @@ vi.mock('@/lib/knowledge/documents/service', () => ({
 }))
 vi.mock('@/lib/knowledge/connectors/sync-engine', () => ({
   executeSync: mockExecuteSync,
+  isConnectorRunnableStatus: (status: string) => status === 'active' || status === 'error',
 }))
 
 import { assertConnectorSyncPayload, dispatchSync } from '@/lib/knowledge/connectors/queue'
@@ -53,6 +54,7 @@ describe('connector sync queue', () => {
     queueTableRows(schemaMock.knowledgeConnector, [
       {
         knowledgeBaseId: 'knowledge-base-1',
+        connectorStatus: 'active',
         connectorArchivedAt: null,
         connectorDeletedAt: null,
         workspaceId: 'workspace-paid',
@@ -80,6 +82,7 @@ describe('connector sync queue', () => {
       {
         connectorId: 'connector-1',
         fullSync: true,
+        requireRunnable: undefined,
         rehydrate: undefined,
         requestId: 'request-1',
         billingAttribution: BILLING_ATTRIBUTION,
@@ -110,6 +113,43 @@ describe('connector sync queue', () => {
     )
   })
 
+  it('carries the runnable requirement into the queued payload', async () => {
+    await dispatchSync('connector-1', {
+      billingAttribution: BILLING_ATTRIBUTION,
+      requireRunnable: true,
+      requestId: 'request-1',
+    })
+
+    expect(mockTrigger).toHaveBeenCalledWith(
+      'knowledge-connector-sync',
+      expect.objectContaining({ connectorId: 'connector-1', requireRunnable: true }),
+      expect.anything()
+    )
+  })
+
+  it('skips automatic dispatch when the connector was paused concurrently', async () => {
+    resetDbChainMock()
+    queueTableRows(schemaMock.knowledgeConnector, [
+      {
+        knowledgeBaseId: 'knowledge-base-1',
+        connectorStatus: 'paused',
+        connectorArchivedAt: null,
+        connectorDeletedAt: null,
+        workspaceId: 'workspace-paid',
+        kbDeletedAt: null,
+      },
+    ])
+
+    await dispatchSync('connector-1', {
+      billingAttribution: BILLING_ATTRIBUTION,
+      requireRunnable: true,
+      requestId: 'request-1',
+    })
+
+    expect(mockTrigger).not.toHaveBeenCalled()
+    expect(mockExecuteSync).not.toHaveBeenCalled()
+  })
+
   it('rejects legacy payloads without billing attribution', () => {
     expect(() =>
       assertConnectorSyncPayload({

apps/sim/lib/knowledge/connectors/queue.ts

@@ -11,14 +11,16 @@ import {
   type BillingAttributionSnapshot,
 } from '@/lib/billing/core/billing-attribution'
 import { resolveTriggerRegion } from '@/lib/core/async-jobs/region'
-import { executeSync } from '@/lib/knowledge/connectors/sync-engine'
+import { executeSync, isConnectorRunnableStatus } from '@/lib/knowledge/connectors/sync-engine'
 import { isTriggerAvailable } from '@/lib/knowledge/documents/service'
 
 const logger = createLogger('ConnectorSyncQueue')
 
 export interface ConnectorSyncPayload {
   connectorId: string
   fullSync?: boolean
+  /** Skip automatic work if the connector is paused or disabled before execution starts. */
+  requireRunnable?: boolean
   /**
    * Force re-hydration + re-indexing of already-synced documents for connectors
    * whose rendered content can drift without a hash change (see
@@ -34,6 +36,7 @@ export interface ConnectorSyncPayload {
 export interface DispatchSyncOptions {
   billingAttribution: BillingAttributionSnapshot
   fullSync?: boolean
+  requireRunnable?: boolean
   rehydrate?: boolean
   requestId?: string
 }
@@ -55,6 +58,9 @@ export function assertConnectorSyncPayload(value: unknown): ConnectorSyncPayload
   if (value.fullSync !== undefined && typeof value.fullSync !== 'boolean') {
     throw new Error('Connector sync payload fullSync must be a boolean when provided')
   }
+  if (value.requireRunnable !== undefined && typeof value.requireRunnable !== 'boolean') {
+    throw new Error('Connector sync payload requireRunnable must be a boolean when provided')
+  }
   if (value.rehydrate !== undefined && typeof value.rehydrate !== 'boolean') {
     throw new Error('Connector sync payload rehydrate must be a boolean when provided')
   }
@@ -65,6 +71,7 @@ export function assertConnectorSyncPayload(value: unknown): ConnectorSyncPayload
   return {
     connectorId: value.connectorId,
     fullSync: value.fullSync as boolean | undefined,
+    requireRunnable: value.requireRunnable as boolean | undefined,
     rehydrate: value.rehydrate as boolean | undefined,
     requestId: value.requestId,
     billingAttribution: assertBillingAttributionSnapshot(value.billingAttribution),
@@ -87,6 +94,7 @@ export async function dispatchSync(
   const payload = assertConnectorSyncPayload({
     connectorId,
     fullSync: options?.fullSync,
+    requireRunnable: options?.requireRunnable,
     rehydrate: options?.rehydrate,
     requestId,
     billingAttribution: options?.billingAttribution,
@@ -95,6 +103,7 @@ export async function dispatchSync(
   const connectorRows = await db
     .select({
       knowledgeBaseId: knowledgeConnector.knowledgeBaseId,
+      connectorStatus: knowledgeConnector.status,
       connectorArchivedAt: knowledgeConnector.archivedAt,
       connectorDeletedAt: knowledgeConnector.deletedAt,
       workspaceId: knowledgeBase.workspaceId,
@@ -134,6 +143,14 @@ export async function dispatchSync(
     })
     return
   }
+  if (payload.requireRunnable && !isConnectorRunnableStatus(row.connectorStatus)) {
+    logger.info('Skipping automatic sync dispatch: connector is not runnable', {
+      connectorId,
+      status: row.connectorStatus,
+      requestId,
+    })
+    return
+  }
   if (!row.workspaceId) {
     throw new Error(`Connector ${connectorId} is missing workspace billing context`)
   }
@@ -161,6 +178,7 @@ export async function dispatchSync(
 
   executeSync(connectorId, {
     fullSync: payload.fullSync,
+    requireRunnable: payload.requireRunnable,
     rehydrate: payload.rehydrate,
     billingAttribution: payload.billingAttribution,
   }).catch((error) => {

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

@@ -7,6 +7,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'
 import {
   classifySuspectListing,
   evaluateListingSafety,
+  isConnectorRunnableStatus,
   isStuckDocumentSweepEligible,
   mergeHydratedDocument,
   type PreviousListingObservation,
@@ -44,6 +45,16 @@ vi.mock('@/connectors/registry.server', () => ({
   },
 }))
 
+describe('isConnectorRunnableStatus', () => {
+  it.each(['active', 'error'])('allows automatic sync from %s', (status) => {
+    expect(isConnectorRunnableStatus(status)).toBe(true)
+  })
+
+  it.each(['paused', 'disabled', 'syncing'])('blocks automatic sync from %s', (status) => {
+    expect(isConnectorRunnableStatus(status)).toBe(false)
+  })
+})
+
 describe('shouldReconcileDeletions', () => {
   it('runs on a clean full listing', async () => {
     const { shouldReconcileDeletions } = await import('@/lib/knowledge/connectors/sync-engine')

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

@@ -91,6 +91,12 @@ const QUEUED_DISPATCH_GRACE_MINUTES = Math.ceil(
 )
 const RETRY_WINDOW_DAYS = 7
 const MAX_CONSECUTIVE_FAILURES = 10
+const RUNNABLE_CONNECTOR_STATUSES = ['active', 'error'] as const
+
+/** Whether an automatic connector sync may begin from this persisted state. */
+export function isConnectorRunnableStatus(status: string): boolean {
+  return RUNNABLE_CONNECTOR_STATUSES.some((runnableStatus) => runnableStatus === status)
+}
 
 /** The processing state the stuck-document sweep decides on, one row at a time. */
 export interface StuckDocumentSweepCandidate {
@@ -777,6 +783,7 @@ export async function executeSync(
   options: {
     billingAttribution: BillingAttributionSnapshot
     fullSync?: boolean
+    requireRunnable?: boolean
     rehydrate?: boolean
   }
 ): Promise<SyncResult> {
@@ -808,6 +815,14 @@ export async function executeSync(
 
   const connector = connectorRows[0]
 
+  if (options.requireRunnable && !isConnectorRunnableStatus(connector.status)) {
+    logger.info('Skipping automatic sync: connector is not runnable', {
+      connectorId,
+      status: connector.status,
+    })
+    return result
+  }
+
   const connectorConfig = CONNECTOR_REGISTRY[connector.connectorType]
   if (!connectorConfig) {
     throw new Error(`Unknown connector type: ${connector.connectorType}`)
@@ -857,7 +872,9 @@ export async function executeSync(
     .where(
       and(
         eq(knowledgeConnector.id, connectorId),
-        ne(knowledgeConnector.status, 'syncing'),
+        options.requireRunnable
+          ? inArray(knowledgeConnector.status, RUNNABLE_CONNECTOR_STATUSES)
+          : ne(knowledgeConnector.status, 'syncing'),
         isNull(knowledgeConnector.archivedAt),
         isNull(knowledgeConnector.deletedAt)
       )

apps/sim/lib/knowledge/orchestration/connectors.test.ts

@@ -274,6 +274,7 @@ describe('performUpdateKnowledgeConnector', () => {
     expect(mockDispatchSync).toHaveBeenCalledWith('conn-1', {
       billingAttribution: BILLING,
       requestId: 'req-1',
+      requireRunnable: true,
     })
   })
 
@@ -361,10 +362,10 @@ describe('performUpdateKnowledgeConnector', () => {
     expect(mockDispatchSync).not.toHaveBeenCalled()
   })
 
-  it('rejects a source replacement that races with synchronization', async () => {
+  it('rejects a source replacement that races with a pause', async () => {
     dbChainMockFns.limit
       .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'active' }])
-      .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'syncing' }])
+      .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'paused' }])
     dbChainMockFns.returning.mockResolvedValueOnce([])
 
     const outcome = await performUpdateKnowledgeConnector({
@@ -379,6 +380,24 @@ describe('performUpdateKnowledgeConnector', () => {
     expect(outcome).toMatchObject({ success: false, errorCode: 'conflict' })
     expect(mockDispatchSync).not.toHaveBeenCalled()
   })
+
+  it('rejects a pause that races with synchronization startup', async () => {
+    dbChainMockFns.limit
+      .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'active' }])
+      .mockResolvedValueOnce([{ id: 'conn-1', connectorType: 'notion', status: 'syncing' }])
+    dbChainMockFns.returning.mockResolvedValueOnce([])
+
+    const outcome = await performUpdateKnowledgeConnector({
+      ...ACTOR,
+      knowledgeBase: KB,
+      connectorId: 'conn-1',
+      updates: { status: 'paused' },
+      resolveBillingAttribution,
+    })
+
+    expect(outcome).toMatchObject({ success: false, errorCode: 'conflict' })
+    expect(mockDispatchSync).not.toHaveBeenCalled()
+  })
 })
 
 describe('performSyncKnowledgeConnector', () => {

apps/sim/lib/knowledge/orchestration/connectors.ts

@@ -9,7 +9,7 @@ import {
 } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { generateId } from '@sim/utils/id'
-import { and, eq, inArray, isNull, ne, sql } from 'drizzle-orm'
+import { and, eq, inArray, isNull, sql } from 'drizzle-orm'
 import { encryptApiKey } from '@/lib/api-key/crypto'
 import type { BillingAttributionSnapshot } from '@/lib/billing/core/billing-attribution'
 import { hasWorkspaceLiveSyncAccess } from '@/lib/billing/core/subscription'
@@ -443,6 +443,9 @@ export async function performUpdateKnowledgeConnector(
       'conflict'
     )
   }
+  if (updates.status !== undefined && existing.status === 'syncing') {
+    return fail('Cannot change connector status while synchronization is in progress', 'conflict')
+  }
 
   if (updates.syncIntervalMinutes !== undefined) {
     if (!kb.workspaceId && updates.syncIntervalMinutes > 0 && updates.syncIntervalMinutes < 60) {
@@ -512,8 +515,8 @@ export async function performUpdateKnowledgeConnector(
       isNull(knowledgeConnector.archivedAt),
       isNull(knowledgeConnector.deletedAt),
     ]
-    if (updates.sourceConfig !== undefined) {
-      updateConditions.push(ne(knowledgeConnector.status, 'syncing'))
+    if (updates.sourceConfig !== undefined || updates.status !== undefined) {
+      updateConditions.push(eq(knowledgeConnector.status, existing.status))
     }
 
     const [row] = await db
@@ -523,14 +526,19 @@ export async function performUpdateKnowledgeConnector(
       .returning()
 
     if (!row) {
-      if (updates.sourceConfig !== undefined) {
+      if (updates.sourceConfig !== undefined || updates.status !== undefined) {
         const current = await getKnowledgeConnector(kb.id, connectorId)
         if (current?.status === 'syncing') {
           return fail(
-            'Cannot update source configuration while connector synchronization is in progress',
+            updates.sourceConfig !== undefined
+              ? 'Cannot update source configuration while connector synchronization is in progress'
+              : 'Cannot change connector status while synchronization is in progress',
             'conflict'
           )
         }
+        if (current) {
+          return fail('Connector status changed during the update; retry the request', 'conflict')
+        }
       }
       return fail('Connector not found', 'not_found')
     }
@@ -564,7 +572,11 @@ export async function performUpdateKnowledgeConnector(
   }
 
   if (dispatchSourceSync && billingAttribution) {
-    dispatchSourceSync(connectorId, { billingAttribution, requestId }).catch((error) => {
+    dispatchSourceSync(connectorId, {
+      billingAttribution,
+      requestId,
+      requireRunnable: true,
+    }).catch((error) => {
       logger.error(
         `[${requestId}] Failed to dispatch source-change sync for connector ${connectorId}`,
         error