fix(connectors): count hard-kill failures and cap deletion blast radius

An OOM is a SIGKILL, so executeSync's catch and finally never run. The
out-of-process stale-lock reaper was the only survivor, and it cleared the lock
without ever incrementing consecutiveFailures — so MAX_CONSECUTIVE_FAILURES was
unreachable for hard kills and a crashing connector looped indefinitely on a
flat 10-minute retry, faster than any healthy interval.

- Move the failure threshold and backoff ladder into sync-limits.ts so the two
  writers cannot drift; the reaper now increments, backs off, and disables in
  the same statement that clears the lock
- Sweep sync-log rows left `started` by a killed run, keyed off the row's own
  startedAt so the sweep is self-healing and drains the existing backlog
- Delete the unreachable finally block; report a lock-contended run as skipped
  rather than as a successful zero-document sync
- Stop stamping lastSyncAt on the failure path

Deletion reconciliation only questioned listings that looked broken, leaving
every partial-outage shape between 10% and 100% unguarded: a source serving half
its documents tombstones the other half and hard-deletes it on the next pass.

- Hold a reconciliation pass whose deletions exceed a share of the corpus,
  all-or-nothing; fullSync remains the documented escape hatch
- Never make a user-excluded document deletion-eligible, guarded at deletion
  eligibility rather than at the select so resurrection still works
- Count listed documents over the same population as the owned count
- Judge the previous run against a corpus at least as large as the one present,
  which un-jams the two-strike purge — shipped with the cap, never before it

Author: waleedlatif1

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

