fix(connectors): stop a reclaimed run from overwriting the reaper's verdict

Review round 2 on #6909.

The terminal connector writes were unguarded, so a run that outlived the stale
lock could still land its result after the reaper reclaimed it: flipping status
back to active, zeroing consecutiveFailures, and erasing a backoff or an
auto-disable the breaker had just applied.

Both terminal paths now go through a single writer that applies the
still-holds-the-lock guard itself, so no future terminal path can be added
without it. The knowledge-base-deleted write stays outside deliberately — it
runs before the lock is acquired, so the guard would silently discard it.

A superseded success is reported as an error rather than a clean sync, matching
the treatment lock contention already gets. The failure path deliberately keeps
its real error message instead: it already reports failure, so overwriting the
cause would lose the diagnostic and gain nothing.

Falls out of the same guard: a connector paused mid-sync is no longer flipped
back to active by the completing run.

Author: waleedlatif1

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

@@ -1091,3 +1091,146 @@ describe('completeSyncLog', () => {
     ).toBe(true)
   })
 })
+
+describe('stillHoldsSyncLock', () => {
+  it('requires the connector to still be syncing', async () => {
+    const { stillHoldsSyncLock } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    /**
+     * Without this a run reclaimed by the stale sweep still writes its terminal
+     * result: clearing the backoff, un-disabling the connector, and resetting a
+     * failure counter the sweep just advanced.
+     */
+    expect(
+      hasMockCondition(
+        stillHoldsSyncLock('c-1'),
+        (node: MockCondition) =>
+          node.type === 'eq' &&
+          node.left === schemaMock.knowledgeConnector.status &&
+          node.right === 'syncing'
+      )
+    ).toBe(true)
+  })
+
+  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')
+
+    expect(
+      hasMockCondition(
+        condition,
+        (node: MockCondition) =>
+          node.type === 'eq' &&
+          node.left === schemaMock.knowledgeConnector.id &&
+          node.right === 'c-1'
+      )
+    ).toBe(true)
+    expect(
+      hasMockCondition(
+        condition,
+        (node: MockCondition) =>
+          node.type === 'isNull' && node.column === schemaMock.knowledgeConnector.archivedAt
+      )
+    ).toBe(true)
+    expect(
+      hasMockCondition(
+        condition,
+        (node: MockCondition) =>
+          node.type === 'isNull' && node.column === schemaMock.knowledgeConnector.deletedAt
+      )
+    ).toBe(true)
+  })
+})
+
+describe('writeTerminalConnectorState', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  it('applies the sync-lock guard itself so no caller can omit it', async () => {
+    const { writeTerminalConnectorState } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    /**
+     * The property that closes the gap a shared-helper-by-convention left open:
+     * 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' })
+
+    const where = dbChainMockFns.where.mock.calls[0][0]
+    expect(
+      hasMockCondition(
+        where,
+        (node: MockCondition) =>
+          node.type === 'eq' &&
+          node.left === schemaMock.knowledgeConnector.status &&
+          node.right === 'syncing'
+      )
+    ).toBe(true)
+    expect(
+      hasMockCondition(
+        where,
+        (node: MockCondition) =>
+          node.type === 'eq' &&
+          node.left === schemaMock.knowledgeConnector.id &&
+          node.right === 'c-1'
+      )
+    ).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)
+
+    expect(dbChainMockFns.set.mock.calls[0][0]).toEqual(values)
+  })
+
+  it('reports whether the write landed', async () => {
+    const { writeTerminalConnectorState } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'c-1' }])
+    expect(await writeTerminalConnectorState('c-1', { status: 'active' })).toBe(true)
+
+    dbChainMockFns.returning.mockResolvedValueOnce([])
+    expect(await writeTerminalConnectorState('c-1', { status: 'active' })).toBe(false)
+  })
+})
+
+describe('applySupersededOutcome', () => {
+  const result = {
+    docsAdded: 3,
+    docsUpdated: 1,
+    docsDeleted: 0,
+    docsUnchanged: 2,
+    docsFailed: 0,
+  }
+
+  it('leaves a run that kept its lock untouched', async () => {
+    const { applySupersededOutcome } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    expect(applySupersededOutcome(result, true)).toEqual(result)
+  })
+
+  it('flags a discarded run so the task wrapper does not report it as clean', async () => {
+    const { applySupersededOutcome, SUPERSEDED_SYNC_ERROR } = await import(
+      '@/lib/knowledge/connectors/sync-engine'
+    )
+
+    const superseded = applySupersededOutcome(result, false)
+
+    // The task wrapper reports `success: !result.error`.
+    expect(superseded.error).toBe(SUPERSEDED_SYNC_ERROR)
+    expect(Boolean(superseded.error)).toBe(true)
+  })
+
+  it('preserves the document counters of the discarded run', async () => {
+    const { applySupersededOutcome } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    // Those writes landed — only the connector-level bookkeeping was discarded.
+    expect(applySupersededOutcome(result, false)).toMatchObject(result)
+  })
+})

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

@@ -334,6 +334,76 @@ 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`.
+ *
+ * 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) {
+  return and(
+    eq(knowledgeConnector.id, connectorId),
+    eq(knowledgeConnector.status, 'syncing'),
+    isNull(knowledgeConnector.archivedAt),
+    isNull(knowledgeConnector.deletedAt)
+  )
+}
+
+/** Columns a terminal write may set. Both paths write a subset of the same set. */
+type ConnectorTerminalUpdate = Partial<typeof knowledgeConnector.$inferInsert>
+
+/**
+ * The only way a sync run writes its terminal state onto the connector row.
+ *
+ * Callers pass their own values and never build a WHERE clause: the
+ * {@link stillHoldsSyncLock} guard is applied here, so there is exactly one
+ * place it can be removed from and a terminal path added later cannot forget
+ * it. Returns whether the write landed — false means the run was reclaimed
+ * mid-flight and its bookkeeping was discarded in favour of whoever took the
+ * row.
+ */
+export async function writeTerminalConnectorState(
+  connectorId: string,
+  values: ConnectorTerminalUpdate
+): Promise<boolean> {
+  const written = await db
+    .update(knowledgeConnector)
+    .set(values)
+    .where(stillHoldsSyncLock(connectorId))
+    .returning({ id: knowledgeConnector.id })
+
+  return written.length > 0
+}
+
+/**
+ * Reported when a run's terminal write matched no rows because the run no longer
+ * held its lock. Its document writes still landed; only its connector-level
+ * bookkeeping was discarded, in favour of whoever reclaimed the row.
+ */
+export const SUPERSEDED_SYNC_ERROR = 'sync_superseded'
+
+/**
+ * Marks a superseded run so the task wrapper's `success: !result.error` does not
+ * report a discarded run as a clean sync — the same reason a lock-contended run
+ * returns `sync_in_progress` rather than an empty success.
+ */
+export function applySupersededOutcome(
+  result: SyncResult,
+  terminalWriteLanded: boolean
+): SyncResult {
+  if (terminalWriteLanded) return result
+  return { ...result, error: SUPERSEDED_SYNC_ERROR }
+}
+
 /**
  * Decides whether deletion reconciliation may run for a sync.
  *
@@ -1791,23 +1861,24 @@ export async function executeSync(
       )
 
     const now = new Date()
-    await db
-      .update(knowledgeConnector)
-      .set(
-        buildSyncSuccessUpdate(
-          now,
-          actualDocCount,
-          calculateNextSyncTime(connector.syncIntervalMinutes),
-          reconciliationHoldNotice
-        )
-      )
-      .where(
-        and(
-          eq(knowledgeConnector.id, connectorId),
-          isNull(knowledgeConnector.archivedAt),
-          isNull(knowledgeConnector.deletedAt)
-        )
+    const successWriteLanded = await writeTerminalConnectorState(
+      connectorId,
+      buildSyncSuccessUpdate(
+        now,
+        actualDocCount,
+        calculateNextSyncTime(connector.syncIntervalMinutes),
+        reconciliationHoldNotice
       )
+    )
+
+    if (!successWriteLanded) {
+      logger.warn('Sync result discarded — connector was reclaimed while this run was executing', {
+        connectorId,
+        syncLogId,
+        ...result,
+      })
+      return applySupersededOutcome(result, false)
+    }
 
     logger.info('Sync completed', { connectorId, ...result })
     return result
@@ -1860,24 +1931,29 @@ export async function executeSync(
         })
       }
 
-      await db
-        .update(knowledgeConnector)
-        .set({
-          status: disabled ? 'disabled' : 'error',
-          lastSyncError: disabled
-            ? 'Connector disabled after repeated sync failures. Please reconnect.'
-            : errorMessage,
-          nextSyncAt: nextSync,
-          consecutiveFailures: failures,
-          updatedAt: now,
-        })
-        .where(
-          and(
-            eq(knowledgeConnector.id, connectorId),
-            isNull(knowledgeConnector.archivedAt),
-            isNull(knowledgeConnector.deletedAt)
-          )
+      const failureWriteLanded = await writeTerminalConnectorState(connectorId, {
+        status: disabled ? 'disabled' : 'error',
+        lastSyncError: disabled
+          ? 'Connector disabled after repeated sync failures. Please reconnect.'
+          : errorMessage,
+        nextSyncAt: nextSync,
+        consecutiveFailures: failures,
+        updatedAt: now,
+      })
+
+      /**
+       * Deliberately does NOT get {@link applySupersededOutcome}. `result.error`
+       * is set to the real failure cause below and the task wrapper already
+       * reports this run as unsuccessful, so overwriting it with
+       * `sync_superseded` would destroy the diagnostic without changing the
+       * reported outcome. The supersession is carried by this log line instead.
+       */
+      if (!failureWriteLanded) {
+        logger.warn(
+          'Sync failure discarded — connector was reclaimed while this run was executing',
+          { connectorId, syncLogId, error: errorMessage }
         )
+      }
     } catch (recoveryError) {
       logger.error('Failed to record sync failure', {
         connectorId,