diff --git a/apps/sim/lib/table/cell-write.test.ts b/apps/sim/lib/table/cell-write.test.ts index 9b921cff8d1..61fbbaee7b7 100644 --- a/apps/sim/lib/table/cell-write.test.ts +++ b/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', () => { diff --git a/apps/sim/lib/table/cell-write.ts b/apps/sim/lib/table/cell-write.ts index 9c28dafffc3..4f3d9f16839 100644 --- a/apps/sim/lib/table/cell-write.ts +++ b/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' } diff --git a/apps/sim/lib/table/rows/errors.ts b/apps/sim/lib/table/rows/errors.ts new file mode 100644 index 00000000000..897dde9a0e9 --- /dev/null +++ b/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' + } +} diff --git a/apps/sim/lib/table/rows/service.ts b/apps/sim/lib/table/rows/service.ts index 91370c1aadc..a7e2cba6f35 100644 --- a/apps/sim/lib/table/rows/service.ts +++ b/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,