fix(connectors): heartbeat every unbounded phase, not just the batch loop

The heartbeat was added where we happened to be looking. The pagination loop —
where a large source spends most of its wall clock, since the batch loop does not
start until every page is fetched — never beat at all, so a long listing on the
uncapped in-process path was still reclaimed as a hard failure. That is the exact
ratchet the heartbeat exists to prevent.

Auditing the remaining phases found something worse in the stuck-document retry:
on the in-process path it handed the entire backlog to a single await that fully
parses, embeds and indexes every document before returning, so no beat placement
could interrupt it. That dispatch is now chunked, with a beat per chunk.

All four call sites share one beatIfDue closure; the two pre-existing inline
blocks were collapsed onto it rather than left as copies.

An await longer than the TTL — one pathological listing page, or a very large
hard delete — is still not covered, and no inline beat can cover it. Closing that
needs a concurrent interval, which is a second mechanism and a separate decision.

Adds the first test that drives executeSync itself, reaching the pagination loop
through the real lock acquisition rather than testing helpers in isolation.

Author: waleedlatif1

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

@@ -8,11 +8,12 @@ import {
   flattenMockConditions,
   hasMockCondition,
   type MockCondition,
+  queueTableRows,
   resetDbChainMock,
   schemaMock,
 } from '@sim/testing'
 import { generateShortId } from '@sim/utils/id'
-import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import {
   classifySuspectListing,
   evaluateListingSafety,
@@ -34,7 +35,14 @@ vi.mock('@/background/knowledge-connector-sync', () => ({
   knowledgeConnectorSync: { trigger: vi.fn() },
 }))
 
-const { mockMapTags } = vi.hoisted(() => ({ mockMapTags: vi.fn() }))
+const { mockMapTags, mockListDocuments } = vi.hoisted(() => ({
+  mockMapTags: vi.fn(),
+  mockListDocuments: vi.fn(),
+}))
+
+vi.mock('@/lib/billing/core/billing-attribution', () => ({
+  assertBillingAttributionSnapshot: (snapshot: unknown) => snapshot,
+}))
 
 vi.mock('@/connectors/registry.server', () => ({
   CONNECTOR_REGISTRY: {
@@ -44,6 +52,11 @@ vi.mock('@/connectors/registry.server', () => ({
     'no-tags': {
       name: 'No Tags',
     },
+    paged: {
+      name: 'Paged',
+      auth: { mode: 'apiKey', optional: true },
+      listDocuments: mockListDocuments,
+    },
   },
 }))
 
@@ -1716,3 +1729,86 @@ describe('heartbeatSyncLock', () => {
     expect(await heartbeatSyncLock('c-1', 'run-a')).toBe(true)
   })
 })
+
+describe('executeSync heartbeats during the listing phase', () => {
+  const CONNECTOR = {
+    id: 'c-1',
+    knowledgeBaseId: 'kb-1',
+    connectorType: 'paged',
+    credentialId: null,
+    encryptedApiKey: null,
+    sourceConfig: {},
+    syncMode: 'full',
+    syncIntervalMinutes: 1440,
+    status: 'active',
+    lastSyncAt: null,
+    lastSyncDocCount: null,
+    consecutiveFailures: 0,
+    syncLockToken: null,
+  }
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+    vi.useFakeTimers()
+    vi.setSystemTime(new Date('2026-08-20T00:00:00.000Z'))
+  })
+
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  /** Drives executeSync as far as the pagination loop. */
+  function primeSyncUpToListing() {
+    queueTableRows(schemaMock.knowledgeConnector, [CONNECTOR])
+    queueTableRows(schemaMock.knowledgeBase, [{ userId: 'u-1', workspaceId: 'ws-1' }])
+    // The lock CAS; every later `.returning()` falls through to the empty default,
+    // which is what makes the heartbeat below report a lost lock.
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'c-1' }])
+  }
+
+  it('beats between pages and abandons the run when the lock was reclaimed', async () => {
+    const { executeSync } = await import('@/lib/knowledge/connectors/sync-engine')
+    const { SYNC_LOCK_HEARTBEAT_INTERVAL_MS } = await import(
+      '@/lib/knowledge/connectors/sync-limits'
+    )
+
+    primeSyncUpToListing()
+
+    /**
+     * Listing is where a large source spends most of its wall clock, so a page
+     * that pushes the run past the heartbeat interval must trigger a beat before
+     * the next page — not only once listing has finished.
+     */
+    mockListDocuments.mockImplementation(async () => {
+      vi.setSystemTime(new Date(Date.now() + SYNC_LOCK_HEARTBEAT_INTERVAL_MS + 1_000))
+      return { documents: [], hasMore: true, nextCursor: 'page-2' }
+    })
+
+    const result = await executeSync('c-1', {
+      billingAttribution: { workspaceId: 'ws-1' } as never,
+    })
+
+    // Aborted on the beat before page 2 rather than paging on under a lost lock.
+    expect(mockListDocuments).toHaveBeenCalledTimes(1)
+    expect(result.error).toBe('sync_superseded')
+  })
+
+  it('does not beat when pages return faster than the interval', async () => {
+    const { executeSync } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    primeSyncUpToListing()
+
+    let pages = 0
+    mockListDocuments.mockImplementation(async () => {
+      pages += 1
+      vi.setSystemTime(new Date(Date.now() + 1_000))
+      return { documents: [], hasMore: pages < 3, nextCursor: `page-${pages}` }
+    })
+
+    await executeSync('c-1', { billingAttribution: { workspaceId: 'ws-1' } as never })
+
+    // All three pages fetched: the time gate keeps a fast listing beat-free.
+    expect(mockListDocuments).toHaveBeenCalledTimes(3)
+  })
+})

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

