fix(connectors): identify which run holds a connector's sync lock

Review round 3 on #6909.

Guarding terminal writes on status='syncing' proved only that *a* run held the
lock, not that this one did. After a stale-lock reclaim dispatched a replacement,
the original run's guard matched the replacement's own lock and clobbered both
its state and the reaper's bookkeeping — and the live run's write was then
rejected. The dead run won and the live one was discarded, which is worse than
the last-write-wins behavior the guard replaced.

A nullable sync_lock_token is stamped in the same statement that claims the lock,
so ownership is established atomically with acquisition and matching it proves
the lock is still this run's. status='syncing' stays alongside it as defence in
depth and to keep a connector paused mid-sync from being flipped back to active.

Rejected using updatedAt as an optimistic-concurrency token: connector updates
bump it unconditionally with no status guard, so a user editing config mid-sync
would strand the connector in syncing until the reaper cleared it two hours
later.

Migration is additive, nullable, no backfill. Existing rows read NULL, which no
in-flight run can match, so a sync spanning the deploy loses only its terminal
write and is re-run by the scheduler.

Author: waleedlatif1

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

@@ -82,6 +82,9 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
         lastSyncError: STALE_LOCK_ERROR_MESSAGE,
         nextSyncAt: reclaimedNextSyncAt(),
         consecutiveFailures: reclaimedFailureCount(),
