fix(tables,knowledge): recover abandoned dispatches and bound the sweep

Three defects measured in production this afternoon.

A dispatcher killed by an OOM left `table_run_dispatches` at `dispatching`
forever. Every terminal transition on that table is user- or flow-initiated, so
nothing reclaimed the row: four dispatches were stranded in one afternoon,
pinning each table's "X running" overlay and blocking re-runs, with no way to
clear them from the product. The `table_run_dispatches_watchdog_idx` index has
existed for this sweep since the table was created, unused.

Liveness comes from a new `heartbeat_at`, stamped by the per-window writes that
already advance `cursor` and `processed_count`, so a slow-but-live dispatch is
spared however long it runs — the in-process path has no duration ceiling, so
ageing from `requested_at` would reclaim live self-hosted work. The sweep reads
`COALESCE(heartbeat_at, requested_at)` so rows written before the column stay
reclaimable rather than NULL-false forever, and runs as the last arm of the
existing stale-execution cron at the same 95-minute window its table-job sibling
uses. Rows are cancelled, not completed: the scope never finished.

The OOM itself is not a leak. Peak RSS is a flat plateau — 457 MB at 20-45s and
461 MB past 200s, so ten times the duration buys four megabytes — that has crept
about two percent per release for a month, from 446 MB in late July to 545 MB,
past the 512 MiB `small-1x` ceiling. CPU peaks at 0.19, so the larger preset is
bought for RAM alone. `maxAttempts` never covered the kill either: Trigger.dev
retries `TASK_PROCESS_OOM_KILLED` only when `retry.outOfMemory.machine` names a
preset, and all four runs recorded `attempt_count = 1` while the docstring
claimed they resumed from the persisted cursor.

The connector stuck-document sweep dispatched without a bound. Its chunk size
paced the loop but the candidate query had no limit, so one connector enqueued
2,959 documents in fifteen seconds onto the queue every workspace shares.
Nothing was double-billed — those documents were genuinely unindexed — but one
connector monopolized the queue, and each dispatch mints a fresh requestId, so
the idempotency key differs every pass and none of it deduplicates. Candidates
are now taken oldest-first and capped per sync; a deeper backlog is deferred to
the next sync rather than dropped.

Author: waleedlatif1

apps/sim/app/api/cron/cleanup-stale-executions/route.ts

@@ -31,6 +31,7 @@ import {
   STALE_SWEEPABLE_EXECUTION_STATUSES,
   type StaleSweepableExecutionStatus,
 } from '@/lib/logs/types'
