fix(table): atomic per-key writes for executions, plus run-op race fixes

The executions blob on user_table_rows was read-modify-written wholesale on every
update. Concurrent writers (a column edit and a manual-retry stamp, two pickup
calls, a cancel and a cascade) each computed a merge from their own snapshot,
and the last writer clobbered keys it never touched — producing stuck "queued"
cells, vanished stamps, and stale completed exec records reappearing after
retries.

Fixes:
- updateRow / batchUpdateRows now apply executionsPatch via a SQL jsonb merge
  expression. Each writer only mutates the keys it explicitly patches; other
  keys are preserved. Eliminates the cross-key clobber.
- writeWorkflowGroupState bypasses the stale-worker guard for `queued` (new
  scheduler stamp) and `cancelled` (authoritative cancel) writes — those ARE
  the new authority for the cell. Previously the new run's stamp was being
  rejected by the same guard meant to block the OLD worker's writes.
- skipScheduler flag on UpdateRowData / BatchUpdateByIdData lets the cancel
  path and runWorkflowGroupsInternal opt out of the implicit auto-fire pass
  (cancel was waking up siblings; manual-run was racing its own scheduler).
- CELL_CONTENT pinned to h-[22px] so status badges don't grow rows.

Author: TheodoreSpeaks

apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/table-grid.tsx

@@ -89,8 +89,11 @@ const CELL_HEADER =
   'border-[var(--border)] border-r border-b bg-[var(--bg)] px-2 py-[7px] text-left align-middle'
 const CELL_HEADER_CHECKBOX =
   'sticky left-0 z-[12] border-[var(--border)] border-r border-b bg-[var(--bg)] px-1 py-[7px] text-center align-middle'
+// Fixed height (not min-) so a Badge-rendered status pill doesn't make the row
+// grow vs a plain-text neighbor. Sized to comfortably contain the badge; the
+// flex centers plain text + badges on the same baseline.
 const CELL_CONTENT =
-  'relative min-h-[20px] min-w-0 overflow-clip text-ellipsis whitespace-nowrap text-small'
+  'relative flex h-[22px] min-w-0 items-center overflow-clip text-ellipsis whitespace-nowrap text-small'
 const SELECTION_OVERLAY =
   'pointer-events-none absolute -top-px -right-px -bottom-px -left-px z-[5] border-[2px] border-[var(--selection)]'
 

apps/sim/background/workflow-column-execution.ts

