fix(knowledge,tables): recover abandoned dispatches, bound the sweep and the workbook preview (#6945)

* 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.

* fix(file-parsers): read officeparser's entry point across module systems

`officeparser` is CommonJS — `main: officeParser.js`, no `type`, no `exports`
map — so what `await import('officeparser')` yields depends on who built the
code. Node and webpack synthesize named exports from `module.exports`, so
`.parseOfficeAsync` is there. esbuild, which builds the Trigger.dev worker
bundle, puts `module.exports` on `.default` and leaves the named export
undefined, and the package is in neither `build.external` nor
`additionalPackages`, so it is bundled.

Reading the named export directly therefore worked everywhere except the
worker, where calling it threw `TypeError: parseOfficeAsync is not a function`.
All four parsers treat that as "the library failed" and answer with a scrape of
the archive, which returns `degraded: true`, and the document pipeline rejects a
degraded parse outright. The visible result was every `.pptx` and legacy `.doc`
reporting "No text could be extracted from this file — it may be scanned,
image-only, or password-protected", naming a cause that had nothing to do with
the fault. 118 pptx and 14 doc failures landed in a single burst when one
connector's sync first succeeded after ten consecutive crashes.

Resolved in one shared loader rather than per bundler: externalizing the package
has to be repeated in every build config this code runs under and regresses
silently the day one is missed.

The shape handling is split into a pure `resolveParseOfficeAsync` because the
failing shape cannot be reproduced by mocking the specifier — Vitest's
module-namespace proxy throws on a missing export rather than yielding the
`undefined` a real bundle produces, so a test going through `import` can only
assert the shape that already worked. That is also why the existing parser
suites never caught this: each mocks `officeparser` with a fabricated named
export, which presupposes the interop being broken here.

* fix(knowledge): bound the workbook preview to the rows it emits

`sheet_to_json` allocates from a worksheet's DECLARED `!ref` range rather than
its populated cells, and Excel routinely writes an inflated range from stray
formatting. The 1,000-row preview cap was applied to the result, so it bounded
the emitted string while the allocation it was meant to bound had already
happened. An 880 KB workbook exhausted an 8 GB worker; the same content
exhausted 16 GB when this ran inside the connector sync. No machine size fixes
that, because the allocation scales with a number the file declares about
itself — fleet p99 for this task is 691 MB against 8 GB, so this is a cliff, not
pressure.

Passing the window into the conversion is what makes the cap real. `defval` goes
with it: defaulting every cell in the range made each row dense, so allocation
scaled with columns x declared rows rather than with populated cells, and
because no row was left empty it silently defeated the `blankrows: false` beside
it. Reported totals still come from the declared range, so bounding the
conversion does not change what the metadata says the workbook holds.

The eleven documents killed this way recorded `attempt_count = 1`: `maxAttempts`
does not cover `TASK_PROCESS_OOM_KILLED`, which Trigger.dev retries only when a
larger preset is named. Adding that escalation is a safety net rather than the
fix, and the same gap the dispatcher had.

Also corrects the machine comment, which claimed `large-1x` was 2 vCPU / 2 GB.
It is 4 vCPU / 8 GB, and believing the stale figure makes a resize look like the
answer when the parser is what is unbounded.

* fix(tables): keep a cancelled dispatch cancelled when a step claims it

`dispatcherStep` reads the dispatch, then awaits the table load before writing
`dispatching`. Keying that write on the id alone resurrected a dispatch
cancelled inside that window — a Stop-all, or now the stale-dispatch sweep —
and the fresh heartbeat it writes would then buy the resurrected row another
full window before the sweep could reclaim it again.

The race predates the sweep, but the sweep is a new writer of `cancelled` that
no user action drives, so it is newly reachable without anyone touching Stop.
Re-asserting the status the step already read is the whole fix.

* fix(tables,knowledge): spare a live window, and restore the truncation notice

A lease needs its heartbeat interval to sit well under its TTL. The dispatch
heartbeat is stamped between windows, not during them, and `batchTriggerAndWait`
checkpoints the loop for the whole window — so the interval is really "one
window", which nothing bounds: the window ends when its cells do, and the
in-process path has no ceiling at all. A window outliving the stale threshold
had its dispatch cancelled while it was plainly alive.

Its cells carry the signal the checkpointed parent cannot — `updatedAt` on every
in-flight row execution, written by the cell tasks themselves. Both signals must
be stale before a dispatch is reclaimed, so a slow window is spared for as long
as its cells keep reporting while a run with nothing beating and nothing
executing is still collected. The subquery rides the partial `(table_id,
status)` index that already covers exactly those three statuses.

Bounding the workbook conversion also made its truncation notice unreachable:
the converted length can no longer exceed the window it was compared against, so
every sheet larger than the preview cap silently stopped reporting that it had
been cut. Compared against the declared row count instead, which is what the
comparison meant before the conversion was bounded.

* fix(tables,knowledge): act on the claim outcome and scope liveness to the dispatch

Three defects, two of them created by the previous round's fixes.

Guarding the pending-to-dispatching claim without reading its outcome was the
worse half of a fix. When a Stop-all or the stale sweep won the race the row
correctly stayed `cancelled`, while the step went on to announce `dispatching`,
stamp cells and enqueue a window for it — and an empty window would then reach
the unguarded `markDispatchComplete` and overwrite `cancelled` with `complete`.
The step now ends when it did not claim the row.

The cell-liveness probe was table-scoped, and `table_row_executions` carries no
dispatch column, so a live dispatch's cells vouched for an abandoned dispatch
beside it and the abandoned row was never reclaimed — turning the stuck overlay
this sweep exists to clear into a permanent one. Narrowed to the dispatch's own
groups, which it already stores. Two active dispatches over the same groups can
still mask each other, but that is the state `markActiveDispatchesCancelled`
already prevents.

Truncation asks whether the window cut the sheet short — a question about the
declared range against the cap. Comparing the converted length to the cap made
it unreachable once the conversion was bounded; comparing the declared count to
the converted length then reported truncation for any sheet merely containing
blank rows, which are now skipped rather than defaulted into existence.

* fix(tables): scope dispatch liveness to its rows, not just its groups

The previous round narrowed the cell-liveness probe to the dispatch's groups on
the reasoning that two active dispatches over the same groups cannot coexist,
because starting a run cancels prior work on its scope. That reasoning was
wrong. `cancelPriorRuns` in `workflow-columns` requires `isManualRun`, so
auto-fired runs never cancel anything, and the per-row path is explicitly a
no-op for dispatch cancellation. Same-group coexistence is ordinary.

A dispatch that names rows now only accepts liveness from those rows, which
covers the auto-fired and row-scoped runs that reach this state. What remains is
two table-wide dispatches over the same groups, where nothing in the row
execution says whose work it is; closing that needs a `dispatch_id` column on
`table_row_executions` threaded through six write sites, including the shared
cell-write path every cell task uses. That residue is a delay rather than a
permanent mask — the live dispatch's cells stop updating when it finishes, and
the next sweep after a quiet window reclaims the abandoned row.

* refactor(tables): name the dispatch liveness predicate and bound its fan-out

Extracts the cell-activity check into `hasRecentCellActivity`, so the stale
predicate reads as its two conditions — nothing beating, nothing executing —
rather than a twenty-line SQL blob nested inside an `and()`. No behaviour
change; this is the code three review rounds found defects in, and being able
to read it is what makes those defects findable.

Bounds the terminal-event fan-out with `mapWithConcurrency`, matching how the
scheduler already fans out. The sibling cancel paths emit over one table's
dispatches; this sweep can carry a whole tick's worth across many tables, and
each event is its own write.

Also repairs the test that covers it. `collectChunks` walks into the
`tableRowExecutions` table object the fragment interpolates, so every column
name appears in the chunks whether the predicate references it or not — the
group, row, table and timestamp assertions all passed with their predicates
deleted. Matching the literal SQL instead makes them fail, which mutating each
clause now confirms.

* fix(tables): make the row bypass NULL-safe and guard the post-wait completion

`jsonb_typeof(scope -> 'rowIds') <> 'array'` was the table-wide bypass, but a
table-wide dispatch has no `rowIds`: the extraction is SQL NULL, `jsonb_typeof`
returns NULL, and `NULL <> 'array'` is UNKNOWN rather than TRUE. The bypass
never fired, so no live cell could satisfy the probe and the sweep reclaimed
exactly the long-running table-wide dispatches the row filter was added to
protect — inverting it. `IS DISTINCT FROM` is the NULL-safe form, and the same
pitfall is already handled with `coalesce` in `markActiveDispatchesCancelled`.

`completeDispatch` also wrote through the unguarded `markDispatchComplete`. Both
its callers run AFTER the window's wait, so a Stop-all or the sweep landing
during that wait leaves the row `cancelled` and the write overwrote it with
`complete`, publishing a completion event after the cancellation one. The claim
guard cannot cover this — the cancel arrives long after the claim. It now goes
through `completeDispatchIfActive`, which already exists for exactly this, and
emits nothing when the transition does not land.

* fix(knowledge): give connector sync logs a retention pass

Nothing pruned `knowledge_connector_sync_log`, so it grew by one row per sync
run forever — a connector on a fifteen-minute interval writes about 35,000 rows
a year by itself. That cost lands on `loadPreviousListingObservation`, which
reads the newest `completed` row per connector through an index covering
`connector_id` alone, so every retained row makes the sort behind the
deletion-safety corroboration slower.

Added as another arm of the cleanup cron, batched the same way as its two
sibling prunes. Two `exists` guards are load-bearing rather than defensive: the
newest row per connector always survives, and so does the newest `completed`
one, because that is the row `loadPreviousListingObservation` reconstructs the
previous listing from — and that reconstruction decides whether a suspect
listing is corroborated, i.e. whether reconciliation may delete documents.
Pruning it would silently change deletion behaviour. `started` rows are never
eligible; they are in flight or waiting on the scheduler's own sweep.

* fix(tables): funnel every post-claim completion through the guarded write

The empty-window exit still wrote through the unguarded `markDispatchComplete`,
and it runs after the claim like the other two — so a cancel landing during its
window query was overwritten with `complete`. Shorter window than the two
post-wait exits, same defect, and leaving one of three unguarded is how this
came back twice already.

All three now route through `completeDispatch`, so the guard lives in one place
and covering it once covers every exit. The redundant test for this path went
with it: it could not be made to fail against the mock, and a test that cannot
fail is worse than none — the guard is held by the test on the shared funnel.

* fix(tables): bound how long cell activity may spare a dispatch

The liveness probe cannot tell whose cells it is looking at when two table-wide
dispatches share a group, because `table_row_executions` carries no dispatch
column. On a quiet table that is only a delay — the neighbour finishes and the
next sweep reclaims — but a busy table with continuous auto-fired work can keep
an abandoned dispatch masked indefinitely, which is the stuck overlay this sweep
exists to clear.

A ceiling bounds it: past a day without a heartbeat, a dispatch is reclaimed
whatever its cells are doing. That is safe because a live dispatch stamps its
heartbeat between windows regardless of cell activity, so only a single window
outliving the ceiling could be reclaimed wrongly, and no window lasts a day on
any path — the Trigger.dev run ceiling is ninety minutes.

The real fix is a `dispatch_id` on the executions row. Threading it through the
patch layer and the upserts underneath it is a change to the hottest write path
in tables and belongs in its own review, not on the sixth round of this one.

* refactor(tables): give the stale predicate one definition of "last beat"

`COALESCE(heartbeat_at, requested_at)` was written twice — once for the stale
threshold and again for the absolute ceiling — so the two could drift into
disagreeing about what proof of life means. One `lastBeat` fragment, one
`notBeatingSince(cutoff)` helper, both cutoffs expressed through it.

Also corrects the ceiling's comment: it triggers a day past the stale
threshold, not a day past now.

* fix(tables): delete the unguarded completion rather than guard it a fourth time

The two pre-claim exits — table missing, no target groups — still wrote through
`markDispatchComplete`. Last round I argued they run before the claim, "where
forcing a terminal state is the intent". That was wrong twice over: the table
lookup is awaited, so a cancel lands in that window like any other, and a
dispatch cancelled mid-lookup has not completed its scope any more than one
cancelled mid-window has.

Routing them through `completeDispatchIfActive` left `markDispatchComplete` with
no callers, so it is gone. That is the part worth having: this is the fourth
place the same defect appeared, each time because an unguarded writer was
sitting there to be reached. With it deleted, `completeDispatchIfActive` is the
only way to complete a dispatch and the class cannot recur.

* fix(tables): re-read the dispatch before committing a window

Several round trips separate the claim from the enqueue — the window query, the
executions prefetch, the tombstone filter — and nothing rechecked the dispatch
across them. A Stop-all or the stale sweep landing in that gap had the step
stamp cells and run a whole window for a dispatch already recorded as cancelled;
the existing recheck sits after the window, which is too late to prevent it.

Mirrors that existing check on the other side of the enqueue. It narrows the gap
to a single statement rather than closing it — a cancel arriving after this read
still races the enqueue, and no check can fix that. The cell-level
`cancellationGuard` and the `isExecCancelledAfter` tombstone filter are what
catch the remainder.

Author: waleedlatif1

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

@@ -396,8 +396,9 @@ describe('stale execution cleanup deadline grace', () => {
     const response = await GET(createRequest())
 
     expect(response.status).toBe(200)
-    expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(8)
-    expect(dbChainMockFns.for).toHaveBeenCalledTimes(8)
+    // Nine batched arms: the connector sync-log retention pass is the newest.
+    expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(9)
+    expect(dbChainMockFns.for).toHaveBeenCalledTimes(9)
     for (const [strength, options] of dbChainMockFns.for.mock.calls) {
       expect(strength).toBe('update')
       expect(options).toEqual({ skipLocked: true })
@@ -469,7 +470,7 @@ describe('stale execution cleanup deadline grace', () => {
     const limits = dbChainMockFns.limit.mock.calls.map(([limit]) => limit)
     expect(limits.filter((limit) => limit === 100)).toHaveLength(20)
     expect(limits.filter((limit) => limit === 1000)).toHaveLength(30)
-    expect(limits.filter((limit) => limit === 2000)).toHaveLength(11)
+    expect(limits.filter((limit) => limit === 2000)).toHaveLength(12)
 
     const workflowUpdates = dbChainMockFns.update.mock.calls.filter(
       ([table]) => table === workflowExecutionLogs

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

@@ -1,6 +1,7 @@
 import { db } from '@sim/db'
 import {
   asyncJobs,
+  knowledgeConnectorSyncLog,
   tableJobs,
   workflowDeploymentOperation,
   workflowExecutionLogs,
@@ -31,6 +32,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,12 +54,33 @@ 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
  * survive a client retry window, so 30 days is generous.
  */
 const DEPLOYMENT_OPERATION_RETENTION_DAYS = 30
+/**
+ * Terminal connector sync logs older than this are pruned. Nothing pruned them
+ * before, so the table grew by one row per sync run forever — a connector on a
+ * fifteen-minute interval writes about 35,000 rows a year on its own. That cost
+ * lands on `loadPreviousListingObservation`, which reads the newest `completed`
+ * row per connector through an index covering `connector_id` alone, so every
+ * retained row makes the sort behind the deletion-safety corroboration slower.
+ */
+const CONNECTOR_SYNC_LOG_RETENTION_DAYS = 30
+const CONNECTOR_SYNC_LOG_PRUNE_BATCH_SIZE = 2000
+const CONNECTOR_SYNC_LOG_MAX_ROWS_PER_RUN = 20_000
 const DEPLOYMENT_OPERATION_PRUNE_BATCH_SIZE = 2000
 const DEPLOYMENT_OPERATION_PRUNE_MAX_BATCHES = 10
 const WORKFLOW_EXECUTION_MUTATION_BATCH_SIZE = 100
@@ -144,6 +167,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
@@ -538,6 +564,90 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       })
     }
 
+    /**
+     * Prune terminal connector sync logs past retention.
+     *
+     * HARD INVARIANT: the newest row per connector must survive, and so must the
+     * newest `completed` row. `loadPreviousListingObservation` reconstructs the
+     * previous listing from the latest `completed` log, and that reconstruction
+     * decides whether a suspect listing is corroborated — i.e. whether
+     * reconciliation may delete documents. Pruning the last `completed` row
+     * would silently change deletion behaviour, so both `exists` guards below
+     * are load-bearing rather than defensive.
+     *
+     * `started` rows are never eligible: they are either in flight or waiting on
+     * the scheduler's own sweep to close them.
+     */
+    let connectorSyncLogsPruned = 0
+    try {
+      const syncLogRetention = new Date(
+        Date.now() - CONNECTOR_SYNC_LOG_RETENTION_DAYS * 24 * 60 * 60 * 1000
+      )
+      const newerSyncLog = alias(knowledgeConnectorSyncLog, 'newer_sync_log')
+      const newerCompletedSyncLog = alias(knowledgeConnectorSyncLog, 'newer_completed_sync_log')
+      const syncLogPredicate = and(
+        inArray(knowledgeConnectorSyncLog.status, ['completed', 'failed']),
+        lt(knowledgeConnectorSyncLog.startedAt, syncLogRetention),
+        exists(
+          db
+            .select({ id: newerSyncLog.id })
+            .from(newerSyncLog)
+            .where(
+              and(
+                eq(newerSyncLog.connectorId, knowledgeConnectorSyncLog.connectorId),
+                gt(newerSyncLog.startedAt, knowledgeConnectorSyncLog.startedAt)
+              )
+            )
+        ),
+        or(
+          ne(knowledgeConnectorSyncLog.status, 'completed'),
+          exists(
+            db
+              .select({ id: newerCompletedSyncLog.id })
+              .from(newerCompletedSyncLog)
+              .where(
+                and(
+                  eq(newerCompletedSyncLog.connectorId, knowledgeConnectorSyncLog.connectorId),
+                  eq(newerCompletedSyncLog.status, 'completed'),
+                  gt(newerCompletedSyncLog.startedAt, knowledgeConnectorSyncLog.startedAt)
+                )
+              )
+          )
+        )
+      )
+      const syncLogResult = await runBatchedMutation({
+        batchSize: CONNECTOR_SYNC_LOG_PRUNE_BATCH_SIZE,
+        maxRowsPerRun: CONNECTOR_SYNC_LOG_MAX_ROWS_PER_RUN,
+        claim: (tx, limit) =>
+          tx
+            .select({ id: knowledgeConnectorSyncLog.id })
+            .from(knowledgeConnectorSyncLog)
+            .where(syncLogPredicate)
+            .limit(limit)
+            .for('update', { skipLocked: true }),
+        mutation: (tx, candidateIds) =>
+          tx
+            .delete(knowledgeConnectorSyncLog)
+            .where(inArray(knowledgeConnectorSyncLog.id, candidateIds))
+            .returning({ id: knowledgeConnectorSyncLog.id }),
+      })
+      connectorSyncLogsPruned = syncLogResult.affected
+      if (connectorSyncLogsPruned > 0) {
+        logger.info(
+          `Pruned ${connectorSyncLogsPruned} old connector sync logs (retention: ${CONNECTOR_SYNC_LOG_RETENTION_DAYS}d)`
+        )
+      }
+      if (syncLogResult.reachedLimit) {
+        logger.info('Deferred remaining connector sync logs after reaching the per-run cap', {
+          maxRowsPerRun: CONNECTOR_SYNC_LOG_MAX_ROWS_PER_RUN,
+        })
+      }
+    } catch (error) {
+      logger.error('Failed to prune old connector sync logs:', {
+        error: toError(error).message,
+      })
+    }
+
     /**
      * Prune terminal deployment operations past retention. HARD INVARIANT:
      * the newest-generation row per workflow must always survive — the next
@@ -604,6 +714,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 +755,14 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       tableJobs: {
         staleMarkedFailed: staleTableJobsMarkedFailed,
       },
+      connectorSyncLogs: {
+        pruned: connectorSyncLogsPruned,
+        retentionDays: CONNECTOR_SYNC_LOG_RETENTION_DAYS,
+      },
+      tableRunDispatches: {
+        staleCancelled: staleDispatchesCancelled,
+        thresholdMinutes: TABLE_DISPATCH_STALE_THRESHOLD_MINUTES,
+      },
       deploymentOperations: {
         pruned: deploymentOperationsPruned,
         retentionDays: DEPLOYMENT_OPERATION_RETENTION_DAYS,

apps/sim/background/knowledge-processing.test.ts

@@ -139,3 +139,17 @@ describe('knowledge processing worker', () => {
     )
   })
 })
+
+describe('knowledge-process-document task configuration', () => {
+  /**
+   * `maxAttempts` does not cover an out-of-memory kill — Trigger.dev retries
+   * `TASK_PROCESS_OOM_KILLED` only when a larger preset is named. Eleven
+   * documents were killed in one afternoon and every one recorded
+   * `attempt_count = 1`, so each was left `failed` having never been retried.
+   */
+  it('escalates to a larger machine on an out-of-memory kill', async () => {
+    const { processDocument } = await import('@/background/knowledge-processing')
+
+    expect(processDocument.retry?.outOfMemory?.machine).toBe('large-2x')
+  })
+})

apps/sim/background/knowledge-processing.ts

@@ -56,12 +56,22 @@ export async function runDocumentProcessing(rawPayload: DocumentProcessingPayloa
 export const processDocument = task({
   id: 'knowledge-process-document',
   maxDuration: envNumber(env.KB_CONFIG_MAX_DURATION, 600),
-  machine: 'large-1x', // 2 vCPU, 2GB RAM - needed for large PDF processing
+  machine: 'large-1x', // 4 vCPU, 8GB RAM - needed for large PDF processing
   retry: {
     maxAttempts: envNumber(env.KB_CONFIG_MAX_ATTEMPTS, 3),
     factor: envNumber(env.KB_CONFIG_RETRY_FACTOR, 2),
     minTimeoutInMs: envNumber(env.KB_CONFIG_MIN_TIMEOUT, 1000),
     maxTimeoutInMs: envNumber(env.KB_CONFIG_MAX_TIMEOUT, 10000),
+    /**
+     * `maxAttempts` does not cover an out-of-memory kill — Trigger.dev retries
+     * `TASK_PROCESS_OOM_KILLED` only when a larger preset is named here. Eleven
+     * documents were killed in one afternoon and every one recorded
+     * `attempt_count = 1`, so each was left `failed` with no retry at all. The
+     * escalation is a safety net, not the fix: the workbook parser's allocation
+     * no longer scales with a sheet's declared range, and fleet p99 memory is
+     * 691 MB against this machine's 8 GB.
+     */
+    outOfMemory: { machine: 'large-2x' },
   },
   queue: {
     concurrencyLimit: envNumber(env.KB_CONFIG_CONCURRENCY_LIMIT, 20),

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/file-parsers/doc-parser.ts

@@ -1,6 +1,7 @@
 import { existsSync } from 'fs'
 import { readFile } from 'fs/promises'
 import { createLogger } from '@sim/logger'
+import { loadParseOfficeAsync } from '@/lib/file-parsers/officeparser-module'
 import type { FileParseResult, FileParser } from '@/lib/file-parsers/types'
 import { sanitizeTextForUTF8 } from '@/lib/file-parsers/utils'
 import { assertOoxmlArchiveWithinLimits } from '@/lib/file-parsers/zip-guard'
@@ -41,8 +42,8 @@ export class DocParser implements FileParser {
       assertOoxmlArchiveWithinLimits(buffer)
 
       try {
-        const officeParser = await import('officeparser')
-        const result = await officeParser.parseOfficeAsync(buffer)
+        const parseOfficeAsync = await loadParseOfficeAsync()
+        const result = await parseOfficeAsync(buffer)
 
         if (result) {
           const resultString = typeof result === 'string' ? result : String(result)

apps/sim/lib/file-parsers/docx-parser.ts

@@ -1,6 +1,7 @@
 import { readFile } from 'fs/promises'
 import { createLogger } from '@sim/logger'
 import mammoth from 'mammoth'
+import { loadParseOfficeAsync } from '@/lib/file-parsers/officeparser-module'
 import type { FileParseResult, FileParser } from '@/lib/file-parsers/types'
 import { sanitizeTextForUTF8 } from '@/lib/file-parsers/utils'
 import { assertOoxmlArchiveWithinLimits } from '@/lib/file-parsers/zip-guard'
@@ -65,8 +66,8 @@ export class DocxParser implements FileParser {
       }
 
       try {
-        const officeParser = await import('officeparser')
-        const result = await officeParser.parseOfficeAsync(buffer)
+        const parseOfficeAsync = await loadParseOfficeAsync()
+        const result = await parseOfficeAsync(buffer)
 
         if (result) {
           const resultString = typeof result === 'string' ? result : String(result)