fix(connectors): require a lock to be held, owned, and heartbeated before sparing its log row

The previous round spared a sync-log row whose id was still the connector's lock
token. That is necessary but not sufficient: the reaper's reclaim filters out
archived and soft-deleted connectors, so their rows keep status syncing with an
intact token indefinitely and would have been spared forever. It also spared a
run that died without ever being reclaimed.

Keying on connector liveness alone has the opposite gap — it cannot tell which
run holds the lock, so an orphan on a connector that is immediately re-locked by
a replacement is spared until a tick happens to catch the connector idle.

Sparing now requires all three: the connector is locked, this row's run is the
holder, and that lock is being heartbeated. An orphan can satisfy at most two,
so no timing window strands one.

This is a per-row liveness predicate, not a restriction of the candidate set —
every stale row is still evaluated. A backlog row predating the token column
fails on two counts and still drains, so the sweep remains self-healing.

Author: waleedlatif1

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

@@ -214,27 +214,64 @@ describe('connector sync scheduler stale-lock reaper', () => {
     ).toBe(true)
   })
 
-  it('spares the log row of a run that still holds its connector lock', async () => {
+  /** The `NOT EXISTS` liveness fragment the sweep's WHERE carries. */
+  function sweepLivenessFragment(): MockSqlFragment {
+    const where = dbChainMockFns.where.mock.calls[1][0]
+    const fragment = flattenMockConditions(where).find(
+      (node: MockCondition) => typeof node.toSQL === 'function'
+    )
+    expect(fragment).toBeDefined()
+    return fragment as unknown as MockSqlFragment
+  }
+
+  it('spares the log row of a run whose lock is still being heartbeated', async () => {
     await runTickRecovering(['connector-1'])
 
     /**
      * The sweep keys on `startedAt`, which no heartbeat refreshes, so age alone
      * would close a legitimately long in-process run's row and record a
-     * successful sync as failed.
+     * successful sync as failed. Every clause is pinned: sparing requires the
+     * connector to be locked, THIS row's run to be the holder, and that lock to
+     * be live — an orphan can satisfy at most two.
      */
-    const where = dbChainMockFns.where.mock.calls[1][0]
-    const liveness = flattenMockConditions(where).find(
-      (node: MockCondition) => typeof node.toSQL === 'function'
+    const rendered = sweepLivenessFragment().toSQL().sql.replace(/\s+/g, ' ').trim()
+
+    expect(rendered).toBe(
+      "NOT EXISTS ( SELECT 1 FROM ? WHERE ? = ? AND ? = ? AND ? = 'syncing' AND ? > ? )"
     )
-    expect(liveness).toBeDefined()
+  })
 
-    const rendered = (liveness as unknown as MockSqlFragment).toSQL().sql
-    expect(rendered).toContain('NOT EXISTS')
-    expect(rendered).toContain("'syncing'")
+  it('identifies the lock holder by token, not merely by the connector syncing', async () => {
+    await runTickRecovering(['connector-1'])
 
-    const bound = (liveness as unknown as MockSqlFragment).values
-    expect(bound).toContain(schemaMock.knowledgeConnector.syncLockToken)
-    expect(bound).toContain(schemaMock.knowledgeConnectorSyncLog.id)
+    /**
+     * Without the token clause the sweep spares every `started` row on a locked
+     * connector — including an orphan from a crashed run whose replacement now
+     * holds the lock, which would then never drain while that connector stays
+     * busy.
+     */
+    expect(sweepLivenessFragment().values).toContain(schemaMock.knowledgeConnector.syncLockToken)
+  })
+
+  it('requires the held lock to be heartbeated, not merely held', async () => {
+    await runTickRecovering(['connector-1'])
+
+    /**
+     * Without the freshness clause a run that died without being reclaimed — or
+     * one on an archived or deleted connector, which the reclaim skips entirely
+     * — keeps `status = 'syncing'` and its token forever, so its row is spared
+     * forever.
+     */
+    const bound = sweepLivenessFragment().values
+    expect(bound).toContain(schemaMock.knowledgeConnector.updatedAt)
+
+    const cutoff = bound.find(
+      (value): value is { value: Date } =>
+        typeof value === 'object' &&
+        value !== null &&
+        (value as { value?: unknown }).value instanceof Date
+    )
+    expect(cutoff).toBeDefined()
   })
 
   it('closes stale sync-log rows even when no connector was reclaimed this tick', async () => {

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

@@ -32,25 +32,37 @@ const DISPATCH_CONCURRENCY = 10
 const STALE_LOCK_ERROR_MESSAGE = 'Sync timed out (stale lock recovered)'
 
 /**
- * Excludes a sync-log row whose run still demonstrably holds its connector's
- * lock.
+ * Excludes a sync-log row belonging to a run that is demonstrably still alive.
  *
  * The sweep keys on `startedAt`, and nothing refreshes that — the heartbeat
  * renews `knowledge_connector.updatedAt`, and the log table has no equivalent
  * column. So a legitimately long in-process run keeps its connector lock but
  * would still have its log row closed as `failed` at the TTL, recording a
  * successful sync as a failure and losing its counters to
- * `loadPreviousListingObservation`. Matching the connector's `syncLockToken`
- * against the row's own id is exactly "this run is still the lock holder", so a
- * live run is spared while every orphan — reclaimed, replaced, or predating the
- * token column, where the token is NULL — is still swept.
+ * `loadPreviousListingObservation`, which reads only `completed` rows.
+ *
+ * The heartbeat is the single source of liveness truth, so this defers to it.
+ * Sparing requires all three of: the connector is locked, THIS row's run is the
+ * lock holder, and that lock is being heartbeated. An orphan can satisfy at most
+ * two, so none is ever stranded:
+ * - reclaimed after a hard kill — connector is `error`, token cleared;
+ * - a replacement holds the lock — the token is the successor's, not this row's;
+ * - died without being reclaimed, including on an archived or deleted connector
+ *   the reclaim skips entirely — `updatedAt` is stale.
+ *
+ * This re-references the connector row, which an earlier fix deliberately moved
+ * away from. That coupling was different: it restricted the sweep's candidate
+ * set to *this tick's reclaims*, which made a pre-existing backlog undrainable.
+ * This is a per-row liveness predicate — every stale row is still a candidate,
+ * so the sweep stays self-healing.
  */
-function runNoLongerHoldsItsLock(): SQL {
+function logRowNotHeldByLiveRun(staleCutoff: Date): SQL {
   return sql`NOT EXISTS (
     SELECT 1 FROM ${knowledgeConnector}
     WHERE ${knowledgeConnector.id} = ${knowledgeConnectorSyncLog.connectorId}
       AND ${knowledgeConnector.syncLockToken} = ${knowledgeConnectorSyncLog.id}
       AND ${knowledgeConnector.status} = 'syncing'
+      AND ${knowledgeConnector.updatedAt} > ${sql.param(staleCutoff, knowledgeConnector.updatedAt)}
   )`
 }
 
@@ -142,8 +154,8 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
      *
      * Age alone does not prove a run is dead: the in-process fallback path has
      * no duration cap, so a large self-hosted sync can genuinely still be
-     * working past the TTL. `runNoLongerHoldsItsLock` is what makes this safe —
-     * a run still holding its connector's lock is spared regardless of age.
+     * working past the TTL. `logRowNotHeldByLiveRun` is what makes this safe —
+     * a run whose lock is still being heartbeated is spared regardless of age.
      * The age predicate is also per-row on `startedAt`, so a fresh run's log row
      * can never be caught by it, even on a connector whose previous run is being
      * reclaimed in this same tick.
@@ -159,7 +171,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
         and(
           eq(knowledgeConnectorSyncLog.status, 'started'),
           lte(knowledgeConnectorSyncLog.startedAt, staleCutoff),
-          runNoLongerHoldsItsLock()
+          logRowNotHeldByLiveRun(staleCutoff)
         )
       )
       .returning({ id: knowledgeConnectorSyncLog.id })