From 1e24c0af3999040c586a76dd93f16a9ecd409055 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Tue, 11 Aug 2026 21:40:27 -0700 Subject: [PATCH 1/2] fix(tables): tolerate row deletion during run cancellation --- apps/sim/lib/table/workflow-columns.test.ts | 115 +++++++++++++++++++- apps/sim/lib/table/workflow-columns.ts | 85 +++++++++------ 2 files changed, 165 insertions(+), 35 deletions(-) diff --git a/apps/sim/lib/table/workflow-columns.test.ts b/apps/sim/lib/table/workflow-columns.test.ts index c9c5884f634..4c979c5d13f 100644 --- a/apps/sim/lib/table/workflow-columns.test.ts +++ b/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,69 @@ 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('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) + }) +}) diff --git a/apps/sim/lib/table/workflow-columns.ts b/apps/sim/lib/table/workflow-columns.ts index 064072577c9..b5ab26fbce6 100644 --- a/apps/sim/lib/table/workflow-columns.ts +++ b/apps/sim/lib/table/workflow-columns.ts @@ -12,7 +12,7 @@ import { userTableRows as userTableRowsTable, } from '@sim/db/schema' import { createLogger } from '@sim/logger' -import { toError } from '@sim/utils/errors' +import { 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 +25,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 +44,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 +717,25 @@ 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) { + if (error instanceof TableRowNotFoundError) return + throw error + } }) cancelledCount += mutations.reduce((total, mutation) => total + mutation.cancelledCount, 0) @@ -783,25 +790,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 + } } ) } From f008e383f5ab5dca568db95670df720991674360 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Tue, 11 Aug 2026 21:45:29 -0700 Subject: [PATCH 2/2] fix(tables): unwrap row deletion errors --- apps/sim/lib/table/workflow-columns.test.ts | 9 +++++++++ apps/sim/lib/table/workflow-columns.ts | 13 +++++++++++-- 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/apps/sim/lib/table/workflow-columns.test.ts b/apps/sim/lib/table/workflow-columns.test.ts index 4c979c5d13f..a63ed47a7e8 100644 --- a/apps/sim/lib/table/workflow-columns.test.ts +++ b/apps/sim/lib/table/workflow-columns.test.ts @@ -349,6 +349,15 @@ describe('cancelWorkflowGroupRuns deletion races', () => { 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]) diff --git a/apps/sim/lib/table/workflow-columns.ts b/apps/sim/lib/table/workflow-columns.ts index b5ab26fbce6..ecfceb9be0e 100644 --- a/apps/sim/lib/table/workflow-columns.ts +++ b/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 { getPostgresConstraintName, getPostgresErrorCode, 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' @@ -733,7 +738,11 @@ export async function cancelWorkflowGroupRuns( ) if (!updated) throw new Error('Authoritative cancellation write was rejected') } catch (error) { - if (error instanceof TableRowNotFoundError) return + const rowNotFound = findCause( + error, + (cause): cause is TableRowNotFoundError => cause instanceof TableRowNotFoundError + ) + if (rowNotFound) return throw error } })