chore(db): regenerate the sync lock token migration with drizzle-kit

The migration was hand-written, which left it inconsistent with every other
migration in the repo and, more importantly, without a schema snapshot. Drizzle
diffs against the latest snapshot to decide what a migration needs to contain,
so the next generate would have seen the column as still missing and emitted it
a second time.

Regenerated properly: drizzle-kit now owns the SQL, the journal entry, and
0297_snapshot.json. The emitted statement matches the house pattern for an
additive nullable column, and check:migrations still reports backward-compatible.

Author: waleedlatif1

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

@@ -1332,7 +1332,7 @@ describe('writeTerminalConnectorState', () => {
   })
 })
 
-describe('applySupersededOutcome', () => {
+describe('markSyncSuperseded', () => {
   const result = {
     docsAdded: 3,
     docsUpdated: 1,
@@ -1341,29 +1341,28 @@ describe('applySupersededOutcome', () => {
     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(
+    const { markSyncSuperseded, 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)
+    expect(markSyncSuperseded(result).error).toBe(SUPERSEDED_SYNC_ERROR)
   })
 
   it('preserves the document counters of the discarded run', async () => {
-    const { applySupersededOutcome } = await import('@/lib/knowledge/connectors/sync-engine')
+    const { markSyncSuperseded } = await import('@/lib/knowledge/connectors/sync-engine')
 
     // Those writes landed — only the connector-level bookkeeping was discarded.
-    expect(applySupersededOutcome(result, false)).toMatchObject(result)
+    expect(markSyncSuperseded(result)).toMatchObject(result)
+  })
+
+  it('does not mutate the result it was handed', async () => {
+    const { markSyncSuperseded } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    markSyncSuperseded(result)
+
+    expect(result).not.toHaveProperty('error')
   })
 })
 

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

@@ -364,9 +364,11 @@ export async function completeSyncLog(
  * 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.
+ * Guards every write a run makes to its own connector row: both terminal paths
+ * and the mid-run heartbeat. 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 — and reusing it
+ * for the heartbeat is what turns a beat into an ownership probe.
  */
 export function stillHoldsSyncLock(connectorId: string, syncLockToken: string) {
   return and(
@@ -396,7 +398,7 @@ export function buildSyncLockAcquisition(syncLogId: string, now: Date) {
 /**
  * Whether a running sync is due to refresh its lock.
  *
- * Time-based rather than batch-count-based: batches vary hugely in cost, so a
+ * Time-based rather than batch-count-based: batches vary hugely in cost, so an
  * every-N-batches beat would fire constantly on small documents and barely at
  * all on large ones — exactly the runs that need it.
  */
@@ -457,9 +459,10 @@ export async function writeTerminalConnectorState(
 }
 
 /**
- * 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.
+ * Reported when a run loses its connector's lock mid-flight — either because a
+ * heartbeat found the lock reclaimed, or because its terminal write matched no
+ * rows. 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'
 
@@ -468,11 +471,7 @@ export const SUPERSEDED_SYNC_ERROR = 'sync_superseded'
  * 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
+export function markSyncSuperseded(result: SyncResult): SyncResult {
   return { ...result, error: SUPERSEDED_SYNC_ERROR }
 }
 
@@ -595,8 +594,8 @@ export function classifySuspectListing(
  * immediately. A genuinely emptied source keeps reconciling: its second sync
  * corroborates the first and tombstones everything, and a later sync — once the
  * tombstoned set is again absent — completes the two-strike purge, subject to
- * {@link capReconciliationDeletions}, which holds a pass whose deletion count
- * exceeds the per-sync blast-radius cap.
+ * {@link capReconciliationDeletions}, which withholds any generation whose
+ * deletion count exceeds the per-sync blast-radius cap.
  *
  * A forced `fullSync` overrides the guard, matching its existing meaning
  * elsewhere here — an explicit human request to reconcile against this listing
@@ -2033,7 +2032,7 @@ export async function executeSync(
         syncLogId,
         ...result,
       })
-      return applySupersededOutcome(result, false)
+      return markSyncSuperseded(result)
     }
 
     logger.info('Sync completed', { connectorId, ...result })
@@ -2050,7 +2049,7 @@ export async function executeSync(
         syncLogId,
         ...result,
       })
-      return applySupersededOutcome(result, false)
+      return markSyncSuperseded(result)
     }
 
     if (error instanceof ConnectorDeletedException) {
@@ -2108,7 +2107,7 @@ export async function executeSync(
       )
 
       /**
-       * Deliberately does NOT get {@link applySupersededOutcome}. `result.error`
+       * Deliberately does NOT get {@link markSyncSuperseded}. `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

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

@@ -38,8 +38,8 @@ export const CONNECTOR_SYNC_STALE_LOCK_TTL_MS = CONNECTOR_SYNC_MAX_DURATION_SECO
  *
  * Shared because two independent writers advance this counter: `executeSync`'s
  * in-process failure path, and the scheduler's out-of-process stale-lock
- * reclaim (a SIGKILL skips `catch`/`finally`, so only the reaper ever sees that
- * failure). A connector that only ever dies hard must still reach the threshold,
+ * reclaim (a SIGKILL unwinds nothing, so the in-process `catch` never runs and
+ * only the reaper ever sees that failure). A connector that only ever dies hard must still reach the threshold,
  * which it cannot if the two disagree on what the threshold is.
  */
 export const MAX_CONSECUTIVE_FAILURES = 10

packages/db/migrations/0297_connector_sync_lock_token.sql

@@ -0,0 +1 @@
+ALTER TABLE "knowledge_connector" ADD COLUMN "sync_lock_token" text;