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.

Author: waleedlatif1

apps/sim/lib/table/dispatcher.ts

@@ -17,6 +17,7 @@ import {
   sql,
 } from 'drizzle-orm'
 import { getJobQueue } from '@/lib/core/async-jobs/config'
+import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import { writeWorkflowGroupState } from '@/lib/table/cell-write'
 import { USER_TABLE_ROWS_SQL_NAME } from '@/lib/table/constants'
 import { isExecCancelledAfter } from '@/lib/table/deps'
@@ -43,6 +44,9 @@ const logger = createLogger('TableRunDispatcher')
 
 const ACTIVE_DISPATCH_STATUSES = ['pending', 'dispatching'] as const
 
+/** Concurrent terminal-event writes when the stale sweep reclaims a batch. */
+const STALE_DISPATCH_EVENT_CONCURRENCY = 10
+
 export type DispatchStatus = 'pending' | 'dispatching' | 'complete' | 'cancelled'
 export type DispatchMode = 'all' | 'incomplete' | 'new'
 
@@ -812,6 +816,44 @@ export async function completeDispatchIfActive(dispatchId: string): Promise<bool
   return transitioned.length > 0
 }
 
+/**
+ * Whether any cell inside a dispatch's own scope has reported since `since`.
+ *
+ * Scoped to the dispatch's groups, and to its rows when it names any, because
+ * `table_row_executions` carries no dispatch column. Left table-wide, a live
+ * dispatch's cells read as evidence that an abandoned dispatch beside it was
+ * still working — and auto-fired and row-scoped runs do NOT cancel overlapping
+ * dispatches (`cancelPriorRuns` requires `isManualRun`, and the per-row path is
+ * a no-op for dispatch cancellation), so sharing a group is ordinary rather than
+ * exceptional.
+ *
+ * Two table-wide dispatches over the same groups can still vouch for each other,
+ * since nothing in the row execution says whose work it is. Closing that needs a
+ * `dispatch_id` on `table_row_executions`, threaded through every cell-write
+ * site. Until then the residue is a delay, not a permanent mask: the live
+ * dispatch's cells stop reporting when it finishes.
+ *
+ * Rides the partial `(table_id, status)` index, which covers exactly these three
+ * statuses.
+ */
+function hasRecentCellActivity(since: Date): SQL {
+  return sql`EXISTS (
+    SELECT 1 FROM ${tableRowExecutions}
+    WHERE ${tableRowExecutions.tableId} = ${tableRunDispatches.tableId}
+      AND ${tableRowExecutions.groupId} IN (
+        SELECT jsonb_array_elements_text(${tableRunDispatches.scope} -> 'groupIds')
+      )
+      AND (
+        jsonb_typeof(${tableRunDispatches.scope} -> 'rowIds') <> 'array'
+        OR ${tableRowExecutions.rowId} IN (
+          SELECT jsonb_array_elements_text(${tableRunDispatches.scope} -> 'rowIds')
+        )
+      )
+      AND ${tableRowExecutions.status} IN ('queued', 'running', 'pending')
+      AND ${tableRowExecutions.updatedAt} >= ${sql.param(since, tableRowExecutions.updatedAt)}
+  )`
+}
+
 /**
  * Cancels dispatches whose holder died without reaching a terminal state.
  *
@@ -881,21 +923,7 @@ export async function cancelStaleDispatches(
     and(
       inArray(tableRunDispatches.status, [...ACTIVE_DISPATCH_STATUSES]),
       sql`COALESCE(${tableRunDispatches.heartbeatAt}, ${tableRunDispatches.requestedAt}) < ${sql.param(staleBefore, tableRunDispatches.heartbeatAt)}`,
-      sql`NOT EXISTS (
-        SELECT 1 FROM ${tableRowExecutions}
-        WHERE ${tableRowExecutions.tableId} = ${tableRunDispatches.tableId}
-          AND ${tableRowExecutions.groupId} IN (
-            SELECT jsonb_array_elements_text(${tableRunDispatches.scope} -> 'groupIds')
-          )
-          AND (
-            jsonb_typeof(${tableRunDispatches.scope} -> 'rowIds') <> 'array'
-            OR ${tableRowExecutions.rowId} IN (
-              SELECT jsonb_array_elements_text(${tableRunDispatches.scope} -> 'rowIds')
-            )
-          )
-          AND ${tableRowExecutions.status} IN ('queued', 'running', 'pending')
-          AND ${tableRowExecutions.updatedAt} >= ${sql.param(staleBefore, tableRowExecutions.updatedAt)}
-      )`
+      sql`NOT ${hasRecentCellActivity(staleBefore)}`
     )
 
   // Claimed as explicit ids first, then updated by id, so the bound is evaluated
@@ -939,22 +967,26 @@ export async function cancelStaleDispatches(
     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,
-      })
-    )
+  /**
+   * 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.
+   *
+   * Bounded rather than a bare `Promise.all`: the sibling cancel paths fan out
+   * over one table's dispatches, while this sweep can carry a whole tick's worth
+   * across many tables, and each event is its own write.
+   */
+  await mapWithConcurrency(dispatches, STALE_DISPATCH_EVENT_CONCURRENCY, (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

apps/sim/lib/table/stale-dispatch-recovery.test.ts

@@ -103,33 +103,32 @@ describe('cancelStaleDispatches', () => {
   it('spares a dispatch whose cells are still reporting', async () => {
     await cancelStaleDispatches(STALE_BEFORE, 200)
 
-    const chunks = collectChunks(dbChainMockFns.where.mock.calls[0][0])
+    /**
+     * Matched on the literal SQL, not on column references: the fragment
+     * interpolates the `tableRowExecutions` table object, so every one of its
+     * column names appears in the chunks whether the predicate uses it or not —
+     * asserting on those passes even with the predicate deleted.
+     */
+    const joined = collectChunks(dbChainMockFns.where.mock.calls[0][0]).join(' ')
+
     /**
      * The dispatch's own heartbeat is stamped between windows, not during them,
      * and the loop is checkpointed for the whole window — so a long window
      * leaves it untouched while the dispatch is plainly alive. Its cells carry
      * the signal the checkpointed parent cannot, and both have to be stale
      * before the row is reclaimed.
      */
-    expect(chunks.some((chunk) => chunk.includes('NOT EXISTS'))).toBe(true)
-    expect(chunks).toContain('tableRowExecutions.updatedAt')
-    expect(chunks).toContain('tableRowExecutions.tableId')
-    /**
-     * Narrowed to the dispatch's own groups: row executions carry no dispatch
-     * column, so a table-wide probe lets a LIVE dispatch's cells vouch for an
-     * ABANDONED one beside it, and the abandoned row is never reclaimed.
-     */
-    expect(chunks).toContain('tableRowExecutions.groupId')
-    expect(chunks.some((chunk) => chunk.includes('jsonb_array_elements_text'))).toBe(true)
+    expect(joined).toContain('NOT ')
+    expect(joined).toContain('EXISTS (')
+
     /**
-     * Rows too, when the dispatch names any. Auto-fired and row-scoped runs do
-     * NOT cancel overlapping dispatches — `cancelPriorRuns` requires
-     * `isManualRun`, and the per-row path is a no-op for dispatch cancellation —
-     * so a live dispatch sharing a group is ordinary, and only the row filter
-     * keeps its cells from vouching for an abandoned neighbour.
+     * Scoped to the dispatch's own groups AND its rows. Auto-fired and
+     * row-scoped runs do NOT cancel overlapping dispatches, so a live dispatch
+     * sharing a table is ordinary — without both filters its cells vouch for an
+     * abandoned neighbour and the abandoned row is never reclaimed.
      */
-    expect(chunks).toContain('tableRowExecutions.rowId')
-    expect(chunks.some((chunk) => chunk.includes("'rowIds'"))).toBe(true)
+    expect(joined).toContain("-> 'groupIds'")
+    expect(joined).toContain("-> 'rowIds'")
   })
 
   it('emits the terminal event so a stuck client overlay clears', async () => {