fix(connectors): heartbeat a running sync and cap deletion generations apart

Review round 4 on #6909. Two regressions the guards introduced, plus two tests
that could not fail.

Counting a stale-lock reclaim as a failure turned the reaper into a one-way
ratchet for any sync that legitimately outran the TTL: the in-process path has
no duration cap, so a long self-hosted sync was reclaimed, its successful
terminal write then failed the ownership guard and was discarded, and its
failure counter never reset. Ten of those and a working connector was disabled
telling the user to reconnect.

A running sync now refreshes updatedAt every five minutes, so the reaper's
staleness predicate means "nobody is working on this" rather than "this started
a long time ago". The beat is guarded on the run's own lock, so it doubles as an
ownership probe: a run whose lock was reclaimed abandons immediately instead of
working for hours and then discarding the result.

The deletion cap summed soft and hard deletes against one ceiling sized for a
single generation, so a connector with steady churn deadlocked from its second
sync onward and got monotonically worse — the all-or-nothing hold blocked the
very hard deletes that would have drained the tombstone backlog. Hard deletes
are confirmations of removals already gated when they were soft-deleted, so each
generation now caps independently.

Both guard tests for the reaper asserted only the bookends of the rendered SQL,
leaving the comparison itself unasserted: an inverted threshold that disabled a
connector on its first hard kill passed. Both now assert the whole expression.

Author: waleedlatif1

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

