fix(tables): handle row deletion during cell execution (#6520)

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

@@ -1,8 +1,9 @@
 /**
  * @vitest-environment node
  */
-import { dbChainMockFns, resetDbChainMock } from '@sim/testing'
+import { dbChainMockFns, queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing'
 import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
+import { TableRowNotFoundError } from '@/lib/table/rows/errors'
 import type { RowExecutionMetadata, TableDefinition, WorkflowGroup } from '@/lib/table/types'
 
 const { mockAppendTableEvent, mockDecryptSecret, mockUpdateRow, mockWriteExecutionsPatch } =
@@ -93,6 +94,7 @@ describe('writeWorkflowGroupState', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     resetDbChainMock()
+    queueTableRows(schemaMock.userTableRows, [{ id: CONTEXT.rowId }])
     mockWriteExecutionsPatch.mockResolvedValue('wrote')
     mockUpdateRow.mockResolvedValue({})
     mockAppendTableEvent.mockResolvedValue(null)
@@ -114,6 +116,7 @@ describe('writeWorkflowGroupState', () => {
       { [GROUP.id]: RUNNING_STATE },
       { groupId: GROUP.id, executionId: CONTEXT.executionId }
     )
+    expect(dbChainMockFns.for).toHaveBeenCalledWith('key share')
     expect(mockUpdateRow).not.toHaveBeenCalled()
     expect(mockAppendTableEvent).toHaveBeenCalledWith({
       kind: 'cell',
@@ -219,6 +222,42 @@ describe('writeWorkflowGroupState', () => {
 
     expect(mockAppendTableEvent).not.toHaveBeenCalled()
   })
+
+  it('skips a status write when the row was deleted before pickup', async () => {
+    resetDbChainMock()
+    mockWriteExecutionsPatch.mockResolvedValue('wrote')
+
+    await expect(writeWorkflowGroupState(CONTEXT, { executionState: RUNNING_STATE })).resolves.toBe(
+      'skipped'
+    )
+
+    expect(mockWriteExecutionsPatch).not.toHaveBeenCalled()
+    expect(mockAppendTableEvent).not.toHaveBeenCalled()
+  })
+
+  it('skips a data write when the row is deleted during execution', async () => {
+    mockUpdateRow.mockRejectedValueOnce(new TableRowNotFoundError())
+
+    await expect(
+      writeWorkflowGroupState(CONTEXT, {
+        executionState: RUNNING_STATE,
+        dataPatch: { 'first-output': 'late' },
+      })
+    ).resolves.toBe('skipped')
+
+    expect(mockAppendTableEvent).not.toHaveBeenCalled()
+  })
+
+  it('still throws unrelated data-write failures', async () => {
+    mockUpdateRow.mockRejectedValueOnce(new Error('database unavailable'))
+
+    await expect(
+      writeWorkflowGroupState(CONTEXT, {
+        executionState: RUNNING_STATE,
+        dataPatch: { 'first-output': 'late' },
+      })
+    ).rejects.toThrow('database unavailable')
+  })
 })
 
 describe('createWorkflowCellProgressWriter', () => {

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

@@ -10,9 +10,12 @@
  */
 
 import { db } from '@sim/db'
+import { userTableRows } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
+import { and, eq } from 'drizzle-orm'
 import { appendTableEvent } from '@/lib/table/events'
 import { pluckByPath } from '@/lib/table/pluck'
+import { TableRowNotFoundError } from '@/lib/table/rows/errors'
 import { writeExecutionsPatch } from '@/lib/table/rows/executions'
 import {
   createTableRowSecretProvenanceFromEncryptedExecution,
@@ -81,30 +84,48 @@ export async function writeWorkflowGroupState(
   let result: unknown
   if (hasDataPatch) {
     const { updateRow } = await import('@/lib/table/rows/service')
-    result = await updateRow(
-      {
-        tableId,
-        rowId,
-        data: dataPatch ?? {},
-        workspaceId,
-        executionsPatch,
-        cancellationGuard,
-        secretProvenance: payload.secretProvenance,
-      },
-      table,
-      requestId,
-      // `computedWrite` is what lets a workflow column keep populating on an
-      // update-locked table; the lock still covers user-authored columns.
-      { computedWrite: true }
-    )
+    try {
+      result = await updateRow(
+        {
+          tableId,
+          rowId,
+          data: dataPatch ?? {},
+          workspaceId,
+          executionsPatch,
+          cancellationGuard,
+          secretProvenance: payload.secretProvenance,
+        },
+        table,
+        requestId,
+        // `computedWrite` is what lets a workflow column keep populating on an
+        // update-locked table; the lock still covers user-authored columns.
+        { computedWrite: true }
+      )
+    } catch (error) {
+      if (!(error instanceof TableRowNotFoundError)) throw error
+      result = null
+    }
   } else {
-    result = await db.transaction((trx) =>
-      writeExecutionsPatch(trx, tableId, rowId, executionsPatch, cancellationGuard)
-    )
+    result = await db.transaction(async (trx) => {
+      const [row] = await trx
+        .select({ id: userTableRows.id })
+        .from(userTableRows)
+        .where(
+          and(
+            eq(userTableRows.id, rowId),
+            eq(userTableRows.tableId, tableId),
+            eq(userTableRows.workspaceId, workspaceId)
+          )
+        )
+        .limit(1)
+        .for('key share')
+      if (!row) return null
+      return writeExecutionsPatch(trx, tableId, rowId, executionsPatch, cancellationGuard)
+    })
   }
   if (result === null || result === 'guard-rejected') {
     logger.info(
-      `Skipping group write — SQL guard rejected stale or cancelled attempt (table=${tableId} row=${rowId} group=${groupId} executionId=${executionId})`
+      `Skipping group write — row missing or SQL guard rejected stale/cancelled attempt (table=${tableId} row=${rowId} group=${groupId} executionId=${executionId})`
     )
     return 'skipped'
   }

apps/sim/lib/table/rows/errors.ts

@@ -0,0 +1,7 @@
+/** Raised when a row disappears before an operation can mutate it. */
+export class TableRowNotFoundError extends Error {
+  constructor() {
+    super('Row not found')
+    this.name = 'TableRowNotFoundError'
+  }
+}

apps/sim/lib/table/rows/service.ts

@@ -41,6 +41,7 @@ import {
   withSeqscanOff,
 } from '@/lib/table/planner'
 import { encodeCursor } from '@/lib/table/rows/cursor'
+import { TableRowNotFoundError } from '@/lib/table/rows/errors'
 import {
   applyExecutionsPatch,
   deriveExecClearsForDataPatch,
@@ -1504,7 +1505,7 @@ export async function updateRow(
   // Get existing row
   const existingRow = await getRowById(data.tableId, data.rowId, data.workspaceId)
   if (!existingRow) {
-    throw new Error('Row not found')
+    throw new TableRowNotFoundError()
   }
 
   // Merge partial update with existing row data so callers can pass only changed fields
@@ -1576,7 +1577,7 @@ export async function updateRow(
             .where(eq(userTableRows.id, data.rowId))
             .returning({ id: userTableRows.id, updatedAt: userTableRows.updatedAt })
           const [updatedRow] = updatedRows
-          if (!updatedRow) throw new Error('Table row no longer exists')
+          if (!updatedRow) throw new TableRowNotFoundError()
 
           const result = await writeExecutionsPatch(
             trx,