@@ -76,6 +76,15 @@ const DEFAULT_OP_SIZE_BYTES = 4 * 1024 * 1024
 const CONTENT_INFLIGHT_BUDGET_BYTES = 64 * 1024 * 1024
 const MAX_PAGES = 500
 const MAX_SAFE_TITLE_LENGTH = 200
+/**
+ * How many stuck documents are re-dispatched per call.
+ *
+ * The retry backlog is unbounded, and on the in-process fallback path
+ * `processDocumentsWithQueue` parses, embeds, and indexes every document it is
+ * given before returning. Handing it the whole backlog made the retry a single
+ * await no heartbeat could interrupt; chunking gives the beat somewhere to run.
+ */
+const STUCK_RETRY_DISPATCH_CHUNK_SIZE = 25
 const STALE_PROCESSING_MINUTES = 45
 /** Largest connector corpus observed in production, which sets the queue drain to beat. */
 const LARGEST_OBSERVED_CORPUS_DOCUMENTS = 7_730
@@ -1309,6 +1318,20 @@ export async function executeSync(
   const syncStartedAt = new Date()
   /** Seeded at lock acquisition, which wrote `updatedAt` itself. */
   let lastHeartbeatAtMs = Date.now()
+
+  /**
+   * Refreshes the lock if the interval has elapsed, and aborts the run if it has
+   * been reclaimed. Called at the top of every unbounded loop in this sync — the
+   * time gate makes each call nearly free, so placement only has to guarantee
+   * that no unbounded phase runs without reaching one.
+   */
+  const beatIfDue = async (): Promise<void> => {
+    if (!shouldHeartbeatSyncLock(Date.now(), lastHeartbeatAtMs)) return
+    if (!(await heartbeatSyncLock(connectorId, syncLogId))) {
+      throw new SyncLockLostException(connectorId)
+    }
+    lastHeartbeatAtMs = Date.now()
+  }
   await db.insert(knowledgeConnectorSyncLog).values({
     id: syncLogId,
     connectorId,
@@ -1416,6 +1439,14 @@ export async function executeSync(
     )
 
     for (let pageNum = 0; hasMore && pageNum < MAX_PAGES; pageNum++) {
+      /**
+       * Listing is where a large source spends most of its wall clock — the
+       * batch loop below does not start until every page has been fetched — so
+       * without this a big listing outran the TTL and was reclaimed as a hard
+       * failure, which is the exact ratchet the heartbeat exists to prevent.
+       */
+      await beatIfDue()
+
       if (pageNum > 0 && connectorConfig.auth.mode === 'oauth') {
         accessToken = await resolveAccessToken(connector, connectorConfig, credentialUserId)
       }
@@ -1593,12 +1624,7 @@ export async function executeSync(
     // per-file cap never hydrate/upload together and exhaust the worker heap.
     const batches = chunkOpsByByteBudget(pendingOps, CONTENT_INFLIGHT_BUDGET_BYTES, SYNC_BATCH_SIZE)
     for (const rawBatch of batches) {
-      if (shouldHeartbeatSyncLock(Date.now(), lastHeartbeatAtMs)) {
-        if (!(await heartbeatSyncLock(connectorId, syncLogId))) {
-          throw new SyncLockLostException(connectorId)
-        }
-        lastHeartbeatAtMs = Date.now()
-      }
+      await beatIfDue()
 
       const liveness = await checkSyncLiveness(connectorId, connector.knowledgeBaseId)
       if (liveness.connectorDeleted) {
@@ -1980,12 +2006,7 @@ export async function executeSync(
       result.docsDeleted += await hardDeleteDocuments(safeHardDeleteIds, syncLogId, connectorId)
     }
 
-    if (shouldHeartbeatSyncLock(Date.now(), lastHeartbeatAtMs)) {
-      if (!(await heartbeatSyncLock(connectorId, syncLogId))) {
-        throw new SyncLockLostException(connectorId)
-      }
-      lastHeartbeatAtMs = Date.now()
-    }
+    await beatIfDue()
 
     const postBatchLiveness = await checkSyncLiveness(connectorId, connector.knowledgeBaseId)
     if (postBatchLiveness.connectorDeleted) {
@@ -2098,9 +2119,11 @@ export async function executeSync(
           }
         })
 
-        if (retryDocs.length > 0) {
+        for (let i = 0; i < retryDocs.length; i += STUCK_RETRY_DISPATCH_CHUNK_SIZE) {
+          await beatIfDue()
+
           await processDocumentsWithQueue(
-            retryDocs.map((doc) => ({
+            retryDocs.slice(i, i + STUCK_RETRY_DISPATCH_CHUNK_SIZE).map((doc) => ({
               documentId: doc.id,
               filename: doc.filename ?? 'document.txt',
               fileUrl: doc.fileUrl ?? '',