fix(kb): let the server say a connector sync is queued

apps/docs/openapi-v2-knowledge.json

@@ -3450,8 +3450,8 @@
           },
           "status": {
             "type": "string",
-            "enum": ["active", "paused", "syncing", "error", "disabled"],
-            "description": "Current connector state."
+            "enum": ["active", "paused", "pending", "syncing", "error", "disabled"],
+            "description": "Current connector state. `pending` means a sync is queued but not yet running."
           },
           "lastSyncAt": {
             "anyOf": [
@@ -3840,8 +3840,8 @@
           },
           "status": {
             "type": "string",
-            "enum": ["active", "paused", "syncing", "error", "disabled"],
-            "description": "Current connector state."
+            "enum": ["active", "paused", "pending", "syncing", "error", "disabled"],
+            "description": "Current connector state. `pending` means a sync is queued but not yet running."
           },
           "lastSyncAt": {
             "anyOf": [

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

@@ -109,6 +109,22 @@ function whereForUpdate(index: number): unknown {
   return dbChainMockFns.where.mock.calls[index][0]
 }
 
+/**
+ * Position of the update targeting a given table, resolved by table rather than
+ * hardcoded: the tick runs several updates and a new one inserted between them
+ * would otherwise silently re-point every later assertion at the wrong chain.
+ */
+function updateIndexFor(table: unknown): number {
+  const index = dbChainMockFns.update.mock.calls.findIndex((call) => call[0] === table)
+  expect(index).toBeGreaterThanOrEqual(0)
+  return index
+}
+
+/** The sync-log sweep's `.where()` condition, whichever chain it ran as. */
+function syncLogSweepWhere(): unknown {
+  return whereForUpdate(updateIndexFor(schemaMock.knowledgeConnectorSyncLog))
+}
+
 beforeEach(() => {
   vi.clearAllMocks()
   resetDbChainMock()
@@ -256,14 +272,14 @@ describe('connector sync scheduler stale-lock reaper', () => {
   it('closes orphaned sync-log rows still marked started', async () => {
     await runTickRecovering(['connector-1', 'connector-2'])
 
-    expect(dbChainMockFns.update.mock.calls[1][0]).toBe(schemaMock.knowledgeConnectorSyncLog)
+    const logUpdateIndex = updateIndexFor(schemaMock.knowledgeConnectorSyncLog)
 
-    const payload = setPayloadForUpdate(1)
+    const payload = setPayloadForUpdate(logUpdateIndex)
     expect(payload.status).toBe('failed')
     expect(renderedSql(payload.completedAt)).toContain('now()')
     expect(payload.errorMessage).toBe('Sync timed out (stale lock recovered)')
 
-    const where = dbChainMockFns.where.mock.calls[1][0]
+    const where = syncLogSweepWhere()
     expect(
       hasMockCondition(
         where,
@@ -284,7 +300,7 @@ describe('connector sync scheduler stale-lock reaper', () => {
 
   /** The `NOT EXISTS` liveness fragment the sweep's WHERE carries. */
   function sweepLivenessFragment(): MockSqlFragment {
-    const where = dbChainMockFns.where.mock.calls[1][0]
+    const where = syncLogSweepWhere()
     const fragment = flattenMockConditions(where).find(
       (node: MockCondition) => typeof node.toSQL === 'function'
     )
@@ -370,17 +386,56 @@ describe('connector sync scheduler stale-lock reaper', () => {
 
     expect(response.status).toBe(200)
 
-    const logUpdateIndex = dbChainMockFns.update.mock.calls.findIndex(
-      (call) => call[0] === schemaMock.knowledgeConnectorSyncLog
+    expect(setPayloadForUpdate(updateIndexFor(schemaMock.knowledgeConnectorSyncLog)).status).toBe(
+      'failed'
     )
-    expect(logUpdateIndex).toBeGreaterThanOrEqual(0)
-    expect(setPayloadForUpdate(logUpdateIndex).status).toBe('failed')
+  })
+
+  it('recovers connectors whose queued sync was never started', async () => {
+    await runTickRecovering(['connector-1'])
+
+    /** Located by its `status = 'pending'` predicate, not by position in the tick. */
+    const pendingIndex = dbChainMockFns.update.mock.calls.findIndex((call, index) => {
+      if (call[0] !== schemaMock.knowledgeConnector) return false
+      return hasMockCondition(
+        whereForUpdate(index),
+        (node: MockCondition) =>
+          node.type === 'eq' &&
+          node.left === schemaMock.knowledgeConnector.status &&
+          node.right === 'pending'
+      )
+    })
+    expect(pendingIndex).toBeGreaterThanOrEqual(0)
+
+    /**
+     * Ages against the lease, not `updatedAt`: a pending connector is still
+     * editable, and `updatedAt` moves on every unrelated write, so using it
+     * would let a config edit defer the recovery indefinitely — the bug the
+     * lease column was introduced to close for `syncing`.
+     */
+    const pendingCutoff = flattenMockConditions(whereForUpdate(pendingIndex)).find(
+      (node: MockCondition) => typeof node.toSQL === 'function'
+    ) as unknown as MockSqlFragment | undefined
+    expect(pendingCutoff?.toSQL().sql).toBe('? <= ?')
+    expectLeaseExpression(pendingCutoff?.values[0])
+    expect((pendingCutoff?.values[1] as { value: Date }).value).toEqual(EXPECTED_STALE_CUTOFF)
+
+    /** Re-enters the shared failure ladder rather than re-queueing every tick. */
+    const payload = setPayloadForUpdate(pendingIndex)
+    expect(renderedSql(payload.status)).toContain('disabled')
+    expect(renderedSql(payload.consecutiveFailures)).toBe('COALESCE(?, 0) + 1')
+
+    /**
+     * Reports a lost hand-off, not a timeout: nothing ran, so the stale-lock
+     * wording would describe a run that never existed.
+     */
+    expect(asFragment(payload.lastSyncError).values).toContain('Sync was queued but never started')
   })
 
   it('never scopes the sync-log sweep to a connector id', async () => {
     await runTickRecovering(['connector-1'])
 
-    const where = dbChainMockFns.where.mock.calls[1][0]
+    const where = syncLogSweepWhere()
 
     /**
      * Checks every position, not just `column`. `eq()` builds `{left, right}`

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

@@ -32,6 +32,14 @@ const DISPATCH_CONCURRENCY = 10
 
 const STALE_LOCK_ERROR_MESSAGE = 'Sync timed out (stale lock recovered)'
 
+/**
+ * A connector left `pending` past the TTL — its sync was queued but no worker
+ * ever took the lock, so the hand-off was lost (the process died between the
+ * two writes, or the queued run was dropped). Distinct from the stale-lock
+ * message because nothing timed out: the sync never started.
+ */
+const LOST_DISPATCH_ERROR_MESSAGE = 'Sync was queued but never started'
+
 /**
  * How long the connector holding the lock has gone without proving it is alive.
  *
@@ -57,8 +65,8 @@ function syncLockLease(): SQL {
  * breaker and this SQL breaker cannot drift into two different messages for one
  * verdict.
  */
-function reclaimedError(): SQL {
-  return sql`CASE WHEN COALESCE(${knowledgeConnector.consecutiveFailures}, 0) + 1 >= ${MAX_CONSECUTIVE_FAILURES} THEN ${CONNECTOR_AUTO_DISABLED_ERROR} ELSE ${STALE_LOCK_ERROR_MESSAGE} END`
+function reclaimedError(message: string): SQL {
+  return sql`CASE WHEN COALESCE(${knowledgeConnector.consecutiveFailures}, 0) + 1 >= ${MAX_CONSECUTIVE_FAILURES} THEN ${CONNECTOR_AUTO_DISABLED_ERROR} ELSE ${message} END`
 }
 
 /**
@@ -123,6 +131,22 @@ function reclaimedNextSyncAt(): SQL {
   return sql`CASE WHEN COALESCE(${knowledgeConnector.consecutiveFailures}, 0) + 1 >= ${MAX_CONSECUTIVE_FAILURES} THEN NULL ELSE now() + LEAST((COALESCE(${knowledgeConnector.consecutiveFailures}, 0) + 1) * ${CONNECTOR_FAILURE_BACKOFF_STEP_MINUTES}, ${CONNECTOR_FAILURE_BACKOFF_CAP_MINUTES}) * INTERVAL '1 minute' END`
 }
 
+/**
+ * The write shared by both reclaims: a connector that stopped making progress
+ * re-enters the failure ladder. Factored so the two callers cannot drift into
+ * different ladders for the same verdict — the same reason
+ * {@link reclaimedError} takes the message rather than hardcoding it.
+ */
+function reclaimPayload(message: string) {
+  return {
+    status: reclaimedStatus(),
+    lastSyncError: reclaimedError(message),
+    nextSyncAt: reclaimedNextSyncAt(),
+    consecutiveFailures: reclaimedFailureCount(),
+    updatedAt: sql`now()`,
+  }
+}
+
 /**
  * Cron endpoint that checks for connectors due for sync and dispatches sync jobs.
  * Should be called every 5 minutes by an external cron service.
@@ -141,29 +165,115 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
 
     const staleCutoff = new Date(now.getTime() - CONNECTOR_SYNC_STALE_LOCK_TTL_MS)
 
-    const recoveredConnectors = await db
-      .update(knowledgeConnector)
-      .set({
-        status: reclaimedStatus(),
-        lastSyncError: reclaimedError(),
-        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, and closes
-        // its lease so a re-locked row starts from a fresh one.
-        syncLockToken: null,
-        syncLockLeaseAt: null,
-        updatedAt: sql`now()`,
-      })
-      .where(
-        and(
-          eq(knowledgeConnector.status, 'syncing'),
-          sql`${syncLockLease()} <= ${sql.param(staleCutoff, knowledgeConnector.syncLockLeaseAt)}`,
-          isNull(knowledgeConnector.archivedAt),
-          isNull(knowledgeConnector.deletedAt)
+    /**
+     * The three recovery passes target disjoint row sets — a held-but-silent
+     * lock, a queue entry that never became one, and a sync-log row orphaned by
+     * a killed run — and none reads another's result, so they go out together
+     * rather than as three serialized round trips.
+     *
+     * `logRowNotHeldByLiveRun` is the one apparent coupling and it is benign:
+     * it spares a log row only while its connector's lease is still live, and
+     * every row the lock reclaim targets has an expired lease, so the sweep
+     * reaches the same verdict against either snapshot.
+     */
+    const [recoveredConnectors, recoveredPendingConnectors, closedSyncLogs] = await Promise.all([
+      db
+        .update(knowledgeConnector)
+        .set({
+          ...reclaimPayload(STALE_LOCK_ERROR_MESSAGE),
+          /**
+           * Releases the reclaimed run's ownership token so its terminal write
+           * can no longer match, even before a replacement takes the lock, and
+           * closes its lease so a re-locked row starts from a fresh one.
+           */
+          syncLockToken: null,
+          syncLockLeaseAt: null,
+        })
+        .where(
+          and(
+            eq(knowledgeConnector.status, 'syncing'),
+            sql`${syncLockLease()} <= ${sql.param(staleCutoff, knowledgeConnector.syncLockLeaseAt)}`,
+            isNull(knowledgeConnector.archivedAt),
+            isNull(knowledgeConnector.deletedAt)
+          )
         )
-      )
-      .returning({ id: knowledgeConnector.id })
+        .returning({ id: knowledgeConnector.id }),
+      /**
+       * Recovers connectors whose queued sync was never picked up.
+       *
+       * `pending` is written just before the hand-off to the queue, so a row that
+       * is still `pending` past the TTL means no worker ever took the lock: the
+       * process died between the two writes, or the queued run was dropped. Left
+       * alone the connector would sit `pending` forever — the stale-lock reclaim
+       * above only looks at `syncing` rows, and the due-sweep below only at
+       * `active`/`error`.
+       *
+       * Flipped to `error` rather than straight back to `active` so it re-enters
+       * through the same failure ladder as any other unsuccessful sync: repeated
+       * lost dispatches back off and eventually disable, instead of re-queueing
+       * every tick forever.
+       *
+       * Ages against {@link syncLockLease}, the same expression the stale-lock
+       * pass reads, because `markSyncPending` opens the lease when it queues.
+       * `updatedAt` would be wrong here for exactly the reason the lease column
+       * exists: a `pending` connector is still editable, so every unrelated write
+       * to the row would renew the recovery it is meant to trigger — a config
+       * edit on a stuck connector could defer it forever.
+       */
+      db
+        .update(knowledgeConnector)
+        .set({
+          ...reclaimPayload(LOST_DISPATCH_ERROR_MESSAGE),
+          /** Releases the queue entry's token so a late hand-off cannot match it. */
+          syncLockToken: null,
+          syncLockLeaseAt: null,
+        })
+        .where(
+          and(
+            eq(knowledgeConnector.status, 'pending'),
+            sql`${syncLockLease()} <= ${sql.param(staleCutoff, knowledgeConnector.syncLockLeaseAt)}`,
+            isNull(knowledgeConnector.archivedAt),
+            isNull(knowledgeConnector.deletedAt)
+          )
+        )
+        .returning({ id: knowledgeConnector.id }),
+      /**
+       * Closes sync-log rows left `started` by a killed run. Nothing else ever
+       * reconciles them, and `loadPreviousListingObservation` reads only
+       * `completed` rows, so a never-closed run silently ages out the previous
+       * observation it should have provided.
+       *
+       * Deliberately independent of this tick's reclaims rather than scoped to
+       * them. A row orphaned before this shipped — or by a transient failure of
+       * this very statement — belongs to a connector already flipped out of
+       * `syncing`, so it would never appear in a future reclaim batch and would
+       * stay stranded forever. Keying off the row's own `startedAt` instead makes
+       * the sweep self-healing and lets it drain the existing backlog.
+       *
+       * 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. `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.
+       */
+      db
+        .update(knowledgeConnectorSyncLog)
+        .set({
+          status: 'failed',
+          completedAt: sql`now()`,
+          errorMessage: STALE_LOCK_ERROR_MESSAGE,
+        })
+        .where(
+          and(
+            eq(knowledgeConnectorSyncLog.status, 'started'),
+            lte(knowledgeConnectorSyncLog.startedAt, staleCutoff),
+            logRowNotHeldByLiveRun(staleCutoff)
+          )
+        )
+        .returning({ id: knowledgeConnectorSyncLog.id }),
+    ])
 
     if (recoveredConnectors.length > 0) {
       logger.warn(
@@ -172,42 +282,12 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       )
     }
 
-    /**
-     * Closes sync-log rows left `started` by a killed run. Nothing else ever
-     * reconciles them, and `loadPreviousListingObservation` reads only
-     * `completed` rows, so a never-closed run silently ages out the previous
-     * observation it should have provided.
-     *
-     * Deliberately independent of this tick's reclaims rather than scoped to
-     * them. A row orphaned before this shipped — or by a transient failure of
-     * this very statement — belongs to a connector already flipped out of
-     * `syncing`, so it would never appear in a future reclaim batch and would
-     * stay stranded forever. Keying off the row's own `startedAt` instead makes
-     * the sweep self-healing and lets it drain the existing backlog.
-     *
-     * 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. `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.
-     */
-    const closedSyncLogs = await db
-      .update(knowledgeConnectorSyncLog)
-      .set({
-        status: 'failed',
-        completedAt: sql`now()`,
-        errorMessage: STALE_LOCK_ERROR_MESSAGE,
-      })
-      .where(
-        and(
-          eq(knowledgeConnectorSyncLog.status, 'started'),
-          lte(knowledgeConnectorSyncLog.startedAt, staleCutoff),
-          logRowNotHeldByLiveRun(staleCutoff)
-        )
+    if (recoveredPendingConnectors.length > 0) {
+      logger.warn(
+        `[${requestId}] Recovered ${recoveredPendingConnectors.length} connectors whose queued sync was never started`,
+        { ids: recoveredPendingConnectors.map((c) => c.id) }
       )
-      .returning({ id: knowledgeConnectorSyncLog.id })
+    }
 
     if (closedSyncLogs.length > 0) {
       logger.warn(`[${requestId}] Closed ${closedSyncLogs.length} orphaned connector sync log(s)`)