@@ -106,11 +106,18 @@ describe('connector sync scheduler stale-lock reaper', () => {
     await runTickRecovering(['connector-1'])
 
     const status = setPayloadForUpdate(0).status
-    const rendered = renderedSql(status)
 
-    expect(rendered).toContain('CASE WHEN COALESCE(')
-    expect(rendered).toContain("THEN 'disabled' ELSE 'error' END")
-    expect(numericBinds(status)).toContain(MAX_CONSECUTIVE_FAILURES)
+    /**
+     * Asserted whole rather than by its bookends: the comparison is the entire
+     * point of this expression, and leaving it in an un-asserted middle let
+     * `+ 2 >=`, `+ 1 >`, and an inverted `+ 1 <=` all pass. The last of those
+     * disables a connector on its first hard kill.
+     */
+    expect(renderedSql(status)).toBe(
+      "CASE WHEN COALESCE(?, 0) + 1 >= ? THEN 'disabled' ELSE 'error' END"
+    )
+    expect(asFragment(status).values[0]).toBe(schemaMock.knowledgeConnector.consecutiveFailures)
+    expect(asFragment(status).values[1]).toBe(MAX_CONSECUTIVE_FAILURES)
   })
 
   it('derives nextSyncAt from the shared failure backoff ladder', async () => {
@@ -119,9 +126,10 @@ describe('connector sync scheduler stale-lock reaper', () => {
     const nextSyncAt = setPayloadForUpdate(0).nextSyncAt
     const rendered = renderedSql(nextSyncAt)
 
-    expect(rendered).toContain('THEN NULL')
-    expect(rendered).toContain('LEAST(')
-    expect(rendered).toContain("INTERVAL '1 minute'")
+    expect(rendered).toBe(
+      'CASE WHEN COALESCE(?, 0) + 1 >= ? THEN NULL ' +
+        "ELSE now() + LEAST((COALESCE(?, 0) + 1) * ?, ?) * INTERVAL '1 minute' END"
+    )
 
     const [threshold, step, cap] = numericBinds(nextSyncAt)
     expect(threshold).toBe(MAX_CONSECUTIVE_FAILURES)
@@ -190,12 +198,32 @@ describe('connector sync scheduler stale-lock reaper', () => {
     await runTickRecovering(['connector-1'])
 
     const where = dbChainMockFns.where.mock.calls[1][0]
+
+    /**
+     * Checks every position, not just `column`. `eq()` builds `{left, right}`
+     * and only `inArray()` builds `{column}`, so a `column`-only assertion
+     * silently permitted an `eq`-scoped sweep — the exact coupling this test
+     * exists to forbid.
+     */
+    const connectorIdColumn = schemaMock.knowledgeConnectorSyncLog.connectorId
     expect(
       hasMockCondition(
         where,
-        (node: MockCondition) => node.column === schemaMock.knowledgeConnectorSyncLog.connectorId
+        (node: MockCondition) =>
+          node.column === connectorIdColumn ||
+          node.left === connectorIdColumn ||
+          node.right === connectorIdColumn
       )
     ).toBe(false)
+
+    // And positively: the sweep is keyed on the row's own age.
+    expect(
+      hasMockCondition(
+        where,
+        (node: MockCondition) =>
+          node.type === 'lte' && node.left === schemaMock.knowledgeConnectorSyncLog.startedAt
+      )
+    ).toBe(true)
   })
 
   it('drives the connector write off a single clock', async () => {

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

@@ -751,7 +751,7 @@ describe('capReconciliationDeletions', () => {
 
     expect(result.held).toBe(false)
     expect(result.cap).toBe(250)
-    expect(result.requested).toBe(250)
+    expect(result.withheld).toBe(0)
     expect(result.softDeleteIds).toEqual(soft)
   })
 
@@ -761,7 +761,8 @@ describe('capReconciliationDeletions', () => {
     const result = capReconciliationDeletions(ids('soft', 251), [], 1000, false)
 
     expect(result.held).toBe(true)
-    expect(result.requested).toBe(251)
+    expect(result.softHeld).toBe(true)
+    expect(result.withheld).toBe(251)
   })
 
   it('returns empty arrays — not the inputs — when held', async () => {
@@ -774,25 +775,55 @@ describe('capReconciliationDeletions', () => {
     expect(result.hardDeleteIds).toEqual([])
   })
 
-  it('counts the union of soft and hard deletions against one cap', async () => {
+  it('caps each generation separately rather than summing them', async () => {
     const { capReconciliationDeletions } = await import('@/lib/knowledge/connectors/sync-engine')
 
-    const overlapping = ids('doc', 200)
-    // Same ids on both lists must count once, not twice.
-    expect(capReconciliationDeletions(overlapping, overlapping, 1000, false).held).toBe(false)
-    expect(capReconciliationDeletions(ids('a', 200), ids('b', 200), 1000, false).held).toBe(true)
+    /**
+     * Hard deletes are the previous generation's soft deletes, already gated by
+     * this cap once. Summing them double-counts the older generation, which is
+     * what deadlocked a churning connector.
+     */
+    const result = capReconciliationDeletions(ids('a', 200), ids('b', 200), 1000, false)
+
+    expect(result.held).toBe(false)
+    expect(result.softDeleteIds).toHaveLength(200)
+    expect(result.hardDeleteIds).toHaveLength(200)
   })
 
-  it('is bypassed by a forced fullSync', async () => {
+  it('holds only the generation that breached the cap', async () => {
     const { capReconciliationDeletions } = await import('@/lib/knowledge/connectors/sync-engine')
 
-    const hard = ids('hard', 1000)
-    const result = capReconciliationDeletions([], hard, 1000, true)
+    const hard = ids('hard', 100)
+    const result = capReconciliationDeletions(ids('soft', 400), hard, 1000, false)
 
-    expect(result.held).toBe(false)
+    expect(result.softHeld).toBe(true)
+    expect(result.hardHeld).toBe(false)
+    expect(result.softDeleteIds).toEqual([])
+    // The confirmed generation still drains, so the backlog cannot ratchet.
     expect(result.hardDeleteIds).toEqual(hard)
   })
 
+  it('is bypassed by a forced fullSync, in both generations', async () => {
+    const { capReconciliationDeletions } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    const hard = ids('hard', 1000)
+    const hardOnly = capReconciliationDeletions([], hard, 1000, true)
+
+    expect(hardOnly.held).toBe(false)
+    expect(hardOnly.hardDeleteIds).toEqual(hard)
+
+    /**
+     * Exercised per generation: asserting only the hard list left the soft
+     * branch's bypass untested, so dropping it there was invisible.
+     */
+    const soft = ids('soft', 1000)
+    const softOnly = capReconciliationDeletions(soft, [], 1000, true)
+
+    expect(softOnly.held).toBe(false)
+    expect(softOnly.softHeld).toBe(false)
+    expect(softOnly.softDeleteIds).toEqual(soft)
+  })
+
   it('applies the small-corpus floor rather than the ratio', async () => {
     const { capReconciliationDeletions } = await import('@/lib/knowledge/connectors/sync-engine')
 
@@ -814,6 +845,28 @@ describe('capReconciliationDeletions', () => {
     ).toBe(true)
   })
 
+  describe('steady churn', () => {
+    it('reaches a stable state instead of ratcheting shut', async () => {
+      const { capReconciliationDeletions } = await import('@/lib/knowledge/connectors/sync-engine')
+
+      /**
+       * 1,000 documents at 15% churn against a cap of 250. Under one summed cap:
+       * sync 1 applied 150 soft; sync 2 requested 150 soft + 150 hard = 300 and
+       * was held in full; the blocked hard deletes then accumulated forever.
+       */
+      const sync1 = capReconciliationDeletions(ids('gen1', 150), [], 1000, false)
+      expect(sync1.held).toBe(false)
+
+      const sync2 = capReconciliationDeletions(ids('gen2', 150), ids('gen1', 150), 1000, false)
+      expect(sync2.held).toBe(false)
+      expect(sync2.hardDeleteIds).toHaveLength(150)
+
+      const sync3 = capReconciliationDeletions(ids('gen3', 150), ids('gen2', 150), 1000, false)
+      expect(sync3.held).toBe(false)
+      expect(sync3.hardDeleteIds).toHaveLength(150)
+    })
+  })
+
   describe('confirmed data-loss shapes', () => {
     it('holds a partial outage that returns half a 1000-document corpus', async () => {
       const { capReconciliationDeletions } = await import('@/lib/knowledge/connectors/sync-engine')
@@ -1341,3 +1394,72 @@ describe('buildSyncLockAcquisition', () => {
     expect(acquisition.status).toBe('syncing')
   })
 })
+
+describe('shouldHeartbeatSyncLock', () => {
+  it('beats once the interval has elapsed', async () => {
+    const { shouldHeartbeatSyncLock } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    expect(shouldHeartbeatSyncLock(1_000, 0, 1_000)).toBe(true)
+    expect(shouldHeartbeatSyncLock(1_001, 0, 1_000)).toBe(true)
+  })
+
+  it('does not beat before the interval has elapsed', async () => {
+    const { shouldHeartbeatSyncLock } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    expect(shouldHeartbeatSyncLock(999, 0, 1_000)).toBe(false)
+    expect(shouldHeartbeatSyncLock(0, 0, 1_000)).toBe(false)
+  })
+
+  it('defaults to an interval far below the reclaim TTL', async () => {
+    const { shouldHeartbeatSyncLock } = await import('@/lib/knowledge/connectors/sync-engine')
+    const { CONNECTOR_SYNC_STALE_LOCK_TTL_MS, SYNC_LOCK_HEARTBEAT_INTERVAL_MS } = await import(
+      '@/lib/knowledge/connectors/sync-limits'
+    )
+
+    /**
+     * A live run must beat many times over before the reclaim cutoff, or
+     * ordinary jitter reclaims a working sync — which is what made the reaper a
+     * one-way ratchet to `disabled` for slow in-process syncs.
+     */
+    expect(SYNC_LOCK_HEARTBEAT_INTERVAL_MS * 4).toBeLessThan(CONNECTOR_SYNC_STALE_LOCK_TTL_MS)
+    expect(shouldHeartbeatSyncLock(SYNC_LOCK_HEARTBEAT_INTERVAL_MS, 0)).toBe(true)
+    expect(shouldHeartbeatSyncLock(SYNC_LOCK_HEARTBEAT_INTERVAL_MS - 1, 0)).toBe(false)
+  })
+})
+
+describe('heartbeatSyncLock', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  it('refreshes updatedAt under the run own lock guard', async () => {
+    const { heartbeatSyncLock } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    await heartbeatSyncLock('c-1', 'run-a')
+
+    expect(dbChainMockFns.set.mock.calls[0][0]).toEqual({ updatedAt: expect.any(Date) })
+
+    // Guarded, so a beat doubles as an ownership probe rather than a blind touch.
+    const where = dbChainMockFns.where.mock.calls[0][0]
+    expect(
+      hasMockCondition(
+        where,
+        (node: MockCondition) =>
+          node.type === 'eq' &&
+          node.left === schemaMock.knowledgeConnector.syncLockToken &&
+          node.right === 'run-a'
+      )
+    ).toBe(true)
+  })
+
+  it('reports a lost lock so the run can stop instead of racing its replacement', async () => {
+    const { heartbeatSyncLock } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    dbChainMockFns.returning.mockResolvedValueOnce([])
+    expect(await heartbeatSyncLock('c-1', 'run-a')).toBe(false)
+
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'c-1' }])
+    expect(await heartbeatSyncLock('c-1', 'run-a')).toBe(true)
+  })
+})