+        // Releases the reclaimed run's ownership token so its terminal write can
+        // no longer match, even before a replacement takes the lock.
+        syncLockToken: null,
         updatedAt: sql`now()`,
       })
       .where(

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

@@ -5,6 +5,7 @@ import {
   authOAuthUtilsMock,
   dbChainMockFns,
   drizzleOrmMock,
+  flattenMockConditions,
   hasMockCondition,
   type MockCondition,
   resetDbChainMock,
@@ -1103,7 +1104,7 @@ describe('stillHoldsSyncLock', () => {
      */
     expect(
       hasMockCondition(
-        stillHoldsSyncLock('c-1'),
+        stillHoldsSyncLock('c-1', 'run-a'),
         (node: MockCondition) =>
           node.type === 'eq' &&
           node.left === schemaMock.knowledgeConnector.status &&
@@ -1115,7 +1116,7 @@ describe('stillHoldsSyncLock', () => {
   it('still scopes to the connector and skips archived or deleted rows', async () => {
     const { stillHoldsSyncLock } = await import('@/lib/knowledge/connectors/sync-engine')
 
-    const condition = stillHoldsSyncLock('c-1')
+    const condition = stillHoldsSyncLock('c-1', 'run-a')
 
     expect(
       hasMockCondition(
@@ -1157,7 +1158,7 @@ describe('writeTerminalConnectorState', () => {
      * both terminal paths route through here and neither builds a WHERE clause,
      * so removing the guard is a single-site edit that this assertion catches.
      */
-    await writeTerminalConnectorState('c-1', { status: 'active' })
+    await writeTerminalConnectorState('c-1', 'run-a', { status: 'active' })
 
     const where = dbChainMockFns.where.mock.calls[0][0]
     expect(
@@ -1178,13 +1179,23 @@ describe('writeTerminalConnectorState', () => {
           node.right === 'c-1'
       )
     ).toBe(true)
+    // The token must be the run's own, not some other value that merely fills the slot.
+    expect(
+      hasMockCondition(
+        where,
+        (node: MockCondition) =>
+          node.type === 'eq' &&
+          node.left === schemaMock.knowledgeConnector.syncLockToken &&
+          node.right === 'run-a'
+      )
+    ).toBe(true)
   })
 
   it('passes the caller values through untouched', async () => {
     const { writeTerminalConnectorState } = await import('@/lib/knowledge/connectors/sync-engine')
 
     const values = { status: 'error', consecutiveFailures: 4, nextSyncAt: null }
-    await writeTerminalConnectorState('c-1', values)
+    await writeTerminalConnectorState('c-1', 'run-a', values)
 
     expect(dbChainMockFns.set.mock.calls[0][0]).toEqual(values)
   })
@@ -1193,10 +1204,10 @@ describe('writeTerminalConnectorState', () => {
     const { writeTerminalConnectorState } = await import('@/lib/knowledge/connectors/sync-engine')
 
     dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'c-1' }])
-    expect(await writeTerminalConnectorState('c-1', { status: 'active' })).toBe(true)
+    expect(await writeTerminalConnectorState('c-1', 'run-a', { status: 'active' })).toBe(true)
 
     dbChainMockFns.returning.mockResolvedValueOnce([])
-    expect(await writeTerminalConnectorState('c-1', { status: 'active' })).toBe(false)
+    expect(await writeTerminalConnectorState('c-1', 'run-a', { status: 'active' })).toBe(false)
   })
 })
 
@@ -1234,3 +1245,99 @@ describe('applySupersededOutcome', () => {
     expect(applySupersededOutcome(result, false)).toMatchObject(result)
   })
 })
+
+/**
+ * Evaluates a mocked drizzle condition tree against a plain row.
+ *
+ * The row-queue mocks return whatever was queued regardless of the predicate, so
+ * "this WHERE admits run B and rejects run A" is only observable by interpreting
+ * the condition tree the guard emits.
+ */
+function conditionMatchesRow(condition: unknown, row: Record<string, unknown>): boolean {
+  return flattenMockConditions(condition).every((node) => {
+    if (node.type === 'eq') return row[node.left as string] === node.right
+    if (node.type === 'isNull') return row[node.column as string] == null
+    throw new Error(`unhandled condition node: ${String(node.type)}`)
+  })
+}
+
+describe('sync lock ownership across a reclaim and reacquire', () => {
+  const RUN_A = 'run-a'
+  const RUN_B = 'run-b'
+
+  /** The connector row once run B has taken the lock that run A used to hold. */
+  const rowHeldByB = {
+    id: 'c-1',
+    status: 'syncing',
+    syncLockToken: RUN_B,
+    archivedAt: null,
+    deletedAt: null,
+  }
+
+  it('rejects the reclaimed run A and admits the live run B', async () => {
+    const { stillHoldsSyncLock } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    /**
+     * A outlived the TTL, the reaper reclaimed its lock, and replacement B took
+     * it — so the row reads `syncing` again. Guarding on status alone matched A
+     * here and let the dead run clobber the live one, then rejected B's own
+     * write as superseded. Exactly inverted.
+     */
+    expect(conditionMatchesRow(stillHoldsSyncLock('c-1', RUN_A), rowHeldByB)).toBe(false)
+    expect(conditionMatchesRow(stillHoldsSyncLock('c-1', RUN_B), rowHeldByB)).toBe(true)
+  })
+
+  it('rejects a run whose lock was reclaimed with no replacement yet', async () => {
+    const { stillHoldsSyncLock } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    const reclaimed = {
+      id: 'c-1',
+      status: 'error',
+      syncLockToken: null,
+      archivedAt: null,
+      deletedAt: null,
+    }
+
+    expect(conditionMatchesRow(stillHoldsSyncLock('c-1', RUN_A), reclaimed)).toBe(false)
+  })
+
+  it('admits the run that still holds its own lock', async () => {
+    const { stillHoldsSyncLock } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    const heldByA = { ...rowHeldByB, syncLockToken: RUN_A }
+
+    expect(conditionMatchesRow(stillHoldsSyncLock('c-1', RUN_A), heldByA)).toBe(true)
+  })
+
+  it('rejects a run whose connector was paused mid-sync', async () => {
+    const { stillHoldsSyncLock } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    const paused = { ...rowHeldByB, status: 'paused', syncLockToken: RUN_A }
+
+    expect(conditionMatchesRow(stillHoldsSyncLock('c-1', RUN_A), paused)).toBe(false)
+  })
+
+  it('releases the token when a run writes its terminal success state', async () => {
+    const { buildSyncSuccessUpdate } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    // A stale token left behind could match a later run reusing the same id.
+    expect(buildSyncSuccessUpdate(new Date(), 1, null, null).syncLockToken).toBeNull()
+  })
+})
+
+describe('buildSyncLockAcquisition', () => {
+  it('claims the lock and stamps ownership in one payload', async () => {
+    const { buildSyncLockAcquisition } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    const now = new Date('2026-08-20T00:00:00.000Z')
+    const acquisition = buildSyncLockAcquisition('run-a', now)
+
+    /**
+     * Without the token here every terminal write would fail to match its own
+     * run, so every sync would report superseded and leave the connector stuck
+     * `syncing` until the reaper cleared it.
+     */
+    expect(acquisition.syncLockToken).toBe('run-a')
+    expect(acquisition.status).toBe('syncing')
+  })
+})

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

@@ -337,27 +337,47 @@ export async function completeSyncLog(
 /**
  * Matches the connector row only while this run still holds its sync lock.
  *
- * `executeSync` sets `status = 'syncing'` when it acquires the lock, so that
- * value means "I am still the writer". Anything else means another actor took
- * the row: the scheduler's stale sweep reclaimed it to `error`/`disabled` and
- * may already have dispatched a replacement, or a user paused it. In every such
- * case this run's terminal write must not land — otherwise it clears a backoff
- * the breaker just set, un-disables a connector, or flips a paused connector
- * back to `active`.
+ * `status = 'syncing'` alone is not enough: it asserts that *a* run holds the
+ * lock, not that *this* run does. Once the scheduler reclaims a stale lock and
+ * dispatches a replacement, the replacement sets `syncing` again — so the
+ * original run would match, overwrite the replacement's in-flight state and the
+ * reclaim's bookkeeping, and then reject the replacement's own write as
+ * superseded. The dead run wins and the live one loses, which is worse than the
+ * unguarded last-write-wins it replaced.
+ *
+ * `syncLockToken` is written in the same CAS that takes the lock, so matching it
+ * proves the lock is still this run's. `status` is kept alongside as defence in
+ * depth and to cover a user pausing the connector mid-run.
  *
  * Guards both terminal paths. The failure path needs it as much as the success
  * path: a reclaimed run's failure would double-increment a counter the sweep
  * already advanced and overwrite its backoff with a shorter one.
  */
-export function stillHoldsSyncLock(connectorId: string) {
+export function stillHoldsSyncLock(connectorId: string, syncLockToken: string) {
   return and(
     eq(knowledgeConnector.id, connectorId),
     eq(knowledgeConnector.status, 'syncing'),
+    eq(knowledgeConnector.syncLockToken, syncLockToken),
     isNull(knowledgeConnector.archivedAt),
     isNull(knowledgeConnector.deletedAt)
   )
 }
 
+/**
+ * The connector row a run writes when it takes the sync lock.
+ *
+ * `syncLockToken` is set here, in the same statement as `status`, so ownership
+ * and the lock are established atomically — a token written afterwards would
+ * leave a window where a terminal write could not identify its own run.
+ */
+export function buildSyncLockAcquisition(syncLogId: string, now: Date) {
+  return {
+    status: 'syncing' as const,
+    syncLockToken: syncLogId,
+    updatedAt: now,
+  }
+}
+
 /** Columns a terminal write may set. Both paths write a subset of the same set. */
 type ConnectorTerminalUpdate = Partial<typeof knowledgeConnector.$inferInsert>
 
@@ -373,12 +393,13 @@ type ConnectorTerminalUpdate = Partial<typeof knowledgeConnector.$inferInsert>
  */
 export async function writeTerminalConnectorState(
   connectorId: string,
+  syncLockToken: string,
   values: ConnectorTerminalUpdate
 ): Promise<boolean> {
   const written = await db
     .update(knowledgeConnector)
     .set(values)
-    .where(stillHoldsSyncLock(connectorId))
+    .where(stillHoldsSyncLock(connectorId, syncLockToken))
     .returning({ id: knowledgeConnector.id })
 
   return written.length > 0
@@ -609,6 +630,8 @@ export function buildSyncSuccessUpdate(
     lastSyncDocCount: actualDocCount,
     nextSyncAt,
     consecutiveFailures: 0,
+    // Releases the lock so a stale token can never match a later run.
+    syncLockToken: null,
     updatedAt: now,
   }
 }
@@ -1038,9 +1061,17 @@ export async function executeSync(
   }
   const sourceConfig = connector.sourceConfig as Record<string, unknown>
 
+  /**
+   * Identifies this run for the terminal writes. Generated before the CAS and
+   * written by it, so ownership is established atomically with the lock — and
+   * reused as the sync-log row id, which makes the connector row point at the
+   * run that holds it.
+   */
+  const syncLogId = generateId()
+
   const lockResult = await db
     .update(knowledgeConnector)
-    .set({ status: 'syncing', updatedAt: new Date() })
+    .set(buildSyncLockAcquisition(syncLogId, new Date()))
     .where(
       and(
         eq(knowledgeConnector.id, connectorId),
@@ -1058,7 +1089,6 @@ export async function executeSync(
     return { ...result, error: 'sync_in_progress' }
   }
 
-  const syncLogId = generateId()
   const syncStartedAt = new Date()
   await db.insert(knowledgeConnectorSyncLog).values({
     id: syncLogId,
@@ -1863,6 +1893,7 @@ export async function executeSync(
     const now = new Date()
     const successWriteLanded = await writeTerminalConnectorState(
       connectorId,
+      syncLogId,
       buildSyncSuccessUpdate(
         now,
         actualDocCount,
@@ -1931,13 +1962,14 @@ export async function executeSync(
         })
       }
 
-      const failureWriteLanded = await writeTerminalConnectorState(connectorId, {
+      const failureWriteLanded = await writeTerminalConnectorState(connectorId, syncLogId, {
         status: disabled ? 'disabled' : 'error',
         lastSyncError: disabled
           ? 'Connector disabled after repeated sync failures. Please reconnect.'
           : errorMessage,
         nextSyncAt: nextSync,
         consecutiveFailures: failures,
+        syncLockToken: null,
         updatedAt: now,
       })
 

packages/db/migrations/0297_connector_sync_lock_token.sql

@@ -0,0 +1,8 @@
+-- Identifies which sync run holds a connector's lock. `status = 'syncing'` only
+-- proves *a* run holds it, so after a stale-lock reclaim + replacement dispatch
+-- the original run still matched and clobbered the replacement's state.
+--
+-- Additive and nullable with no backfill: existing rows read NULL, which no
+-- in-flight run can match, so the worst case for a sync already running across
+-- the deploy is that its terminal write is skipped and the scheduler re-runs it.
+ALTER TABLE "knowledge_connector" ADD COLUMN IF NOT EXISTS "sync_lock_token" text;

packages/db/migrations/meta/_journal.json

@@ -2073,6 +2073,13 @@
       "when": 1787254479175,
       "tag": "0296_organization_byok_keys",
       "breakpoints": true
+    },
+    {
+      "idx": 297,
+      "version": "7",
+      "when": 1787254579175,
+      "tag": "0297_connector_sync_lock_token",
+      "breakpoints": true
     }
   ]
 }

packages/db/schema.ts

@@ -4321,6 +4321,15 @@ export const knowledgeConnector = pgTable(
     lastSyncDocCount: integer('last_sync_doc_count'),
     nextSyncAt: timestamp('next_sync_at'),
     consecutiveFailures: integer('consecutive_failures').notNull().default(0),
+    /**
+     * Identifies the sync run that currently holds this connector's lock.
+     *
+     * `status = 'syncing'` only says *a* run holds it. After the scheduler
+     * reclaims a stale lock and dispatches a replacement, the original run would
+     * still see `syncing` and overwrite the replacement's state. Terminal writes
+     * match this token so a run can prove the lock is still *its own*.
+     */
+    syncLockToken: text('sync_lock_token'),
     createdAt: timestamp('created_at').notNull().defaultNow(),
     updatedAt: timestamp('updated_at').notNull().defaultNow(),
     archivedAt: timestamp('archived_at'),

packages/testing/src/mocks/schema.mock.ts

@@ -1269,6 +1269,7 @@ export const schemaMock = {
     lastSyncDocCount: 'lastSyncDocCount',
     nextSyncAt: 'nextSyncAt',
     consecutiveFailures: 'consecutiveFailures',
+    syncLockToken: 'syncLockToken',
     createdAt: 'createdAt',
     updatedAt: 'updatedAt',
     archivedAt: 'archivedAt',