Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 40 additions & 1 deletion apps/sim/lib/table/cell-write.test.ts
Original file line number Diff line number Diff line change
@@ -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 } =
Expand Down Expand Up @@ -93,6 +94,7 @@ describe('writeWorkflowGroupState', () => {
beforeEach(() => {
vi.clearAllMocks()
resetDbChainMock()
queueTableRows(schemaMock.userTableRows, [{ id: CONTEXT.rowId }])
mockWriteExecutionsPatch.mockResolvedValue('wrote')
mockUpdateRow.mockResolvedValue({})
mockAppendTableEvent.mockResolvedValue(null)
Expand All @@ -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',
Expand Down Expand Up @@ -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', () => {
Expand Down
61 changes: 41 additions & 20 deletions apps/sim/lib/table/cell-write.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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'
}
Expand Down
7 changes: 7 additions & 0 deletions apps/sim/lib/table/rows/errors.ts
Original file line number Diff line number Diff line change
@@ -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'
}
}
5 changes: 3 additions & 2 deletions apps/sim/lib/table/rows/service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down
Loading