fix(knowledge): make connector sync recovery durable

Author: TheodoreSpeaks

apps/sim/app/api/knowledge/connectors/sync/route.ts

@@ -102,7 +102,11 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
           throw new Error(`Connector ${connector.id} is missing workspace billing context`)
         }
         const billingAttribution = await resolveSystemBillingAttribution(connector.workspaceId)
-        await dispatchSync(connector.id, { billingAttribution, requestId })
+        await dispatchSync(connector.id, {
+          billingAttribution,
+          requestId,
+          requireRunnable: true,
+        })
       } catch (error) {
         logger.error(`[${requestId}] Failed to dispatch sync for connector ${connector.id}`, error)
       }

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

@@ -882,7 +882,12 @@ export async function executeSync(
     .returning({ id: knowledgeConnector.id })
 
   if (lockResult.length === 0) {
-    logger.info('Sync already in progress, skipping', { connectorId })
+    logger.info(
+      options.requireRunnable
+        ? 'Connector is not runnable or sync is already in progress, skipping'
+        : 'Sync already in progress, skipping',
+      { connectorId }
+    )
     return result
   }
 

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

@@ -254,10 +254,20 @@ describe('performUpdateKnowledgeConnector', () => {
 
   it('queues synchronization after replacing an active connector source', async () => {
     dbChainMockFns.limit.mockResolvedValueOnce([
-      { id: 'conn-1', connectorType: 'notion', status: 'active' },
+      {
+        id: 'conn-1',
+        connectorType: 'notion',
+        status: 'active',
+        syncIntervalMinutes: 0,
+      },
     ])
     dbChainMockFns.returning.mockResolvedValueOnce([
-      { id: 'conn-1', connectorType: 'notion', status: 'active' },
+      {
+        id: 'conn-1',
+        connectorType: 'notion',
+        status: 'active',
+        syncIntervalMinutes: 0,
+      },
     ])
 
     const outcome = await performUpdateKnowledgeConnector({
@@ -278,12 +288,22 @@ describe('performUpdateKnowledgeConnector', () => {
     })
   })
 
-  it('reports a queue failure after replacing an active connector source', async () => {
+  it('reports a queue failure and leaves the source sync due for retry', async () => {
     dbChainMockFns.limit.mockResolvedValueOnce([
-      { id: 'conn-1', connectorType: 'notion', status: 'active' },
+      {
+        id: 'conn-1',
+        connectorType: 'notion',
+        status: 'active',
+        syncIntervalMinutes: 0,
+      },
     ])
     dbChainMockFns.returning.mockResolvedValueOnce([
-      { id: 'conn-1', connectorType: 'notion', status: 'active' },
+      {
+        id: 'conn-1',
+        connectorType: 'notion',
+        status: 'active',
+        syncIntervalMinutes: 0,
+      },
     ])
     mockDispatchSync.mockRejectedValueOnce(new Error('queue unavailable'))
 
@@ -302,6 +322,12 @@ describe('performUpdateKnowledgeConnector', () => {
       error: 'queue unavailable',
     })
     expect(dbChainMockFns.update).toHaveBeenCalledOnce()
+    expect(dbChainMockFns.set).toHaveBeenCalledWith(
+      expect.objectContaining({
+        sourceConfig: { database: 'next' },
+        nextSyncAt: expect.any(Date),
+      })
+    )
     expect(mockDispatchSync).toHaveBeenCalledOnce()
   })
 

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

@@ -483,7 +483,10 @@ export async function performUpdateKnowledgeConnector(
     }
   }
 
-  const values: Partial<typeof knowledgeConnector.$inferInsert> = { updatedAt: new Date() }
+  const updateTimestamp = new Date()
+  const values: Partial<typeof knowledgeConnector.$inferInsert> = {
+    updatedAt: updateTimestamp,
+  }
   if (updates.sourceConfig !== undefined) {
     values.sourceConfig = updates.sourceConfig
   }
@@ -506,6 +509,9 @@ export async function performUpdateKnowledgeConnector(
       }
     }
   }
+  if (shouldDispatchSourceSync) {
+    values.nextSyncAt = updateTimestamp
+  }
 
   let updated: ConnectorRow
   try {