@@ -0,0 +1,207 @@
+/**
+ * Tests for the connector sync scheduler's stale-lock reaper.
+ *
+ * A hard kill (OOM/SIGKILL) skips `executeSync`'s `catch` and `finally`, so this
+ * reaper is the only writer that ever records that failure. These tests pin the
+ * shape of the SQL it writes, which is the part no shape-agnostic mock can enforce.
+ *
+ * @vitest-environment node
+ */
+import {
+  dbChainMockFns,
+  hasMockCondition,
+  type MockCondition,
+  resetDbChainMock,
+  schemaMock,
+} from '@sim/testing'
+import type { NextRequest } from 'next/server'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import {
+  connectorFailureBackoffMinutes,
+  MAX_CONSECUTIVE_FAILURES,
+} from '@/lib/knowledge/connectors/sync-limits'
+
+const { mockVerifyCronAuth, mockDispatchSync, mockResolveSystemBillingAttribution } = vi.hoisted(
+  () => ({
+    mockVerifyCronAuth: vi.fn().mockReturnValue(null),
+    mockDispatchSync: vi.fn().mockResolvedValue(undefined),
+    mockResolveSystemBillingAttribution: vi.fn().mockResolvedValue({ workspaceId: 'ws-1' }),
+  })
+)
+
+vi.mock('@/lib/auth/internal', () => ({ verifyCronAuth: mockVerifyCronAuth }))
+vi.mock('@/lib/knowledge/connectors/queue', () => ({ dispatchSync: mockDispatchSync }))
+vi.mock('@/lib/billing/core/billing-attribution', () => ({
+  resolveSystemBillingAttribution: mockResolveSystemBillingAttribution,
+}))
+
+import { GET } from '@/app/api/knowledge/connectors/sync/route'
+
+/** A drizzle `sql` fragment as the shared test mock renders it. */
+interface MockSqlFragment {
+  values: unknown[]
+  toSQL: () => { sql: string; params: unknown[] }
+}
+
+function isSqlFragment(value: unknown): value is MockSqlFragment {
+  return typeof value === 'object' && value !== null && 'toSQL' in value && 'values' in value
+}
+
+function asFragment(value: unknown): MockSqlFragment {
+  expect(isSqlFragment(value)).toBe(true)
+  return value as MockSqlFragment
+}
+
+function renderedSql(value: unknown): string {
+  return asFragment(value).toSQL().sql
+}
+
+function numericBinds(value: unknown): number[] {
+  return asFragment(value).values.filter((v): v is number => typeof v === 'number')
+}
+
+function cronRequest(): NextRequest {
+  return new Request('https://sim.ai/api/knowledge/connectors/sync', {
+    headers: { authorization: 'Bearer test-cron-secret' },
+  }) as unknown as NextRequest
+}
+
+/** Runs one scheduler tick that reclaims the given stale connector ids. */
+async function runTickRecovering(ids: string[]) {
+  dbChainMockFns.returning.mockResolvedValueOnce(ids.map((id) => ({ id })))
+  const response = await GET(cronRequest())
+  expect(response.status).toBe(200)
+}
+
+/** The `.set()` payload of the nth `db.update()` chain in call order. */
+function setPayloadForUpdate(index: number): Record<string, unknown> {
+  return dbChainMockFns.set.mock.calls[index][0] as Record<string, unknown>
+}
+
+beforeEach(() => {
+  vi.clearAllMocks()
+  resetDbChainMock()
+  mockVerifyCronAuth.mockReturnValue(null)
+})
+
+describe('connector sync scheduler stale-lock reaper', () => {
+  it('increments consecutiveFailures in the same statement that flips the lock', async () => {
+    await runTickRecovering(['connector-1'])
+
+    expect(dbChainMockFns.update.mock.calls[0][0]).toBe(schemaMock.knowledgeConnector)
+
+    const payload = setPayloadForUpdate(0)
+    expect(payload.consecutiveFailures).toBeDefined()
+    expect(typeof payload.consecutiveFailures).not.toBe('number')
+
+    const rendered = renderedSql(payload.consecutiveFailures)
+    expect(rendered).toContain('COALESCE(')
+    expect(rendered).toContain(', 0) + 1')
+    expect(asFragment(payload.consecutiveFailures).values).toContain(
+      schemaMock.knowledgeConnector.consecutiveFailures
+    )
+  })
+
+  it('disables at the threshold and errors below it', async () => {
+    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)
+  })
+
+  it('derives nextSyncAt from the shared failure backoff ladder', async () => {
+    await runTickRecovering(['connector-1'])
+
+    const nextSyncAt = setPayloadForUpdate(0).nextSyncAt
+    const rendered = renderedSql(nextSyncAt)
+
+    expect(rendered).toContain('THEN NULL')
+    expect(rendered).toContain('LEAST(')
+    expect(rendered).toContain("INTERVAL '1 minute'")
+
+    const [threshold, step, cap] = numericBinds(nextSyncAt)
+    expect(threshold).toBe(MAX_CONSECUTIVE_FAILURES)
+
+    // Recomputing the ladder from the binds the SQL actually carries makes this
+    // fail the moment the route and `connectorFailureBackoffMinutes` drift apart.
+    for (const failures of [1, 2, 5, 10, 47, 48, 100]) {
+      expect(Math.min(failures * step, cap)).toBe(connectorFailureBackoffMinutes(failures))
+    }
+  })
+
+  it('does not stamp lastSyncAt when reclaiming a stale lock', async () => {
+    await runTickRecovering(['connector-1'])
+
+    expect(setPayloadForUpdate(0)).not.toHaveProperty('lastSyncAt')
+  })
+
+  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 payload = setPayloadForUpdate(1)
+    expect(payload.status).toBe('failed')
+    expect(renderedSql(payload.completedAt)).toContain('now()')
+    expect(payload.errorMessage).toEqual(expect.any(String))
+
+    const where = dbChainMockFns.where.mock.calls[1][0]
+    expect(
+      hasMockCondition(
+        where,
+        (node: MockCondition) =>
+          node.type === 'eq' &&
+          node.left === schemaMock.knowledgeConnectorSyncLog.status &&
+          node.right === 'started'
+      )
+    ).toBe(true)
+    expect(
+      hasMockCondition(
+        where,
+        (node: MockCondition) =>
+          node.type === 'lte' && node.left === schemaMock.knowledgeConnectorSyncLog.startedAt
+      )
+    ).toBe(true)
+  })
+
+  it('closes stale sync-log rows even when no connector was reclaimed this tick', async () => {
+    /**
+     * The self-healing assertion. A row orphaned before this sweep existed —
+     * or by a transient failure of the sweep itself — belongs to a connector
+     * already flipped out of `syncing`, so it can never appear in a reclaim
+     * batch again. Scoping the close to this tick's reclaims strands it forever.
+     */
+    const response = await GET(cronRequest())
+
+    expect(response.status).toBe(200)
+
+    const logUpdateIndex = dbChainMockFns.update.mock.calls.findIndex(
+      (call) => call[0] === schemaMock.knowledgeConnectorSyncLog
+    )
+    expect(logUpdateIndex).toBeGreaterThanOrEqual(0)
+    expect(setPayloadForUpdate(logUpdateIndex).status).toBe('failed')
+  })
+
+  it('never scopes the sync-log sweep to a connector id', async () => {
+    await runTickRecovering(['connector-1'])
+
+    const where = dbChainMockFns.where.mock.calls[1][0]
+    expect(
+      hasMockCondition(
+        where,
+        (node: MockCondition) => node.column === schemaMock.knowledgeConnectorSyncLog.connectorId
+      )
+    ).toBe(false)
+  })
+
+  it('drives the connector write off a single clock', async () => {
+    await runTickRecovering(['connector-1'])
+
+    // `updatedAt` shares the server clock the nextSyncAt interval math uses.
+    expect(renderedSql(setPayloadForUpdate(0).updatedAt)).toContain('now()')
+  })
+})

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

