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

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)