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

Author: waleedlatif1

apps/sim/app/api/copilot/feedback/route.test.ts

@@ -357,7 +357,7 @@ edges:
 
       const { eq } = await import('drizzle-orm')
       expect(dbChainMockFns.where).toHaveBeenCalled()
-      expect(eq).toHaveBeenCalledWith('userId', 'user-123')
+      expect(eq).toHaveBeenCalledWith('copilotFeedback.userId', 'user-123')
     })
 
     it('should handle database errors gracefully', async () => {

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

@@ -1,15 +1,21 @@
 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_AUTO_DISABLED_ERROR,
+  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 +30,99 @@ 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)'
+
+/**
+ * How long the connector holding the lock has gone without proving it is alive.
+ *
+ * `sync_lock_lease_at` is written only by lock acquisition and the heartbeat, so
+ * it is the lease; `updated_at` is the row's modification time and merely used
+ * to double as one. Read through `COALESCE` rather than backfilled: a plain
+ * `lease <= cutoff` is NULL-false, so a row already `syncing` when this column
+ * shipped would never be reclaimed — strictly worse than the behaviour it
+ * replaces. The fallback also keeps the reaper correct against any future
+ * writer that takes the lock without opening a lease.
+ */
+function syncLockLease(): SQL {
+  return sql`COALESCE(${knowledgeConnector.syncLockLeaseAt}, ${knowledgeConnector.updatedAt})`
+}
+
+/**
+ * The error a reclaimed connector reports.
+ *
+ * Mirrors {@link reclaimedStatus}: once the reclaim disables the connector,
+ * {@link reclaimedNextSyncAt} sets no next attempt, so telling the operator the
+ * sync merely timed out describes a retry that will never happen. The disabled
+ * wording is the shared one `buildSyncFailureUpdate` writes, so the in-process
+ * 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`
+}
+
+/**
+ * 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`, which reads only `completed` rows.
+ *
+ * The heartbeat is the single source of liveness truth, so this defers to it,
+ * reading the same {@link syncLockLease} expression the reclaim predicate does.
+ * 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 — its lease 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 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 ${syncLockLease()} > ${sql.param(staleCutoff, knowledgeConnector.syncLockLeaseAt)}
+  )`
+}
+
+/**
+ * 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,15 +144,21 @@ 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: 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'),
-          lte(knowledgeConnector.updatedAt, staleCutoff),
+          sql`${syncLockLease()} <= ${sql.param(staleCutoff, knowledgeConnector.syncLockLeaseAt)}`,
           isNull(knowledgeConnector.archivedAt),
           isNull(knowledgeConnector.deletedAt)
         )
@@ -67,6 +172,47 @@ 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)
+        )
+      )
+      .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,

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

@@ -137,6 +137,9 @@ describe('Knowledge Utils', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     resetDbChainMock()
+    // The document claim gates on the row it writes back, so an unstubbed
+    // `returning()` would abort processing before any completion write.
+    dbChainMockFns.returning.mockResolvedValue([{ id: 'doc1' }])
     // `unstubGlobals: true` removes the module-scope fetch stub after the
     // first test in the worker; re-stub it per test.
     vi.stubGlobal('fetch', createEmbeddingFetchMock())

apps/sim/app/api/knowledge/utils.test.ts

@@ -352,7 +352,7 @@ describe('POST /api/mothership/chats/[chatId]/fork', () => {
     // they vanish from VFS listings and name resolution…
     expect(dbChainMockFns.where).toHaveBeenCalledWith({
       type: 'inArray',
-      column: 'id',
+      column: 'workspaceFiles.id',
       values: ['wf_dead1', 'wf_dead2'],
     })
     // …and their resource chips are dropped from the new chat.

apps/sim/app/api/mothership/chats/[chatId]/fork/route.test.ts

@@ -52,8 +52,8 @@ describe('POST /api/mothership/chats/read', () => {
     expect(orClause).toBeDefined()
     expect(orClause?.conditions).toEqual(
       expect.arrayContaining([
-        { type: 'isNull', column: 'lastSeenAt' },
-        { type: 'lt', left: 'lastSeenAt', right: 'updatedAt' },
+        { type: 'isNull', column: 'copilotChats.lastSeenAt' },
+        { type: 'lt', left: 'copilotChats.lastSeenAt', right: 'copilotChats.updatedAt' },
       ])
     )
   })

apps/sim/app/api/mothership/chats/read/route.test.ts

@@ -385,7 +385,9 @@ describe('time-pause resume admission', () => {
     expect(
       inArrayMock.mock.calls.some(
         ([column, values]) =>
-          column === 'id' && Array.isArray(values) && values.join(',') === 'paused-2,paused-3'
+          column === 'pausedExecutions.id' &&
+          Array.isArray(values) &&
+          values.join(',') === 'paused-2,paused-3'
       )
     ).toBe(true)
     expect(executionSnapshotFromJsonMock).toHaveBeenCalledTimes(2)
@@ -537,7 +539,7 @@ describe('time-pause resume admission', () => {
       [LEGACY_PAUSED_SNAPSHOT_FALLBACK_CHUNK_SIZE],
     ])
     const snapshotIdBatches = inArrayMock.mock.calls
-      .filter(([column]) => column === 'id')
+      .filter(([column]) => column === 'pausedExecutions.id')
       .map(([, ids]) => ids as string[])
     expect(snapshotIdBatches.map((ids) => ids.length)).toEqual([10, 4, 4, 2])
     expect(

apps/sim/app/api/resume/poll/route.test.ts

@@ -4,6 +4,10 @@ import type {
   V2KnowledgeTaggedDocument,
 } from '@/lib/api/contracts/v2/knowledge'
 import { ALL_TAG_SLOTS, type AllTagSlot } from '@/lib/knowledge/constants'
+import {
+  DOCUMENT_PROCESSING_STATUSES,
+  type DocumentProcessingStatus,
+} from '@/lib/knowledge/documents/types'
 import type { DocumentTagDefinition } from '@/lib/knowledge/tags/types'
 import type { KnowledgeBaseWithCounts } from '@/lib/knowledge/types'
 import { getUserEmailsByIds, requireResolvedUserEmail } from '@/lib/users/queries'
@@ -39,9 +43,7 @@ export function toV2DocumentTags(
   return tags
 }
 
-const PROCESSING_STATUSES = ['pending', 'processing', 'completed', 'failed'] as const
-
-type V2DocumentProcessingStatus = (typeof PROCESSING_STATUSES)[number]
+type V2DocumentProcessingStatus = DocumentProcessingStatus
 
 /**
  * Narrows a stored processing status onto the published enum. An absent value
@@ -50,7 +52,7 @@ type V2DocumentProcessingStatus = (typeof PROCESSING_STATUSES)[number]
  */
 function toProcessingStatus(status: string | null | undefined): V2DocumentProcessingStatus {
   if (status === null || status === undefined) return 'pending'
-  const known = PROCESSING_STATUSES.find((candidate) => candidate === status)
+  const known = DOCUMENT_PROCESSING_STATUSES.find((candidate) => candidate === status)
   if (!known) throw new Error(`Unexpected knowledge document processing status: ${status}`)
   return known
 }