@@ -1,15 +1,20 @@
 import { db } from '@sim/db'
-import { knowledgeBase, knowledgeConnector } from '@sim/db/schema'
+import { knowledgeBase, knowledgeConnector, knowledgeConnectorSyncLog } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { and, asc, eq, inArray, isNull, lte } from 'drizzle-orm'
+import { and, asc, eq, inArray, isNull, lte, type SQL, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
 import { resolveSystemBillingAttribution } from '@/lib/billing/core/billing-attribution'
 import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { dispatchSync } from '@/lib/knowledge/connectors/queue'
-import { CONNECTOR_SYNC_STALE_LOCK_TTL_MS } from '@/lib/knowledge/connectors/sync-limits'
+import {
+  CONNECTOR_FAILURE_BACKOFF_CAP_MINUTES,
+  CONNECTOR_FAILURE_BACKOFF_STEP_MINUTES,
+  CONNECTOR_SYNC_STALE_LOCK_TTL_MS,
+  MAX_CONSECUTIVE_FAILURES,
+} from '@/lib/knowledge/connectors/sync-limits'
 
 export const dynamic = 'force-dynamic'
 
@@ -24,6 +29,34 @@ const MAX_DISPATCHES_PER_TICK = 200
 /** Each dispatch does a joined SELECT + conditional UPDATE against the shared pool. */
 const DISPATCH_CONCURRENCY = 10
 
+const STALE_LOCK_ERROR_MESSAGE = 'Sync timed out (stale lock recovered)'
+
+/**
+ * The reclaimed connector's new consecutive-failure count.
+ *
+ * A hard kill (OOM/SIGKILL) skips `executeSync`'s `catch` and `finally`
+ * entirely, so this reaper is the ONLY writer that ever observes that failure.
+ * Computed in SQL rather than read-then-written because two overlapping cron
+ * ticks reclaiming the same row would otherwise both read the same value and
+ * write the same increment, losing one.
+ */
+function reclaimedFailureCount(): SQL {
+  return sql`COALESCE(${knowledgeConnector.consecutiveFailures}, 0) + 1`
+}
+
+/** Disables the connector once the reclaimed count reaches the shared threshold. */
+function reclaimedStatus(): SQL {
+  return sql`CASE WHEN COALESCE(${knowledgeConnector.consecutiveFailures}, 0) + 1 >= ${MAX_CONSECUTIVE_FAILURES} THEN 'disabled' ELSE 'error' END`
+}
+
+/**
+ * The reclaimed connector's next attempt, on the shared failure ladder
+ * (`connectorFailureBackoffMinutes`). A disabled connector gets no next attempt.
+ */
+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`
+}
+
 /**
  * Cron endpoint that checks for connectors due for sync and dispatches sync jobs.
  * Should be called every 5 minutes by an external cron service.
@@ -45,10 +78,11 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     const recoveredConnectors = await db
       .update(knowledgeConnector)
       .set({
-        status: 'error',
-        lastSyncError: 'Sync timed out (stale lock recovered)',
-        nextSyncAt: new Date(now.getTime() + 10 * 60 * 1000),
-        updatedAt: now,
+        status: reclaimedStatus(),
+        lastSyncError: STALE_LOCK_ERROR_MESSAGE,
+        nextSyncAt: reclaimedNextSyncAt(),
+        consecutiveFailures: reclaimedFailureCount(),
+        updatedAt: sql`now()`,
       })
       .where(
         and(
@@ -67,6 +101,45 @@ 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.
+     *
+     * Safe on liveness: the run ceiling is
+     * {@link CONNECTOR_SYNC_MAX_DURATION_SECONDS} and this TTL is twice that, so
+     * a row older than the cutoff belongs to a run the platform has already
+     * killed and which cannot still be writing. The predicate is 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)
+        )
+      )
+      .returning({ id: knowledgeConnectorSyncLog.id })
+
+    if (closedSyncLogs.length > 0) {
+      logger.warn(`[${requestId}] Closed ${closedSyncLogs.length} orphaned connector sync log(s)`)
+    }
+
     const dueConnectors = await db
       .select({
         id: knowledgeConnector.id,