fix(tables): tolerate row deletion during run cancellation (#6600)

* fix(tables): tolerate row deletion during run cancellation

* fix(tables): unwrap row deletion errors

Author: TheodoreSpeaks

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

@@ -1,8 +1,16 @@
 /**
  * @vitest-environment node
  */
-import { resetEnvFlagsMock, setEnvFlags } from '@sim/testing'
+import {
+  dbChainMockFns,
+  queueTableRows,
+  resetDbChainMock,
+  resetEnvFlagsMock,
+  schemaMock,
+  setEnvFlags,
+} from '@sim/testing'
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
+import { TableRowNotFoundError } from '@/lib/table/rows/errors'
 import type {
   RowExecutionMetadata,
   TableDefinition,
@@ -15,11 +23,25 @@ const {
   mockResolveSystemBillingAttribution,
   mockRunsCancel,
   mockRunsList,
+  mockGetJobQueue,
+  mockGetTableById,
+  mockListActiveDispatches,
+  mockMarkActiveDispatchesCancelled,
+  mockQueueCancelByKey,
+  mockQueueCancelJob,
+  mockUpdateRow,
 } = vi.hoisted(() => ({
   mockResolveBillingAttribution: vi.fn(),
   mockResolveSystemBillingAttribution: vi.fn(),
   mockRunsCancel: vi.fn(),
   mockRunsList: vi.fn(),
+  mockGetJobQueue: vi.fn(),
+  mockGetTableById: vi.fn(),
+  mockListActiveDispatches: vi.fn(),
+  mockMarkActiveDispatchesCancelled: vi.fn(),
+  mockQueueCancelByKey: vi.fn(),
+  mockQueueCancelJob: vi.fn(),
+  mockUpdateRow: vi.fn(),
 }))
 
 const SYSTEM_BILLING_ATTRIBUTION = {
@@ -48,15 +70,40 @@ vi.mock('@trigger.dev/sdk', () => ({
   },
 }))
 
+vi.mock('@/lib/core/async-jobs/config', () => ({
+  getJobQueue: mockGetJobQueue,
+}))
+
+vi.mock('@/lib/table/dispatcher', () => ({
+  listActiveDispatches: mockListActiveDispatches,
+  markActiveDispatchesCancelled: mockMarkActiveDispatchesCancelled,
+}))
+
+vi.mock('@/lib/table/rows/service', () => ({
+  updateRow: mockUpdateRow,
+}))
+
+vi.mock('@/lib/table/service', () => ({
+  getTableById: mockGetTableById,
+}))
+
 import {
   buildEnqueueItems,
   cancelCellRunsByTags,
+  cancelWorkflowGroupRuns,
   pickNextEligibleGroupForRow,
   type WorkflowGroupCellPayload,
 } from '@/lib/table/workflow-columns'
 
 beforeEach(() => {
   vi.clearAllMocks()
+  resetDbChainMock()
+  mockGetJobQueue.mockResolvedValue({
+    cancelByKey: mockQueueCancelByKey,
+    cancelJob: mockQueueCancelJob,
+  })
+  mockListActiveDispatches.mockResolvedValue([])
+  mockMarkActiveDispatchesCancelled.mockResolvedValue([])
   mockResolveBillingAttribution.mockImplementation(
     ({ actorUserId, workspaceId }: { actorUserId: string; workspaceId: string }) =>
       Promise.resolve({
@@ -271,3 +318,78 @@ describe('cancelCellRunsByTags', () => {
     )
   })
 })
+
+describe('cancelWorkflowGroupRuns deletion races', () => {
+  const group = makeGroup({ id: 'g1' })
+  const table = makeTable([group])
+  const inFlightExecution = {
+    tableId: table.id,
+    rowId: 'row1',
+    groupId: group.id,
+    status: 'running',
+    executionId: 'execution-1',
+    jobId: null,
+    workflowId: group.workflowId,
+    error: null,
+    runningBlockIds: [],
+    blockErrors: {},
+    cancelledAt: null,
+  }
+
+  beforeEach(() => {
+    setEnvFlags({ isTriggerDevEnabled: false, isBillingEnabled: true })
+    mockGetTableById.mockResolvedValue(table)
+  })
+
+  it('ignores a row deleted after its in-flight execution was selected', async () => {
+    queueTableRows(schemaMock.tableRowExecutions, [inFlightExecution])
+    mockUpdateRow.mockRejectedValueOnce(new TableRowNotFoundError())
+
+    await expect(cancelWorkflowGroupRuns(table.id)).resolves.toBe(1)
+    expect(mockUpdateRow).toHaveBeenCalledOnce()
+  })
+
+  it('ignores a transaction-wrapped row deletion', async () => {
+    queueTableRows(schemaMock.tableRowExecutions, [inFlightExecution])
+    mockUpdateRow.mockRejectedValueOnce(
+      new Error('Failed query', { cause: new TableRowNotFoundError() })
+    )
+
+    await expect(cancelWorkflowGroupRuns(table.id)).resolves.toBe(1)
+  })
+
+  it('rethrows unrelated cancellation write failures', async () => {
+    const error = new Error('database unavailable')
+    queueTableRows(schemaMock.tableRowExecutions, [inFlightExecution])
+    mockUpdateRow.mockRejectedValueOnce(error)
+
+    await expect(cancelWorkflowGroupRuns(table.id)).rejects.toBe(error)
+  })
+
+  it('ignores a tombstone foreign-key failure caused by a deleted row', async () => {
+    mockListActiveDispatches.mockResolvedValueOnce([
+      { id: 'dispatch-1', scope: { groupIds: [group.id], rowIds: ['row1'] } },
+    ])
+    const cause = Object.assign(new Error('foreign key violation'), {
+      code: '23503',
+      constraint_name: 'table_row_executions_row_id_user_table_rows_id_fk',
+    })
+    dbChainMockFns.onConflictDoNothing.mockRejectedValueOnce(new Error('Failed query', { cause }))
+
+    await expect(cancelWorkflowGroupRuns(table.id, 'row1')).resolves.toBe(0)
+  })
+
+  it('rethrows tombstone failures from any other constraint', async () => {
+    mockListActiveDispatches.mockResolvedValueOnce([
+      { id: 'dispatch-1', scope: { groupIds: [group.id], rowIds: ['row1'] } },
+    ])
+    const cause = Object.assign(new Error('foreign key violation'), {
+      code: '23503',
+      constraint_name: 'table_row_executions_table_id_user_table_definitions_id_fk',
+    })
+    const error = new Error('Failed query', { cause })
+    dbChainMockFns.onConflictDoNothing.mockRejectedValueOnce(error)
+
+    await expect(cancelWorkflowGroupRuns(table.id, 'row1')).rejects.toBe(error)
+  })
+})

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

@@ -12,7 +12,12 @@ import {
   userTableRows as userTableRowsTable,
 } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { toError } from '@sim/utils/errors'
+import {
+  findCause,
+  getPostgresConstraintName,
+  getPostgresErrorCode,
+  toError,
+} from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { and, asc, eq, gt, inArray, notInArray, or, sql } from 'drizzle-orm'
 import type { BillingAttributionSnapshot } from '@/lib/billing/core/billing-attribution'
@@ -25,6 +30,7 @@ import {
 import { OrchestrationError } from '@/lib/core/orchestration/types'
 import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import { buildCancelledExecution } from '@/lib/table/cell-write'
+import { TableRowNotFoundError } from '@/lib/table/rows/errors'
 import type {
   Filter,
   RowData,
@@ -43,6 +49,7 @@ const TABLE_CANCELLATION_MAX_ROWS = 5_000
 const TABLE_CANCELLATION_CONCURRENCY = 10
 const TABLE_TRIGGER_CANCELLATION_MAX_RUNS = 5_000
 const TABLE_TRIGGER_CANCELLATION_RETENTION_MS = 14 * 24 * 60 * 60_000
+const TABLE_ROW_EXECUTIONS_ROW_FK = 'table_row_executions_row_id_user_table_rows_id_fk'
 
 import { getColumnId } from '@/lib/table/column-keys'
 import { USER_TABLE_ROWS_SQL_NAME } from '@/lib/table/constants'
@@ -715,20 +722,29 @@ export async function cancelWorkflowGroupRuns(
     )
 
     await mapWithConcurrency(mutations, TABLE_CANCELLATION_CONCURRENCY, async (mutation) => {
-      const updated = await updateRow(
-        {
-          tableId,
-          rowId: mutation.rowId,
-          data: {},
-          /** No cell values are written, so there is nothing to stamp. */
-          secretProvenance: undefined,
-          workspaceId: table.workspaceId,
-          executionsPatch: mutation.executionsPatch,
-        },
-        table,
-        `wfgrp-cancel-${mutation.rowId}`
-      )
-      if (!updated) throw new Error('Authoritative cancellation write was rejected')
+      try {
+        const updated = await updateRow(
+          {
+            tableId,
+            rowId: mutation.rowId,
+            data: {},
+            /** No cell values are written, so there is nothing to stamp. */
+            secretProvenance: undefined,
+            workspaceId: table.workspaceId,
+            executionsPatch: mutation.executionsPatch,
+          },
+          table,
+          `wfgrp-cancel-${mutation.rowId}`
+        )
+        if (!updated) throw new Error('Authoritative cancellation write was rejected')
+      } catch (error) {
+        const rowNotFound = findCause(
+          error,
+          (cause): cause is TableRowNotFoundError => cause instanceof TableRowNotFoundError
+        )
+        if (rowNotFound) return
+        throw error
+      }
     })
     cancelledCount += mutations.reduce((total, mutation) => total + mutation.cancelledCount, 0)
 
@@ -783,25 +799,35 @@ export async function cancelWorkflowGroupRuns(
         needsTombstone,
         TABLE_CANCELLATION_CONCURRENCY,
         async (tombstone) => {
-          await db
-            .insert(tableRowExecutions)
-            .values({
-              tableId,
-              rowId,
-              groupId: tombstone.groupId,
-              status: 'cancelled',
-              executionId: null,
-              jobId: null,
-              workflowId: tombstone.workflowId,
-              error: 'Cancelled',
-              runningBlockIds: [],
-              blockErrors: {},
-              cancelledAt: now,
-              updatedAt: now,
-            })
-            .onConflictDoNothing({
-              target: [tableRowExecutions.rowId, tableRowExecutions.groupId],
-            })
+          try {
+            await db
+              .insert(tableRowExecutions)
+              .values({
+                tableId,
+                rowId,
+                groupId: tombstone.groupId,
+                status: 'cancelled',
+                executionId: null,
+                jobId: null,
+                workflowId: tombstone.workflowId,
+                error: 'Cancelled',
+                runningBlockIds: [],
+                blockErrors: {},
+                cancelledAt: now,
+                updatedAt: now,
+              })
+              .onConflictDoNothing({
+                target: [tableRowExecutions.rowId, tableRowExecutions.groupId],
+              })
+          } catch (error) {
+            if (
+              getPostgresErrorCode(error) === '23503' &&
+              getPostgresConstraintName(error) === TABLE_ROW_EXECUTIONS_ROW_FK
+            ) {
+              return
+            }
+            throw error
+          }
         }
       )
     }