@@ -353,10 +353,10 @@ export const workflowGroupCellTask = task({
   machine: 'medium-1x',
   retry: { maxAttempts: 1 },
   // Combined with `concurrencyKey: tableId`, caps each table's sub-queue to
-  // 10 in-flight cell jobs while letting different tables run in parallel.
+  // 20 in-flight cell jobs while letting different tables run in parallel.
   queue: {
     name: 'workflow-group-cell',
-    concurrencyLimit: 10,
+    concurrencyLimit: 20,
   },
   run: (payload: WorkflowGroupCellPayload, { signal }) =>
     executeWorkflowGroupCellJob(payload, signal),

apps/sim/lib/table/cell-write.ts

@@ -55,6 +55,25 @@ export async function writeWorkflowGroupState(
     return 'wrote'
   }
   const current = row.executions?.[groupId] as RowExecutionMetadata | undefined
+  // Stale-worker guard: only blocks writes FROM an old worker (status =
+  // running / completed / error / pending). A `queued` stamp is the scheduler
+  // claiming the cell for a brand-new run — the new executionId is supposed
+  // to overwrite whatever was there. Same for `cancelled` (authoritative).
+  // Without this carve-out, the new run's stamp gets rejected and the cell
+  // is stuck in its old state forever.
+  const isAuthoritativeNewStamp =
+    payload.executionState.status === 'queued' || payload.executionState.status === 'cancelled'
+  if (
+    !isAuthoritativeNewStamp &&
+    current &&
+    current.executionId &&
+    current.executionId !== executionId
+  ) {
+    logger.info(
+      `Skipping group write — stale worker (table=${tableId} row=${rowId} group=${groupId} mine=${executionId} active=${current.executionId})`
+    )
+    return 'skipped'
+  }
   if (
     current?.status === 'cancelled' &&
     current.executionId === executionId &&
@@ -66,11 +85,11 @@ export async function writeWorkflowGroupState(
     return 'skipped'
   }
   // Skip writing `cancelled` state with the guard — that's an authoritative
-  // write from `cancelWorkflowGroupRuns` and must always land. Cell-task
-  // writes (running/completed/error) get the SQL guard so an in-flight
-  // partial can't clobber a stop click that already committed.
-  const cancellationGuard =
-    payload.executionState.status === 'cancelled' ? undefined : { groupId, executionId }
+  // write from `cancelWorkflowGroupRuns` and must always land. New `queued`
+  // stamps from the scheduler also bypass — they ARE the new authority. Cell-
+  // task writes (running/completed/error) get the SQL guard so an in-flight
+  // partial can't clobber a stop click or a newer run that already committed.
+  const cancellationGuard = isAuthoritativeNewStamp ? undefined : { groupId, executionId }
   const result = await updateRow(
     {
       tableId,

apps/sim/lib/table/service.ts

@@ -12,7 +12,7 @@ import { userTableDefinitions, userTableRows, workflowExecutionLogs } from '@sim
 import { createLogger } from '@sim/logger'
 import { getPostgresErrorCode } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
-import { and, count, eq, gt, gte, inArray, isNull, sql } from 'drizzle-orm'
+import { and, count, eq, gt, gte, inArray, isNull, type SQL, sql } from 'drizzle-orm'
 import { env } from '@/lib/core/config/env'
 import { generateRestoreName } from '@/lib/core/utils/restore-name'
 import { getSocketServerUrl } from '@/lib/core/utils/urls'
@@ -1597,6 +1597,35 @@ function applyExecutionsPatch(
   return next
 }
 
+/**
+ * Builds a SQL expression that applies the given `executionsPatch` to the
+ * row's `executions` jsonb in-place — set keys for non-null values, delete
+ * keys for `null` values. Returns null when the patch is empty/missing.
+ *
+ * Why server-side: read-modify-write on the entire jsonb blob races between
+ * concurrent writers (e.g., a column edit and a manual-retry stamp), so the
+ * last writer wins for keys it didn't touch and clobbers other writers'
+ * exec updates. Patching keys at the SQL level keeps each writer's changes
+ * atomic per-key.
+ */
+function buildExecutionsSqlPatch(
+  patch: Record<string, RowExecutionMetadata | null> | undefined
+): SQL | null {
+  if (!patch) return null
+  const entries = Object.entries(patch)
+  if (entries.length === 0) return null
+
+  let expr: SQL = sql`coalesce(${userTableRows.executions}, '{}'::jsonb)`
+  for (const [gid, value] of entries) {
+    if (value === null) {
+      expr = sql`(${expr}) - ${gid}::text`
+    } else {
+      expr = sql`(${expr}) || jsonb_build_object(${gid}::text, ${JSON.stringify(value)}::jsonb)`
+    }
+  }
+  return expr
+}
+
 /**
  * Updates a single row.
  *
@@ -1653,26 +1682,46 @@ export async function updateRow(
   const now = new Date()
 
   // Cell-task partial writes pass `cancellationGuard` so the SQL update is a
-  // no-op when a stop click already wrote `cancelled` for this run between
-  // the in-process read and now. Without this, an in-flight `running`
-  // partial-write can land after `cancelled` and clobber it.
+  // no-op when (a) a stop click already wrote `cancelled` for this run, or
+  // (b) a newer run has taken over the cell with a different executionId. The
+  // worker is "this run's writes only land if this run is still the active
+  // run on the cell." Authoritative cancel writes from `cancelWorkflowGroupRuns`
+  // skip the guard entirely (they don't pass `cancellationGuard`).
+  //
+  // SQL-level for atomicity: an in-process read + update would race a
+  // concurrent stop or rerun. The two clauses are joined by AND because
+  // either failing means the worker is no longer authoritative.
   const guard = data.cancellationGuard
-  // The guard rejects writes only when the DB *already* shows
-  // `cancelled` + matching executionId. Wrap the JSON traversals in
-  // `IS DISTINCT FROM` so a missing `executions[groupId]` (NULL) cleanly
-  // evaluates as "different" — Postgres three-valued logic would otherwise
-  // make the whole expression NULL and the UPDATE would mistakenly become
-  // a no-op for any row that has no prior execution record.
   const whereClause = guard
     ? and(
         eq(userTableRows.id, data.rowId),
-        sql`(executions->${guard.groupId}->>'status' IS DISTINCT FROM 'cancelled' OR executions->${guard.groupId}->>'executionId' IS DISTINCT FROM ${guard.executionId})`
+        // Reject writes that would land on top of an already-`cancelled` state
+        // for this same run. Wrapped in IS DISTINCT FROM so a missing exec
+        // (NULL) cleanly evaluates as "different" rather than NULL-poisoning.
+        sql`(executions->${guard.groupId}->>'status' IS DISTINCT FROM 'cancelled' OR executions->${guard.groupId}->>'executionId' IS DISTINCT FROM ${guard.executionId})`,
+        // Reject writes from a stale worker — the cell's active run has moved
+        // on. `OR exec IS NULL` lets the worker land its first `running`
+        // stamp on a row that has no prior exec record (initial stamp from
+        // the scheduler may not have committed yet).
+        sql`(executions->${guard.groupId} IS NULL OR executions->${guard.groupId}->>'executionId' = ${guard.executionId})`
       )
     : eq(userTableRows.id, data.rowId)
 
+  // Apply the executions patch at the SQL level — we never overwrite the full
+  // executions blob, only the keys the caller explicitly patched. Without
+  // this, concurrent updateRow calls (e.g., a column edit and a manual
+  // retry's stamp) would each compute `mergedExecutions` from their own
+  // in-memory snapshot and the last writer wins, clobbering the other's
+  // exec keys. The data field still does last-writer-wins because that's
+  // the user's edit, but exec records are independently keyed by groupId.
+  const executionsExpr = buildExecutionsSqlPatch(data.executionsPatch)
   const updated = await db
     .update(userTableRows)
-    .set({ data: mergedData, executions: mergedExecutions, updatedAt: now })
+    .set({
+      data: mergedData,
+      ...(executionsExpr ? { executions: executionsExpr } : {}),
+      updatedAt: now,
+    })
     .where(whereClause)
     .returning({ id: userTableRows.id })
 
@@ -1710,7 +1759,7 @@ export async function updateRow(
   notifyTableRowUpdated(data.tableId, updatedRow)
   // Awaited (not `void`) so cell tasks dispatch their cascade before the
   // trigger.dev worker tears down on `run()` resolve.
-  await scheduleRunsForRows(table, [updatedRow])
+  if (!data.skipScheduler) await scheduleRunsForRows(table, [updatedRow])
 
   return updatedRow
 }
@@ -1928,6 +1977,7 @@ export async function batchUpdateRows(
     rowId: string
     mergedData: RowData
     mergedExecutions: RowExecutions
+    executionsPatch?: Record<string, RowExecutionMetadata | null>
   }> = []
   for (const update of data.updates) {
     const existing = existingMap.get(update.rowId)!
@@ -1944,7 +1994,12 @@ export async function batchUpdateRows(
       throw new Error(`Row ${update.rowId}: ${schemaValidation.errors.join(', ')}`)
     }
 
-    mergedUpdates.push({ rowId: update.rowId, mergedData: merged, mergedExecutions })
+    mergedUpdates.push({
+      rowId: update.rowId,
+      mergedData: merged,
+      mergedExecutions,
+      executionsPatch: update.executionsPatch,
+    })
   }
 
   const uniqueColumns = getUniqueColumns(table.schema)
@@ -1968,12 +2023,20 @@ export async function batchUpdateRows(
     await setTableTxTimeouts(trx, { statementMs: 60_000 })
     for (let i = 0; i < mergedUpdates.length; i += TABLE_LIMITS.UPDATE_BATCH_SIZE) {
       const batch = mergedUpdates.slice(i, i + TABLE_LIMITS.UPDATE_BATCH_SIZE)
-      const updatePromises = batch.map(({ rowId, mergedData, mergedExecutions }) =>
-        trx
+      // Same as `updateRow`: patch executions at the SQL level when a patch
+      // is set, so concurrent writers don't clobber each other's keys via
+      // last-writer-wins on the full jsonb blob.
+      const updatePromises = batch.map(({ rowId, mergedData, executionsPatch }) => {
+        const executionsExpr = buildExecutionsSqlPatch(executionsPatch)
+        return trx
           .update(userTableRows)
-          .set({ data: mergedData, executions: mergedExecutions, updatedAt: now })
+          .set({
+            data: mergedData,
+            ...(executionsExpr ? { executions: executionsExpr } : {}),
+            updatedAt: now,
+          })
           .where(eq(userTableRows.id, rowId))
-      )
+      })
       await Promise.all(updatePromises)
     }
   })
@@ -2006,7 +2069,7 @@ export async function batchUpdateRows(
   // so the scheduler's later per-write notifications (pending/running) land
   // last and stick in the client cache.
   for (const row of updatedRowsForTrigger) notifyTableRowUpdated(data.tableId, row)
-  void scheduleRunsForRows(table, updatedRowsForTrigger)
+  if (!data.skipScheduler) void scheduleRunsForRows(table, updatedRowsForTrigger)
 
   return {
     affectedCount: mergedUpdates.length,

apps/sim/lib/table/types.ts

@@ -308,6 +308,14 @@ export interface UpdateRowData {
    * state. `updateRow` returns `null` when the guard rejects the write.
    */
   cancellationGuard?: { groupId: string; executionId: string }
+  /**
+   * When true, the post-write `scheduleRunsForRows` call is skipped. Used by
+   * the cancel path (which is tearing rows down, not waking them up) and by
+   * the manual-run path (which fires its own `scheduleRunsForRows` with
+   * `isManualRun: true` and doesn't want a duplicate auto-fire pass on the
+   * cleared cells). Default false: every other write fires the reactor.
+   */
+  skipScheduler?: boolean
 }
 
 export interface BulkUpdateData {
@@ -326,6 +334,8 @@ export interface BatchUpdateByIdData {
     executionsPatch?: Record<string, RowExecutionMetadata | null>
   }>
   workspaceId: string
+  /** Same semantics as `UpdateRowData.skipScheduler`. */
+  skipScheduler?: boolean
 }
 
 export interface BulkDeleteData {

apps/sim/lib/table/workflow-columns.ts

@@ -253,8 +253,8 @@ interface RunGroupCellOptions {
   executionId: string
 }
 
-/** Per-table concurrency cap. Mirrors trigger.dev's `concurrencyLimit: 10`. */
-const TABLE_CONCURRENCY_LIMIT = 10
+/** Per-table concurrency cap. Mirrors trigger.dev's `concurrencyLimit: 20`. */
+const TABLE_CONCURRENCY_LIMIT = 20
 
 async function stampQueuedOrCancel(
   queue: Awaited<ReturnType<typeof getJobQueue>>,
@@ -364,6 +364,11 @@ export async function cancelWorkflowGroupRuns(tableId: string, rowId?: string):
       )
     )
   )
+  // `skipScheduler: true` — we're tearing rows down, not waking them up. The
+  // auto-fire reactor would otherwise see independent (row, group) pairs whose
+  // deps are now satisfied (because the upstream group already wrote its
+  // output before the cancel) and re-enqueue them, which is exactly what the
+  // user clicked Stop to prevent.
   await Promise.allSettled(
     mutations.map((m) =>
       updateRow(
@@ -373,6 +378,7 @@ export async function cancelWorkflowGroupRuns(tableId: string, rowId?: string):
           data: {},
           workspaceId: table.workspaceId,
           executionsPatch: m.executionsPatch,
+          skipScheduler: true,
         },
         table,
         `wfgrp-cancel-${m.rowId}`
@@ -471,7 +477,11 @@ async function runWorkflowGroupsInternal(opts: {
 
   if (updates.length === 0) return { triggered: 0 }
 
-  await batchUpdateRows({ tableId, updates, workspaceId }, table, requestId)
+  // `skipScheduler: true` because we fire `scheduleRunsForRows` ourselves
+  // below with `isManualRun: true`. Without the skip, batchUpdateRows runs the
+  // auto-fire reactor first and any autoRun=true sibling group whose deps are
+  // satisfied would race the manual call.
+  await batchUpdateRows({ tableId, updates, workspaceId, skipScheduler: true }, table, requestId)
 
   return scheduleRunsForRows(table, clearedRows, { isManualRun: true })
 }