+import { cancelStaleDispatches } from '@/lib/table/dispatcher'
 import { deleteFile } from '@/lib/uploads/core/storage-service'
 import {
   carrierNotIrrecoverableSql,
@@ -52,6 +53,16 @@ const EXECUTION_DEADLINE_ERROR = getTimeoutErrorMessage(undefined)
 const TABLE_JOB_STALE_THRESHOLD_MINUTES = 95
 /** Terminal table-jobs older than this are pruned; only the latest job per table is ever read. */
 const TABLE_JOB_RETENTION_HOURS = 24
+/**
+ * A table run dispatch whose holder has not made progress for this long is
+ * treated as dead. Same shape and window as the table-job threshold above: the
+ * 90-minute Trigger.dev task ceiling (`maxDuration` in `trigger.config.ts`) plus
+ * five minutes of cleanup grace, measured from the dispatcher's own per-window
+ * heartbeat rather than from when the run was requested.
+ */
+const TABLE_DISPATCH_STALE_THRESHOLD_MINUTES = 95
+/** Per-run ceiling on reaped dispatches, so one tick cannot fan out unbounded SSE. */
+const TABLE_DISPATCH_MAX_PER_RUN = 200
 /**
  * Terminal deployment operations older than this are pruned. Every reader of
  * this table is latest-generation-only, and idempotency keys only need to
@@ -144,6 +155,9 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     const staleTableJobThreshold = new Date(
       now.getTime() - TABLE_JOB_STALE_THRESHOLD_MINUTES * 60 * 1000
     )
+    const staleDispatchThreshold = new Date(
+      now.getTime() - TABLE_DISPATCH_STALE_THRESHOLD_MINUTES * 60 * 1000
+    )
 
     let staleExecutionsFound = 0
     let cleaned = 0
@@ -604,6 +618,29 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       })
     }
 
+    /**
+     * Cancel table run dispatches abandoned by a dead dispatcher. Nothing else
+     * reclaims them — every other terminal transition is user- or flow-initiated
+     * — so a dispatcher killed mid-loop left the row `dispatching` forever and
+     * the client's "X running" overlay with it. Ages from the dispatcher's
+     * per-window heartbeat, so a slow-but-live dispatch is spared.
+     */
+    let staleDispatchesCancelled = 0
+    try {
+      staleDispatchesCancelled = (
+        await cancelStaleDispatches(staleDispatchThreshold, TABLE_DISPATCH_MAX_PER_RUN)
+      ).length
+      if (staleDispatchesCancelled > 0) {
+        logger.warn(`Cancelled ${staleDispatchesCancelled} abandoned table run dispatches`, {
+          thresholdMinutes: TABLE_DISPATCH_STALE_THRESHOLD_MINUTES,
+        })
+      }
+    } catch (error) {
+      logger.error('Failed to cancel abandoned table run dispatches:', {
+        error: toError(error).message,
+      })
+    }
+
     return NextResponse.json({
       success: true,
       executions: {
@@ -622,6 +659,10 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       tableJobs: {
         staleMarkedFailed: staleTableJobsMarkedFailed,
       },
+      tableRunDispatches: {
+        staleCancelled: staleDispatchesCancelled,
+        thresholdMinutes: TABLE_DISPATCH_STALE_THRESHOLD_MINUTES,
+      },
       deploymentOperations: {
         pruned: deploymentOperationsPruned,
         retentionDays: DEPLOYMENT_OPERATION_RETENTION_DAYS,

apps/sim/background/table-run-dispatcher.test.ts

@@ -0,0 +1,37 @@
+/**
+ * @vitest-environment node
+ */
+import { describe, expect, it, vi } from 'vitest'
+
+const { mockTask } = vi.hoisted(() => ({
+  mockTask: vi.fn((config) => config),
+}))
+
+vi.mock('@trigger.dev/sdk', () => ({ task: mockTask }))
+vi.mock('@/lib/table/dispatcher', () => ({
+  runDispatcherToCompletion: vi.fn(),
+}))
+
+import { tableRunDispatcherTask } from '@/background/table-run-dispatcher'
+
+describe('table-run-dispatcher task configuration', () => {
+  /**
+   * Peak RSS is a flat 457-464 MB plateau independent of run length, and it has
+   * crept ~2% per release — 446 MB in late July to 545 MB, past the 512 MiB
+   * `small-1x` ceiling, which killed four runs in one afternoon.
+   */
+  it('runs on a preset whose memory clears the observed plateau', () => {
+    expect(tableRunDispatcherTask.machine).toBe('small-2x')
+  })
+
+  /**
+   * `maxAttempts` alone does NOT cover `TASK_PROCESS_OOM_KILLED` — Trigger.dev
+   * retries an OOM only when `retry.outOfMemory.machine` names a larger preset.
+   * Every one of the four killed runs recorded `attempt_count = 1`, so the
+   * documented "retries and resumes from the persisted cursor" never happened.
+   */
+  it('escalates to a larger machine on an out-of-memory kill', () => {
+    expect(tableRunDispatcherTask.retry?.outOfMemory?.machine).toBe('medium-1x')
+    expect(tableRunDispatcherTask.retry?.maxAttempts).toBe(3)
+  })
+})

apps/sim/background/table-run-dispatcher.ts

@@ -17,13 +17,28 @@ export interface TableRunDispatcherPayload {
  * dispatcher loop for the dispatch's entire lifetime — each iteration
  * processes a window of cells via `batchTriggerAndWait`, which checkpoints
  * the parent via CRIU during the wait so we don't pay compute while cells
- * execute. The cursor is persisted in DB; if this run crashes, trigger.dev
- * retries and the next attempt resumes from the persisted cursor.
+ * execute. The cursor is persisted in DB, so an attempt that starts after a
+ * crash resumes from it rather than replaying the dispatch.
+ *
+ * `maxAttempts` alone does NOT cover an OOM: Trigger.dev retries
+ * `TASK_PROCESS_OOM_KILLED` only when `retry.outOfMemory.machine` names a
+ * larger preset. Four runs were killed this way and every one recorded
+ * `attempt_count = 1` — no retry happened, and the dispatch row was left
+ * `dispatching` forever. The escalating preset is what makes the documented
+ * resume actually reachable; the cleanup sweep is the backstop for a dispatch
+ * whose holder dies without one.
  */
 export const tableRunDispatcherTask = task({
   id: 'table-run-dispatcher',
-  machine: 'small-1x',
-  retry: { maxAttempts: 3 },
+  /**
+   * Memory, not CPU. Peak RSS sits at a flat 457-464 MB plateau regardless of
+   * run length (10x the duration moves it ~4 MB), and it has crept ~2% per
+   * release for a month — 446 MB in late July to 545 MB, past the 512 MiB
+   * `small-1x` ceiling. Meanwhile CPU utilization peaks at 0.19 and sits at
+   * 0.03 for p90, so the larger preset is bought for its RAM.
+   */
+  machine: 'small-2x',
+  retry: { maxAttempts: 3, outOfMemory: { machine: 'medium-1x' } },
   queue: {
     name: 'table-run-dispatcher',
     concurrencyLimit: 8,

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

@@ -2074,6 +2074,46 @@ describe('executeSync hard-delete reconciliation', () => {
     expect(calls.flatMap((call) => call[0] as string[])).toEqual(missingIds)
   })
 
+  it('bounds and orders the stuck-document sweep instead of draining a backlog at once', async () => {
+    const { executeSync, STUCK_RETRY_MAX_CANDIDATES_PER_SYNC } = await import(
+      '@/lib/knowledge/connectors/sync-engine'
+    )
+    const { hardDeleteDocuments } = await import('@/lib/knowledge/documents/service')
+
+    primeReconciliation()
+    vi.mocked(hardDeleteDocuments).mockResolvedValue(0)
+    /**
+     * Every batch and the post-batch check re-read the connector to confirm the
+     * sync target still exists; without enough rows the run exits as
+     * connector-deleted before the sweep it is meant to exercise.
+     */
+    for (let i = 0; i < 40; i++) {
+      queueTableRows(schemaMock.knowledgeConnector, [
+        { connectorArchivedAt: null, connectorDeletedAt: null, kbDeletedAt: null },
+      ])
+    }
+    // The sweep's own candidate read: no stuck documents, so it dispatches none.
+    queueTableRows(schemaMock.document, [])
+
+    await executeSync('c-1', {
+      billingAttribution: { workspaceId: 'ws-1' } as never,
+      fullSync: true,
+    })
+
+    /**
+     * The dispatch loop's chunk size paced the sweep but never bounded it — the
+     * candidate query had no limit, so one connector enqueued its entire backlog
+     * (2,959 documents in fifteen seconds) onto the queue every workspace shares.
+     */
+    expect(dbChainMockFns.limit).toHaveBeenCalledWith(STUCK_RETRY_MAX_CANDIDATES_PER_SYNC)
+    /**
+     * Ordered, so the bound takes the most overdue documents first and can never
+     * starve one indefinitely. An unordered limit takes an arbitrary subset each
+     * sync, which is a cap that silently loses work rather than deferring it.
+     */
+    expect(dbChainMockFns.orderBy).toHaveBeenCalled()
+  })
+
   it('releases the lock when it errors a connector whose knowledge base is gone', async () => {
     const { executeSync } = await import('@/lib/knowledge/connectors/sync-engine')
 

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

@@ -10,7 +10,20 @@ import { createLogger } from '@sim/logger'
 import { getErrorMessage, toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { randomInt } from '@sim/utils/random'
-import { and, desc, eq, exists, gt, inArray, isNotNull, isNull, lt, ne, sql } from 'drizzle-orm'
+import {
+  and,
+  asc,
+  desc,
+  eq,
+  exists,
+  gt,
+  inArray,
+  isNotNull,
+  isNull,
+  lt,
+  ne,
+  sql,
+} from 'drizzle-orm'
 import { decryptApiKey } from '@/lib/api-key/crypto'
 import {
   assertBillingAttributionSnapshot,
@@ -94,6 +107,25 @@ const MAX_SAFE_TITLE_LENGTH = 200
  */
 const STUCK_RETRY_DISPATCH_CHUNK_SIZE = 25
 
+/**
+ * How many stuck-document candidates one sync will consider.
+ *
+ * {@link STUCK_RETRY_DISPATCH_CHUNK_SIZE} paces the dispatch loop but does not
+ * bound it — the candidate query had no limit, so a connector carrying a large
+ * backlog dispatched the whole thing at once. One did: 2,959 documents enqueued
+ * in fifteen seconds onto a queue every workspace shares, at
+ * {@link PROCESSING_QUEUE_CONCURRENCY} concurrent runs. Nothing was
+ * double-billed — those documents were genuinely unindexed — but one connector
+ * monopolized the queue, and each dispatch mints a fresh `requestId`, so the
+ * Trigger.dev idempotency key differs every pass and none of it deduplicates.
+ *
+ * 200 keeps a single sync's contribution to roughly ten minutes of queue
+ * occupancy at the default concurrency. A backlog larger than this is not
+ * dropped: candidates are taken oldest-first and whatever is left stays
+ * eligible, so consecutive syncs drain it steadily instead of in one burst.
+ */
+export const STUCK_RETRY_MAX_CANDIDATES_PER_SYNC = 200
+
 /**
  * How many documents reconciliation hard-deletes per call.
  *
@@ -2312,6 +2344,13 @@ export async function executeSync(
           isNull(document.deletedAt)
         )
       )
+      /**
+       * Oldest first, so the most overdue documents drain before newer ones and
+       * the bound below can never starve a document indefinitely. Without an
+       * order the limit would take an arbitrary subset each sync.
+       */
+      .orderBy(asc(document.uploadedAt))
+      .limit(STUCK_RETRY_MAX_CANDIDATES_PER_SYNC)
     const stuckDocs = sweepCandidates
       .filter((row): row is typeof row & { processingStatus: DocumentProcessingStatus } =>
         isDocumentProcessingStatus(row.processingStatus)

apps/sim/lib/table/dispatcher.ts

@@ -407,7 +407,10 @@ export async function dispatcherStep(
   if (dispatch.status === 'pending') {
     await db
       .update(tableRunDispatches)
-      .set({ status: 'dispatching' })
+      // Opens the heartbeat at the instant a holder takes the dispatch, so the
+      // cleanup sweep ages it from that rather than from `requested_at`, which
+      // was stamped when the run was merely requested.
+      .set({ status: 'dispatching', heartbeatAt: new Date() })
       .where(eq(tableRunDispatches.id, dispatchId))
     // Announce the dispatch the moment it starts — before the first window's
     // cells finish. Without this, auto-fired and capped dispatches (no client-
@@ -672,7 +675,10 @@ export async function dispatcherStep(
 async function incrementProcessedCount(dispatchId: string, delta: number): Promise<void> {
   await db
     .update(tableRunDispatches)
-    .set({ processedCount: sql`${tableRunDispatches.processedCount} + ${delta}` })
+    .set({
+      processedCount: sql`${tableRunDispatches.processedCount} + ${delta}`,
+      heartbeatAt: new Date(),
+    })
     .where(eq(tableRunDispatches.id, dispatchId))
 }
 
@@ -724,7 +730,7 @@ async function stampQueuedForBatch(
 async function advanceCursor(dispatchId: string, newCursor: number): Promise<void> {
   await db
     .update(tableRunDispatches)
-    .set({ cursor: newCursor })
+    .set({ cursor: newCursor, heartbeatAt: new Date() })
     .where(eq(tableRunDispatches.id, dispatchId))
 }
 
@@ -781,6 +787,101 @@ export async function completeDispatchIfActive(dispatchId: string): Promise<bool
   return transitioned.length > 0
 }
 
+/**
+ * Cancels dispatches whose holder died without reaching a terminal state.
+ *
+ * `table_run_dispatches` had no reaper: the only paths to a terminal status are
+ * user- or flow-initiated ({@link cancelDispatchById},
+ * {@link completeDispatchIfActive}, {@link markActiveDispatchesCancelled}), so a
+ * dispatcher killed mid-loop left its row `dispatching` forever — pinning the
+ * client's "X running" overlay and blocking re-runs of that table. Four rows
+ * were stranded this way by a single afternoon of OOM kills, and nothing in the
+ * product could clear them. The `table_run_dispatches_watchdog_idx` index has
+ * existed since the table was created for exactly this sweep, unused.
+ *
+ * Liveness comes from `heartbeatAt`, which the per-window `advanceCursor` and
+ * `incrementProcessedCount` writes stamp, so a slow-but-live dispatch is spared
+ * however long it runs. That matters because the in-process path
+ * (`isTriggerDevEnabled === false`) has no duration ceiling at all — ageing from
+ * `requestedAt` would reclaim live self-hosted work. `COALESCE` keeps rows
+ * written before the column existed reclaimable rather than NULL-false forever.
+ *
+ * Cancelled rather than completed: the dispatch did not finish its scope, and
+ * reporting it complete would tell the user work happened that did not. The
+ * cursor is left in place, so a re-run resumes rather than replays.
+ */
+export async function cancelStaleDispatches(
+  staleBefore: Date,
+  limit: number
+): Promise<DispatchRow[]> {
+  const isStale = () =>
+    and(
+      inArray(tableRunDispatches.status, [...ACTIVE_DISPATCH_STATUSES]),
+      sql`COALESCE(${tableRunDispatches.heartbeatAt}, ${tableRunDispatches.requestedAt}) < ${sql.param(staleBefore, tableRunDispatches.heartbeatAt)}`
+    )
+
+  // Claimed as explicit ids first, then updated by id, so the bound is evaluated
+  // exactly once — the pattern every other bulk cleanup in this codebase uses.
+  // The update re-asserts staleness, so a dispatch that finished in between is
+  // left alone rather than cancelled out from under its own terminal write.
+  const claimed = await db
+    .select({ id: tableRunDispatches.id })
+    .from(tableRunDispatches)
+    .where(isStale())
+    .limit(limit)
+  if (claimed.length === 0) return []
+
+  const cancelled = await db
+    .update(tableRunDispatches)
+    .set({ status: 'cancelled', cancelledAt: new Date() })
+    .where(
+      and(
+        isStale(),
+        inArray(
+          tableRunDispatches.id,
+          claimed.map(({ id }) => id)
+        )
+      )
+    )
+    .returning()
+
+  const dispatches = cancelled.map((row) => ({
+    id: row.id,
+    tableId: row.tableId,
+    workspaceId: row.workspaceId,
+    requestId: row.requestId,
+    mode: row.mode as DispatchMode,
+    scope: row.scope as DispatchScope,
+    status: 'cancelled' as DispatchStatus,
+    cursor: row.cursor,
+    limit: (row.limit as DispatchLimit | null) ?? null,
+    processedCount: row.processedCount,
+    isManualRun: row.isManualRun,
+    triggeredByUserId: row.triggeredByUserId,
+    requestedAt: row.requestedAt,
+  }))
+
+  // Same terminal event every other cancel path emits — without it the row goes
+  // terminal in the database while the client overlay stays stuck, which is the
+  // symptom this function exists to clear.
+  await Promise.all(
+    dispatches.map((d) =>
+      appendTableEvent({
+        kind: 'dispatch',
+        tableId: d.tableId,
+        dispatchId: d.id,
+        status: 'cancelled',
+        scope: d.scope,
+        cursor: d.cursor,
+        mode: d.mode,
+        isManualRun: d.isManualRun,
+      })
+    )
+  )
+
+  return dispatches
+}
+
 /** Mark every active dispatch on this table as cancelled. Single atomic
  *  UPDATE so the dispatcher's next iteration observes the cancel. Returns the
  *  dispatches that were cancelled so the caller can emit per-dispatch SSE