From 58caadbccd533e7e17dd2b4cd2b7a49543dd2eea Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 4 Nov 2026 21:14:12 -0800 Subject: [PATCH 1/9] fix(tables): keep table cells working through lost database connections --- ...w-column-execution-connection-loss.test.ts | 304 ++++++++++++++++++ .../background/workflow-column-execution.ts | 137 ++++++-- .../sim/lib/db/connection-loss.integration.ts | 138 ++++++++ apps/sim/lib/db/read-retry.test.ts | 65 +++- apps/sim/lib/db/read-retry.ts | 76 ++++- apps/sim/lib/table/cell-write.integration.ts | 255 +++++++++++++++ apps/sim/lib/table/cell-write.test.ts | 4 +- apps/sim/lib/table/cell-write.ts | 55 +++- apps/sim/lib/table/constants.ts | 13 + 9 files changed, 990 insertions(+), 57 deletions(-) create mode 100644 apps/sim/background/workflow-column-execution-connection-loss.test.ts create mode 100644 apps/sim/lib/db/connection-loss.integration.ts create mode 100644 apps/sim/lib/table/cell-write.integration.ts diff --git a/apps/sim/background/workflow-column-execution-connection-loss.test.ts b/apps/sim/background/workflow-column-execution-connection-loss.test.ts new file mode 100644 index 00000000000..c7642da9887 --- /dev/null +++ b/apps/sim/background/workflow-column-execution-connection-loss.test.ts @@ -0,0 +1,304 @@ +/** + * What a table cell does when a database socket is reset underneath it, written from the ways it + * has failed in production: + * + * - A reset on a read before the run crashed the task and left the cell `pending` for good. + * - A reset on the read that advances the cascade stranded the row's remaining groups. + * - Driver text, and sometimes SQL with bound values, was stored as the cell's visible error. + * + * The guard that must never regress runs the other way: a block has effects, so nothing here may + * start a workflow twice. What a repeated state write does is proven against real PostgreSQL in + * `lib/table/cell-write.integration.ts`. + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const mocks = vi.hoisted(() => ({ + getTableById: vi.fn(), + getRowById: vi.fn(), + getRowSummaryById: vi.fn(), + readDispatch: vi.fn(), + writeWorkflowGroupState: vi.fn(), + markWorkflowGroupPickedUp: vi.fn(), + executeWorkflow: vi.fn(), + loadDeployedWorkflowState: vi.fn(), + preprocessExecution: vi.fn(), + pickNextEligibleGroupForRow: vi.fn(), + selectWorkflow: vi.fn(), +})) + +vi.mock('@sim/utils/helpers', async (importOriginal) => ({ + ...(await importOriginal()), + sleep: vi.fn(async () => {}), +})) +vi.mock('@sim/db', () => ({ + db: { + select: () => ({ from: () => ({ where: () => ({ limit: mocks.selectWorkflow }) }) }), + }, +})) +vi.mock('@/lib/table/service', () => ({ getTableById: mocks.getTableById })) +vi.mock('@/lib/table/rows/service', () => ({ + getRowById: mocks.getRowById, + getRowSummaryById: mocks.getRowSummaryById, +})) +vi.mock('@/lib/table/rows/executions', () => ({ readStampedCapabilitySubject: vi.fn() })) +vi.mock('@/lib/table/dispatcher', () => ({ readDispatch: mocks.readDispatch })) +vi.mock('@/lib/table/workflow-columns', () => ({ + buildWorkflowGroupExecutionCorrelation: vi.fn(), + pickNextEligibleGroupForRow: mocks.pickNextEligibleGroupForRow, + stashCellContextForResume: vi.fn(), +})) +vi.mock('@/lib/table/cell-write', () => ({ + buildCancelledExecution: vi.fn(), + writeWorkflowGroupState: vi.fn(), + writeWorkflowGroupStateDurably: mocks.writeWorkflowGroupState, + markWorkflowGroupPickedUp: mocks.markWorkflowGroupPickedUp, + createWorkflowCellProgressWriter: () => ({ + onBlockStart: vi.fn(), + onBlockComplete: vi.fn(), + finish: vi.fn(async () => {}), + getEventOutputs: () => ({}), + getPendingDataPatch: () => ({ out: 'value' }), + getPendingSecretProvenance: () => ({}), + getBlockErrors: () => ({}), + }), +})) +vi.mock('@/lib/workflows/executor/execute-workflow', () => ({ + executeWorkflow: mocks.executeWorkflow, +})) +vi.mock('@/lib/workflows/persistence/utils', () => ({ + loadDeployedWorkflowState: mocks.loadDeployedWorkflowState, +})) +vi.mock('@/lib/execution/preprocessing', () => ({ + preprocessExecution: mocks.preprocessExecution, +})) +vi.mock('@/lib/core/rate-limiter/rate-limiter', () => ({ + RateLimiter: class { + checkRateLimitWithSubscription = vi.fn(async () => ({ allowed: true })) + }, +})) +vi.mock('@/lib/table/workflow-cell-result', () => ({ + classifyWorkflowCellTerminalResult: () => ({ status: 'completed', error: null }), +})) + +import { runRowCascadeLoop } from '@/background/workflow-column-execution' + +const GROUP = { id: 'group-1', workflowId: 'workflow-1', outputs: [] } +const NEXT_GROUP = { id: 'group-2', workflowId: 'workflow-2', outputs: [] } +const TABLE = { + id: 'table-1', + schema: { columns: [], workflowGroups: [GROUP, NEXT_GROUP] }, +} +const ROW = { id: 'row-1', data: {}, executions: {} } +const PAYLOAD = { + tableId: 'table-1', + tableName: 'Table', + rowId: 'row-1', + groupId: 'group-1', + workflowId: 'workflow-1', + workspaceId: 'workspace-1', + executionId: 'execution-1', + dispatchId: 'dispatch-1', + executionTimeoutMs: 60_000, + capabilityGovernedUserId: null, + billingAttribution: { + actorUserId: 'user-1', + workspaceId: 'workspace-1', + organizationId: null, + billedAccountUserId: 'user-1', + billingEntity: { type: 'user' as const, id: 'user-1' }, + billingPeriod: { start: '2026-07-01T00:00:00.000Z', end: '2026-08-01T00:00:00.000Z' }, + payerSubscription: null, + }, +} + +/** Drizzle's wrapper around a reset socket, the shape a reset read rejects with in production. */ +function resetRead(): Error { + return Object.assign( + new Error('Failed query: select "id" from "x"\nparams: secret-value', { + cause: Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET', syscall: 'read' }), + }), + { query: 'select "id" from "x"', params: ['secret-value'] } + ) +} + +/** The bare driver error a reset during `BEGIN` rejects with: no Drizzle wrapper, no cause. */ +function resetBegin(): Error { + const error = Object.assign(new Error('read ECONNRESET'), { + code: 'ECONNRESET', + syscall: 'read', + }) + for (const property of ['query', 'parameters', 'args', 'types']) { + Object.defineProperty(error, property, { value: undefined }) + } + return error +} + +const writtenStatuses = () => + mocks.writeWorkflowGroupState.mock.calls.map(([, write]) => write.executionState.status) +const lastWrite = () => mocks.writeWorkflowGroupState.mock.calls.at(-1)?.[1] + +beforeEach(() => { + vi.clearAllMocks() + mocks.getTableById.mockResolvedValue(TABLE) + mocks.getRowById.mockResolvedValue(ROW) + mocks.getRowSummaryById.mockResolvedValue({ id: 'row-1', data: {} }) + mocks.readDispatch.mockResolvedValue({ id: 'dispatch-1', status: 'active' }) + mocks.writeWorkflowGroupState.mockResolvedValue('wrote') + mocks.markWorkflowGroupPickedUp.mockResolvedValue('wrote') + mocks.executeWorkflow.mockResolvedValue({ status: 'completed', success: true, output: {} }) + mocks.loadDeployedWorkflowState.mockResolvedValue({ + blocks: { start: { id: 'start', type: 'start_trigger', subBlocks: {}, outputs: {} } }, + edges: [], + loops: {}, + parallels: {}, + }) + mocks.preprocessExecution.mockResolvedValue({ + success: true, + actorUserId: 'user-1', + actorSubscription: null, + billingAttribution: PAYLOAD.billingAttribution, + }) + mocks.pickNextEligibleGroupForRow.mockReturnValue(null) + mocks.selectWorkflow.mockResolvedValue([ + { id: 'workflow-1', userId: 'user-1', workspaceId: 'workspace-1', variables: {} }, + ]) +}) + +describe('a table cell losing a database socket', () => { + it('recovers a reset on the dispatch read and runs the cell', async () => { + mocks.readDispatch.mockRejectedValueOnce(resetRead()) + + await runRowCascadeLoop(PAYLOAD) + + expect(mocks.executeWorkflow).toHaveBeenCalledOnce() + expect(writtenStatuses()).toEqual(['completed']) + }) + + it('terminalizes the cell when the dispatch read never recovers, instead of stranding it', async () => { + mocks.readDispatch.mockRejectedValue(resetRead()) + + await expect(runRowCascadeLoop(PAYLOAD)).resolves.toBeUndefined() + + expect(mocks.executeWorkflow).not.toHaveBeenCalled() + expect(lastWrite()?.executionState.status).toBe('error') + }) + + it('fails the run loudly when the cell cannot be terminalized either', async () => { + const lost = resetBegin() + mocks.readDispatch.mockRejectedValue(resetRead()) + mocks.writeWorkflowGroupState.mockRejectedValue(lost) + + await expect(runRowCascadeLoop(PAYLOAD)).rejects.toBe(lost) + }) + + it('never restarts a workflow whose run lost its connection', async () => { + mocks.executeWorkflow.mockRejectedValue(resetRead()) + + await runRowCascadeLoop(PAYLOAD) + + expect(mocks.executeWorkflow).toHaveBeenCalledOnce() + expect(lastWrite()?.executionState.status).toBe('error') + }) + + it('marks the cell failed when its terminal write cannot be completed', async () => { + mocks.writeWorkflowGroupState.mockRejectedValueOnce(resetBegin()) + + await runRowCascadeLoop(PAYLOAD) + + expect(mocks.executeWorkflow).toHaveBeenCalledOnce() + expect(writtenStatuses()).toEqual(['completed', 'error']) + }) + + it('ends the carrier instead of starting the next group when a Stop owns the failed cell', async () => { + mocks.readDispatch.mockRejectedValue(resetRead()) + mocks.writeWorkflowGroupState.mockResolvedValue('skipped') + + await runRowCascadeLoop(PAYLOAD) + + expect(mocks.pickNextEligibleGroupForRow).not.toHaveBeenCalled() + expect(mocks.executeWorkflow).not.toHaveBeenCalled() + }) + + it('records a pre-run failure on an enrichment cell under its enrichment id', async () => { + mocks.readDispatch.mockRejectedValue(resetRead()) + mocks.getTableById.mockResolvedValue({ + ...TABLE, + schema: { + columns: [], + workflowGroups: [{ ...GROUP, type: 'enrichment', enrichmentId: 'work-email' }], + }, + }) + + await runRowCascadeLoop(PAYLOAD) + + expect(lastWrite()?.executionState).toMatchObject({ status: 'error', workflowId: 'work-email' }) + }) + + it('never stores a thrown message from before the cell reached a terminal state', async () => { + mocks.readDispatch.mockRejectedValue( + new Error( + 'Coalesced producer for "usage-gate:org-1:billed-user" did not settle within 5000ms' + ) + ) + + await runRowCascadeLoop(PAYLOAD) + + expect(lastWrite()?.executionState.error).not.toMatch(/usage-gate|org-1|billed-user/) + }) + + it('advances the cascade past a reset on the read that picks the next group', async () => { + mocks.getRowById + .mockResolvedValueOnce(ROW) + .mockRejectedValueOnce(resetRead()) + .mockResolvedValue(ROW) + mocks.pickNextEligibleGroupForRow.mockReturnValueOnce(NEXT_GROUP).mockReturnValue(null) + + await runRowCascadeLoop(PAYLOAD) + + expect(mocks.executeWorkflow).toHaveBeenCalledTimes(2) + }) + + it.each([ + ['a reset read', resetRead], + ['a reset BEGIN', resetBegin], + ])('does not store driver text or SQL as the cell error after %s', async (_name, failure) => { + mocks.executeWorkflow.mockRejectedValue(failure()) + + await runRowCascadeLoop(PAYLOAD) + + const stored = lastWrite()?.executionState.error + expect(stored).not.toMatch(/ECONNRESET|Failed query|select|secret-value/) + }) + + it('keeps a block error that names the block over a database cause', async () => { + const message = 'Agent 1: An internal error occurred while executing the block.' + mocks.executeWorkflow.mockRejectedValue(new Error(message, { cause: resetRead() })) + + await runRowCascadeLoop(PAYLOAD) + + expect(lastWrite()?.executionState.error).toBe(message) + }) + + it('does not tell the user to re-run a query that will fail the same way', async () => { + mocks.executeWorkflow.mockRejectedValue( + Object.assign(new Error('Failed query: update "x"\nparams: secret-value'), { + query: 'update "x"', + params: ['secret-value'], + cause: Object.assign(new Error('unsupported Unicode escape'), { code: '22P05' }), + }) + ) + + await runRowCascadeLoop(PAYLOAD) + + const stored = lastWrite()?.executionState.error + expect(stored).not.toMatch(/Failed query|secret-value|again/) + }) + + it('keeps the message of a failure that has nothing to do with the database', async () => { + mocks.executeWorkflow.mockRejectedValue(new Error('Workflow has no blocks')) + + await runRowCascadeLoop(PAYLOAD) + + expect(lastWrite()?.executionState.error).toBe('Workflow has no blocks') + }) +}) diff --git a/apps/sim/background/workflow-column-execution.ts b/apps/sim/background/workflow-column-execution.ts index 6f7bc2b281c..05e5d9fb2f4 100644 --- a/apps/sim/background/workflow-column-execution.ts +++ b/apps/sim/background/workflow-column-execution.ts @@ -1,7 +1,7 @@ import { db } from '@sim/db' import { tableRowExecutions, workflow as workflowTable } from '@sim/db/schema' import { createLogger, runWithRequestContext } from '@sim/logger' -import { describeError, toError } from '@sim/utils/errors' +import { classifyDatabaseFailure, describeError, toError } from '@sim/utils/errors' import { sleep } from '@sim/utils/helpers' import { generateId } from '@sim/utils/id' import { backoffWithJitter } from '@sim/utils/retry' @@ -25,6 +25,7 @@ import { } from '@/lib/core/execution-limits' import { withResourceOutboundScope } from '@/lib/core/network/resource-scope.server' import { RateLimiter } from '@/lib/core/rate-limiter/rate-limiter' +import { withDatabaseReadRetry } from '@/lib/db/read-retry' import { registerManualExecutionAborter, unregisterManualExecutionAborter, @@ -34,6 +35,7 @@ import { retryTableAdmission } from '@/lib/table/admission-retry' import { withCascadeLock } from '@/lib/table/cascade-lock' import { fillMissingColumns, mapInputValues, namedRowMapper } from '@/lib/table/cell-format' import { getColumnId } from '@/lib/table/column-keys' +import { CELL_DATABASE_RETRY } from '@/lib/table/constants' import { isEmptyCellValue, isExecCancelled } from '@/lib/table/deps' import { getMaxTableDispatchConcurrency } from '@/lib/table/dispatch-concurrency' import { appendTableEvent } from '@/lib/table/events' @@ -65,6 +67,29 @@ export type { WorkflowGroupCellPayload } const logger = createLogger('TriggerWorkflowGroupCell') +/** + * Retries in process and not through Trigger.dev on purpose. A task retry replays from the payload's + * group, which would run a billed workflow again for any group already finished. + */ +function readForCell(label: string, read: () => Promise): Promise { + return withDatabaseReadRetry(read, { label, ...CELL_DATABASE_RETRY }) +} + +/** + * The error a cell shows. Text that would expose a query, or a bare driver error, says what happened + * instead, because anyone viewing the table reads it. Anything else keeps its own message, including + * a block's already-safe error over a database cause, which names the block that failed. + */ +function getCellFailureMessage(error: unknown): string { + const failure = toError(error) + const exposesQuery = failure.message.includes('Failed query: ') + if (!exposesQuery && !Object.hasOwn(failure, 'code')) return failure.message + if (classifyDatabaseFailure(failure) === 'connection') { + return 'Lost the database connection. Run this cell again.' + } + return exposesQuery ? 'A database error stopped this cell.' : failure.message +} + /** * Fails before workflow execution when saved table mappings are incompatible * with the latest active deployment loaded for this cell run. @@ -324,9 +349,11 @@ export async function executeWorkflowGroupCellJob( // marked, so stop re-driving this row. if (outcome.result === 'blocked') break if (carrierSignal.aborted) break - const freshTable = await getTableById(tableId) + const freshTable = await readForCell('getTableById', () => getTableById(tableId)) if (!freshTable) break - const freshRow = await getRowById(tableId, rowId, workspaceId) + const freshRow = await readForCell('getRowById', () => + getRowById(tableId, rowId, workspaceId) + ) if (!freshRow) break const next = pickNextEligibleGroupForRow(freshTable, freshRow) if (!next) break @@ -352,7 +379,9 @@ export async function executeWorkflowGroupCellJob( * belongs to the person who asked for it, so take the subject off the * stamp rather than carrying our own into someone else's request. */ - capabilityGovernedUserId: await readStampedCapabilitySubject(rowId, next.id), + capabilityGovernedUserId: await readForCell('readStampedCapabilitySubject', () => + readStampedCapabilitySubject(rowId, next.id) + ), } } } finally { @@ -395,7 +424,7 @@ export async function runRowCascadeLoop( break } - const freshTable = await getTableById(tableId) + const freshTable = await readForCell('getTableById', () => getTableById(tableId)) if (!freshTable) { logger.warn(`Table ${tableId} vanished mid-cascade`) break @@ -425,7 +454,7 @@ export async function runRowCascadeLoop( // would re-pick the still-pending queued marker and spin. if (result === 'blocked') return 'blocked' - const freshRow = await getRowById(tableId, rowId, workspaceId) + const freshRow = await readForCell('getRowById', () => getRowById(tableId, rowId, workspaceId)) if (!freshRow) break const next = pickNextEligibleGroupForRow(freshTable, freshRow, currentGroupId) if (!next) break @@ -438,7 +467,9 @@ export async function runRowCascadeLoop( */ currentCapabilityGovernedUserId = nextExec?.status === 'pending' && nextExec.executionId == null - ? await readStampedCapabilitySubject(rowId, next.id) + ? await readForCell('readStampedCapabilitySubject', () => + readStampedCapabilitySubject(rowId, next.id) + ) : currentCapabilityGovernedUserId currentGroupId = next.id currentWorkflowId = next.workflowId @@ -472,6 +503,7 @@ async function runWorkflowAndWriteTerminal( buildCancelledExecution, createWorkflowCellProgressWriter, writeWorkflowGroupState, + writeWorkflowGroupStateDurably, markWorkflowGroupPickedUp, } = await import('@/lib/table/cell-write') const { classifyWorkflowCellTerminalResult } = await import( @@ -486,7 +518,7 @@ async function runWorkflowAndWriteTerminal( eventOutputs?: RowData, secretProvenance?: TableRowSecretProvenanceWrite ) => - writeWorkflowGroupState(cellCtx, { + writeWorkflowGroupStateDurably(cellCtx, { executionState, dataPatch, eventOutputs, @@ -515,7 +547,7 @@ async function runWorkflowAndWriteTerminal( */ if (dispatchId) { const { readDispatch } = await import('@/lib/table/dispatcher') - const owningDispatch = await readDispatch(dispatchId) + const owningDispatch = await readForCell('readDispatch', () => readDispatch(dispatchId)) if (!owningDispatch || owningDispatch.status === 'cancelled') { logger.info( `Skipping cell — owning dispatch is cancelled (table=${tableId} row=${rowId} group=${groupId} dispatch=${dispatchId})` @@ -561,7 +593,7 @@ async function runWorkflowAndWriteTerminal( return 'error' } - const row = await getRowById(tableId, rowId, workspaceId) + const row = await readForCell('getRowById', () => getRowById(tableId, rowId, workspaceId)) if (!row) { logger.warn(`Row ${rowId} vanished before enrichment`) return 'error' @@ -575,7 +607,9 @@ async function runWorkflowAndWriteTerminal( * Gate the exact workspace payer and member cap before hosted-key cost. * A denial clears the cell pre-stamp and surfaces the upgrade state. */ - const usage = await checkExecutionUsageLimits(enrichmentBillingAttribution) + const usage = await readForCell('checkExecutionUsageLimits', () => + checkExecutionUsageLimits(enrichmentBillingAttribution) + ) if (usage.isExceeded) { logger.warn( `Usage limit reached — halting enrichment (table=${tableId} row=${rowId} group=${groupId})` @@ -624,7 +658,9 @@ async function runWorkflowAndWriteTerminal( { userId: enrichmentBillingAttribution.actorUserId, workspaceId }, new Set(enrichmentInputMappings.map((mapping) => mapping.columnName)) ) - const inputSource = await getRowSummaryById(tableId, rowId, workspaceId, readProvenance) + const inputSource = await readForCell('getRowSummaryById', () => + getRowSummaryById(tableId, rowId, workspaceId, readProvenance) + ) if (!inputSource) { logger.warn(`Row ${rowId} vanished before enrichment input could be read`) return 'error' @@ -786,7 +822,7 @@ async function runWorkflowAndWriteTerminal( executionId, jobId: null, workflowId: statusId, - error: toError(err).message, + error: getCellFailureMessage(err), }) return 'error' } @@ -795,11 +831,9 @@ async function runWorkflowAndWriteTerminal( let progressWriter: ReturnType | null = null try { - const [workflowRecord] = await db - .select() - .from(workflowTable) - .where(eq(workflowTable.id, workflowId)) - .limit(1) + const [workflowRecord] = await readForCell('selectWorkflow', () => + db.select().from(workflowTable).where(eq(workflowTable.id, workflowId)).limit(1) + ) if (!workflowRecord) { await writeState({ @@ -814,7 +848,9 @@ async function runWorkflowAndWriteTerminal( let normalizedData: Awaited> try { - normalizedData = await loadDeployedWorkflowState(workflowId, workspaceId) + normalizedData = await readForCell('loadDeployedWorkflowState', () => + loadDeployedWorkflowState(workflowId, workspaceId) + ) assertWorkflowGroupMatchesLatestDeployment(group, normalizedData) } catch (err) { await writeState({ @@ -822,7 +858,7 @@ async function runWorkflowAndWriteTerminal( executionId, jobId: null, workflowId, - error: toError(err).message, + error: getCellFailureMessage(err), }) return 'error' } @@ -838,7 +874,7 @@ async function runWorkflowAndWriteTerminal( return 'error' } - const row = await getRowById(tableId, rowId, workspaceId) + const row = await readForCell('getRowById', () => getRowById(tableId, rowId, workspaceId)) if (!row) { logger.warn(`Row ${rowId} vanished before execution`) return 'error' @@ -993,7 +1029,9 @@ async function runWorkflowAndWriteTerminal( // fires (cancelByKey is a no-op there), so re-check the DB tombstone and // release this concurrency slot promptly instead of sleeping out the // full retry budget. - const refreshed = await getRowById(tableId, rowId, workspaceId) + const refreshed = await readForCell('getRowById', () => + getRowById(tableId, rowId, workspaceId) + ) if (!refreshed || cancelledBeforeRun(refreshed.executions?.[groupId])) { return refreshed ? 'cancelled' : 'error' } @@ -1001,10 +1039,7 @@ async function runWorkflowAndWriteTerminal( // SQL guard also rejects if a stop click stamped `cancelled` between this // check and pickup. - const pickedUp = await markWorkflowGroupPickedUp(cellCtx, { - workflowId, - jobId: null, - }) + const pickedUp = await markWorkflowGroupPickedUp(cellCtx, { workflowId, jobId: null }) if (pickedUp === 'skipped') return 'error' // Output columns produced by THIS group are skipped on input — they're @@ -1028,7 +1063,9 @@ async function runWorkflowAndWriteTerminal( ...inputMappings.map((mapping) => mapping.columnName), ]) ) - const inputSource = await getRowSummaryById(tableId, rowId, workspaceId, readProvenance) + const inputSource = await readForCell('getRowSummaryById', () => + getRowSummaryById(tableId, rowId, workspaceId, readProvenance) + ) if (!inputSource) { logger.warn(`Row ${rowId} vanished before workflow input could be read`) return 'error' @@ -1068,6 +1105,7 @@ async function runWorkflowAndWriteTerminal( progressWriter = createWorkflowCellProgressWriter({ group, signal: attemptSignal, + // The writer carries a failed write's patch into its next one, so it needs no retry here. writeProgress: ({ dataPatch, eventOutputs, @@ -1075,8 +1113,8 @@ async function runWorkflowAndWriteTerminal( runningBlockIds, blockErrors, }) => - writeState( - { + writeWorkflowGroupState(cellCtx, { + executionState: { status: 'running', executionId, jobId: null, @@ -1087,8 +1125,8 @@ async function runWorkflowAndWriteTerminal( }, dataPatch, eventOutputs, - secretProvenance - ), + secretProvenance, + }), onWriteError: (err) => { logger.warn( `Per-block partial write failed (table=${tableId} row=${rowId} group=${groupId})`, @@ -1204,11 +1242,10 @@ async function runWorkflowAndWriteTerminal( ) return terminalResult.status } catch (err) { - const message = toError(err).message logger.error( `Workflow group cell execution failed (table=${tableId} row=${rowId} group=${groupId})`, { - error: message, + error: toError(err).message, executionId, cause: describeError(err), retryable: isRetryableInfrastructureError(err), @@ -1221,7 +1258,7 @@ async function runWorkflowAndWriteTerminal( executionId, jobId: null, workflowId, - error: message, + error: getCellFailureMessage(err), runningBlockIds: [], blockErrors: progressWriter?.getBlockErrors() ?? {}, }) @@ -1235,6 +1272,38 @@ async function runWorkflowAndWriteTerminal( return 'error' } }) + } catch (err) { + /** + * Whatever got here left the cell without a terminal state, usually before anything ran. The + * dispatcher has moved on by then and no sweep revisits the cell, so a `pending` left behind + * renders as Queued for good. Reporting it as an error lets the cascade continue and the user + * re-run it. The text is fixed because anything can be thrown here, and the log holds the rest. + * + * A rejected write means a Stop or a newer run owns the cell, so this carrier ends rather than + * start the next group. If the write fails too the throw fails the run, the alert to act on. + */ + logger.error( + `Workflow group cell failed before reaching a terminal state (table=${tableId} row=${rowId} group=${groupId})`, + { + executionId, + cause: describeError(err), + retryable: isRetryableInfrastructureError(err), + } + ) + const { writeWorkflowGroupStateDurably } = await import('@/lib/table/cell-write') + const written = await writeWorkflowGroupStateDurably( + { tableId, rowId, workspaceId, groupId, executionId, requestId, table }, + { + executionState: { + status: 'error', + executionId, + jobId: null, + workflowId: (group.type === 'enrichment' && group.enrichmentId) || workflowId, + error: 'This cell stopped before it finished. Run it again.', + }, + } + ) + return written === 'wrote' ? 'error' : 'cancelled' } finally { timeoutController.cleanup() } diff --git a/apps/sim/lib/db/connection-loss.integration.ts b/apps/sim/lib/db/connection-loss.integration.ts new file mode 100644 index 00000000000..2d9fa120947 --- /dev/null +++ b/apps/sim/lib/db/connection-loss.integration.ts @@ -0,0 +1,138 @@ +/** + * Sends real TCP resets through a proxy between postgres.js and real PostgreSQL, using the pool + * options `@sim/db` uses, because the error a reset produces depends on what the connection was + * doing and no mock can stand in for that. Rows live in a session-independent scratch table. + */ +import net, { type Socket } from 'node:net' +import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure' +import { generateShortId } from '@sim/utils/id' +import { sql } from 'drizzle-orm' +import { drizzle } from 'drizzle-orm/postgres-js' +import postgres from 'postgres' +import { afterAll, beforeAll, describe, expect, it } from 'vitest' +import { withDatabaseReadRetry, withIdempotentWriteRetry } from '@/lib/db/read-retry' + +const TABLE = sql.identifier( + `connection_loss_${generateShortId(8).replace(/\W/g, '').toLowerCase()}` +) + +const live = new Set() +let proxy: net.Server +let client: ReturnType +let db: ReturnType +let admin: ReturnType +let adminDb: ReturnType + +const resetEverySocket = () => { + for (const socket of live) socket.resetAndDestroy() +} + +/** Lets the statement reach the server before the socket is reset under it. */ +const shortly = () => new Promise((resolve) => setTimeout(resolve, 75)) + +beforeAll(async () => { + const url = new URL(readTestDatabaseUrl()) + proxy = net.createServer((downstream) => { + const upstream = net.connect(Number(url.port || 5432), url.hostname) + live.add(downstream) + downstream.pipe(upstream) + upstream.pipe(downstream) + downstream.on('error', () => {}) + upstream.on('error', () => downstream.destroy()) + downstream.on('close', () => { + live.delete(downstream) + upstream.destroy() + }) + }) + await new Promise((resolve) => proxy.listen(0, '127.0.0.1', resolve)) + const { port } = proxy.address() as net.AddressInfo + const proxied = new URL(url) + proxied.hostname = '127.0.0.1' + proxied.port = String(port) + + const poolOptions = { prepare: false, fetch_types: false, idle_timeout: 20, onnotice: () => {} } + client = postgres(proxied.toString(), { ...poolOptions, max: 5 }) + db = drizzle(client) + admin = postgres(url.toString(), { ...poolOptions, max: 1 }) + adminDb = drizzle(admin) + await adminDb.execute( + sql`CREATE TABLE ${TABLE} (id text PRIMARY KEY, status text NOT NULL, writes integer NOT NULL)` + ) +}) + +afterAll(async () => { + await adminDb.execute(sql`DROP TABLE IF EXISTS ${TABLE}`) + await client.end({ timeout: 1 }) + await admin.end({ timeout: 1 }) + await new Promise((resolve) => proxy.close(resolve)) +}) + +describe('real connection resets', () => { + it('recovers a read reset while its statement is running', async () => { + let attempts = 0 + const rows = await withDatabaseReadRetry(async () => { + attempts++ + const statement = + attempts === 1 + ? db.execute(sql`SELECT pg_sleep(1), 1 AS answer`).then((result) => result) + : db.execute(sql`SELECT 1 AS answer`).then((result) => result) + if (attempts === 1) { + await shortly() + resetEverySocket() + } + return statement + }) + + expect(attempts).toBe(2) + expect(rows[0]).toMatchObject({ answer: 1 }) + }) + + it('reuses the pool after every idle socket in it is reset', async () => { + await Promise.all( + Array.from({ length: 5 }, () => db.execute(sql`SELECT pg_sleep(0.02)`).then((rows) => rows)) + ) + resetEverySocket() + + const rows = await Promise.all( + Array.from({ length: 5 }, () => + withDatabaseReadRetry(() => db.execute(sql`SELECT 1 AS answer`).then((result) => result)) + ) + ) + + expect(rows).toHaveLength(5) + }) + + it('completes a transactional write reset before it commits, applying it once', async () => { + let attempts = 0 + await withIdempotentWriteRetry(async () => { + attempts++ + await db.transaction(async (trx) => { + await trx.execute(sql` + INSERT INTO ${TABLE} (id, status, writes) VALUES ('uncommitted', 'completed', 1) + ON CONFLICT (id) DO UPDATE SET writes = ${TABLE}.writes + 1 + `) + if (attempts === 1) { + resetEverySocket() + await trx.execute(sql`SELECT pg_sleep(1)`) + } + }) + }) + + const rows = await adminDb.execute(sql`SELECT writes FROM ${TABLE} WHERE id = 'uncommitted'`) + expect(attempts).toBe(2) + expect(rows[0]).toMatchObject({ writes: 1 }) + }) + + it('does not repeat a write the database rejected', async () => { + let attempts = 0 + await expect( + withIdempotentWriteRetry(async () => { + attempts++ + await db.execute( + sql`INSERT INTO ${TABLE} (id, status, writes) VALUES ('rejected', NULL, 1)` + ) + }) + ).rejects.toThrow() + expect(attempts).toBe(1) + }) +}) diff --git a/apps/sim/lib/db/read-retry.test.ts b/apps/sim/lib/db/read-retry.test.ts index 973cbce6cae..3cd4f9545cc 100644 --- a/apps/sim/lib/db/read-retry.test.ts +++ b/apps/sim/lib/db/read-retry.test.ts @@ -1,6 +1,10 @@ import { sleep } from '@sim/utils/helpers' import { afterEach, describe, expect, it, vi } from 'vitest' -import { isTransientDatabaseReadError, withDatabaseReadRetry } from '@/lib/db/read-retry' +import { + isTransientDatabaseReadError, + withDatabaseReadRetry, + withIdempotentWriteRetry, +} from '@/lib/db/read-retry' function driverError(code: string): Error { return new Error('query wrapper', { cause: Object.assign(new Error('driver failure'), { code }) }) @@ -87,4 +91,63 @@ describe('independent database read retries', () => { await expect(withDatabaseReadRetry(read)).rejects.toBe(error) expect(read).toHaveBeenCalledOnce() }) + + it('paces retries by the caller backoff, so a budget can outlast a burst of resets', async () => { + vi.useFakeTimers() + const read = vi.fn().mockRejectedValue(driverError('ECONNRESET')) + const started = Date.now() + const result = expect( + withDatabaseReadRetry(read, { maxAttempts: 2, backoff: { baseMs: 1000, maxMs: 1000 } }) + ).rejects.toThrow() + await vi.runAllTimersAsync() + await result + expect(Date.now() - started).toBeGreaterThanOrEqual(800) + }) +}) + +describe('idempotent database write retries', () => { + /** A write's own error carries its statement, which is what marks a socket error as the database's. */ + function lostConnection(code = 'ECONNRESET'): Error { + return Object.assign(new Error('Failed query: update "x"'), { + query: 'update "x"', + params: [], + cause: Object.assign(new Error('read ECONNRESET'), { code }), + }) + } + + it('repeats a write whose connection was lost and returns its result', async () => { + vi.useFakeTimers() + const write = vi.fn().mockRejectedValueOnce(lostConnection()).mockResolvedValueOnce('wrote') + const result = withIdempotentWriteRetry(write) + await vi.runAllTimersAsync() + await expect(result).resolves.toBe('wrote') + expect(write).toHaveBeenCalledTimes(2) + }) + + it.each(['55P03', '40P01', '40001', '53300', '57014', '23505'])( + 'surfaces %s rather than holding a worker through contention or repeating a rejected write', + async (code) => { + const error = Object.assign(new Error('rejected'), { code }) + const write = vi.fn().mockRejectedValue(error) + await expect(withIdempotentWriteRetry(write)).rejects.toBe(error) + expect(write).toHaveBeenCalledOnce() + } + ) + + it('leaves a socket error that did not come from a query to the caller', async () => { + const error = Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' }) + const write = vi.fn().mockRejectedValue(error) + await expect(withIdempotentWriteRetry(write)).rejects.toBe(error) + expect(write).toHaveBeenCalledOnce() + }) + + it('stops at its attempt ceiling and preserves the final error', async () => { + vi.useFakeTimers() + const error = lostConnection() + const write = vi.fn().mockRejectedValue(error) + const result = expect(withIdempotentWriteRetry(write, { maxAttempts: 2 })).rejects.toBe(error) + await vi.runAllTimersAsync() + await result + expect(write).toHaveBeenCalledTimes(2) + }) }) diff --git a/apps/sim/lib/db/read-retry.ts b/apps/sim/lib/db/read-retry.ts index 2e95a093090..08061d5ef39 100644 --- a/apps/sim/lib/db/read-retry.ts +++ b/apps/sim/lib/db/read-retry.ts @@ -1,7 +1,7 @@ import { createLogger } from '@sim/logger' -import { getPostgresErrorCode } from '@sim/utils/errors' +import { classifyDatabaseFailure, getPostgresErrorCode } from '@sim/utils/errors' import { sleep } from '@sim/utils/helpers' -import { backoffWithJitter } from '@sim/utils/retry' +import { type BackoffOptions, backoffWithJitter } from '@sim/utils/retry' const logger = createLogger('DatabaseReadRetry') @@ -43,17 +43,21 @@ export function isTransientDatabaseReadError(error: unknown): boolean { return code !== undefined && TRANSIENT_READ_CODES.has(code) } -/** - * Retries independent reads, rebuilding each statement or fresh read-only transaction - * after the previous attempt has fully rolled back. Never retry writes, locking reads, - * or statements inside a caller-owned transaction. - * - * `maxElapsedMs` bounds retry admission, including backoff. The callback receives the - * remaining budget and must enforce it at the database to bound an in-flight statement. - */ -export async function withDatabaseReadRetry( - read: (remainingMs?: number) => Promise, - options: { label?: string; maxAttempts?: number; maxElapsedMs?: number } = {} +interface DatabaseRetryOptions { + label?: string + maxAttempts?: number + maxElapsedMs?: number + /** Pacing between attempts. The default suits a blip; a longer budget can outlast a burst. */ + backoff?: BackoffOptions +} + +const DEFAULT_RETRY_BACKOFF: BackoffOptions = { baseMs: 100, maxMs: 500 } + +async function retryDatabaseOperation( + operation: (remainingMs?: number) => Promise, + kind: 'read' | 'write', + isRetryable: (error: unknown) => boolean, + options: DatabaseRetryOptions ): Promise { const maxAttempts = options.maxAttempts ?? 3 const deadline = @@ -68,12 +72,14 @@ export async function withDatabaseReadRetry( } for (let attempt = 1; ; attempt++) { try { - return await read(deadline === undefined ? undefined : Math.max(1, deadline - Date.now())) + return await operation( + deadline === undefined ? undefined : Math.max(1, deadline - Date.now()) + ) } catch (error) { - if (attempt >= maxAttempts || !isTransientDatabaseReadError(error)) throw error - const delay = backoffWithJitter(attempt, null, { baseMs: 100, maxMs: 500 }) + if (attempt >= maxAttempts || !isRetryable(error)) throw error + const delay = backoffWithJitter(attempt, null, options.backoff ?? DEFAULT_RETRY_BACKOFF) if (deadline !== undefined && Date.now() + delay >= deadline) throw error - logger.warn('Retrying transient database read', { + logger.warn(`Retrying transient database ${kind}`, { label: options.label, attempt, code: getPostgresErrorCode(error), @@ -83,3 +89,39 @@ export async function withDatabaseReadRetry( } } } + +/** + * Retries independent reads, rebuilding each statement or fresh read-only transaction + * after the previous attempt has fully rolled back. Never retry writes, locking reads, + * or statements inside a caller-owned transaction. + * + * `maxElapsedMs` bounds retry admission, including backoff. The callback receives the + * remaining budget and must enforce it at the database to bound an in-flight statement. + */ +export function withDatabaseReadRetry( + read: (remainingMs?: number) => Promise, + options: DatabaseRetryOptions = {} +): Promise { + return retryDatabaseOperation(read, 'read', isTransientDatabaseReadError, options) +} + +/** + * Repeats a write whose database connection was lost. A lost connection leaves the outcome + * unknown, because the commit may have landed before the reset, so the write must leave the same + * state however many times it applies: use this only for a write that sets fixed values under a + * guard, never one that increments, appends, or fires an effect from inside its transaction. + * + * Only a lost connection qualifies. Lock timeouts, deadlocks, and capacity errors mean the database + * answered and is contended, and holding a worker through backoff there adds to the contention. + */ +export function withIdempotentWriteRetry( + write: () => Promise, + options: DatabaseRetryOptions = {} +): Promise { + return retryDatabaseOperation( + write, + 'write', + (error) => classifyDatabaseFailure(error) === 'connection', + options + ) +} diff --git a/apps/sim/lib/table/cell-write.integration.ts b/apps/sim/lib/table/cell-write.integration.ts new file mode 100644 index 00000000000..12f9b91b6ce --- /dev/null +++ b/apps/sim/lib/table/cell-write.integration.ts @@ -0,0 +1,255 @@ +/** + * What a cell's durable state write does when the connection dies around its commit, against the + * real schema, `updateRow`, and SQL guards. A proxy between `@sim/db` and PostgreSQL forwards the + * COMMIT, withholds the server's reply, and then sends a real TCP reset: the commit landed and the + * client cannot know it, which is the only case where repeating a write can do harm. + */ +import net, { type Socket } from 'node:net' +import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure' +import { tableBillingMock } from '@sim/testing/mocks/table-billing.mock' +import { tableEventsMock, tableEventsMockFns } from '@sim/testing/mocks/table-events.mock' +import { tableTriggerMock } from '@sim/testing/mocks/table-trigger.mock' +import { tableWorkflowColumnsMock } from '@sim/testing/mocks/table-workflow-columns.mock' +import { sleep } from '@sim/utils/helpers' +import { generateId } from '@sim/utils/id' +import postgres from 'postgres' +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest' + +vi.mock('@/lib/table/billing', () => tableBillingMock) +vi.mock('@/lib/table/events', () => tableEventsMock) +vi.mock('@/lib/table/trigger', () => tableTriggerMock) +vi.mock('@/lib/table/workflow-columns', () => tableWorkflowColumnsMock) + +type Interception = { + /** Which client statement to act on. */ + match: RegExp + /** Forward it and lose the reply (it lands), or drop it (it never reaches the server). */ + forward: boolean + /** Runs after the statement landed, before the client sees the reset. */ + beforeReset?: () => Promise + /** Runs once the client has the reset, inside the write's retry backoff. */ + afterReset?: () => Promise +} + +const realUrl = new URL(readTestDatabaseUrl()) +const control = postgres(realUrl.toString(), { max: 2, onnotice: () => {} }) +let interception: Interception | null = null + +const proxy = net.createServer((client: Socket) => { + const server = net.connect(Number(realUrl.port || 5432), realUrl.hostname) + let replyLost = false + client.on('data', (chunk) => { + const armed = interception + if (!armed?.match.test(chunk.toString('latin1'))) { + server.write(chunk) + return + } + interception = null + replyLost = true + if (armed.forward) server.write(chunk) + void (async () => { + await sleep(100) + await armed.beforeReset?.() + client.resetAndDestroy() + await armed.afterReset?.() + })() + }) + server.on('data', (chunk) => { + if (!replyLost) client.write(chunk) + }) + client.on('error', () => {}) + server.on('error', () => client.destroy()) + client.on('close', () => server.destroy()) +}) +await new Promise((resolve) => proxy.listen(0, '127.0.0.1', resolve)) +const proxied = new URL(realUrl) +proxied.hostname = '127.0.0.1' +proxied.port = String((proxy.address() as net.AddressInfo).port) +process.env.DATABASE_URL = proxied.toString() + +const { db } = await import('@sim/db') +const { userTableDefinitions, userTableRows } = await import('@sim/db/schema') +const { writeExecutionsPatch } = await import('@/lib/table/rows/executions') +const { getTableById } = await import('@/lib/table/service') +const { writeWorkflowGroupStateDurably } = await import('@/lib/table/cell-write') + +const LANDED_COMMIT = { match: /commit/i, forward: true } +const workspaceId = generateId() +const userId = generateId() +const groupId = 'group-1' +const executionId = 'execution-1' + +async function createCell() { + const tableId = generateId() + const rowId = generateId() + await db.insert(userTableDefinitions).values({ + id: tableId, + workspaceId, + name: tableId, + schema: { columns: [{ id: 'out', name: 'out', type: 'string' }] }, + createdBy: userId, + }) + await db + .insert(userTableRows) + .values({ id: rowId, tableId, workspaceId, data: { out: 'before' }, orderKey: 'a' }) + const running = { + status: 'running' as const, + executionId, + jobId: null, + workflowId: 'workflow-1', + error: null, + } + await writeExecutionsPatch(db, tableId, rowId, { [groupId]: running }) + const table = await getTableById(tableId) + if (!table) throw new Error('Fixture table was not created') + return { ctx: { tableId, rowId, workspaceId, groupId, executionId, table }, tableId, rowId } +} + +const completed = (dataPatch: { out: string }) => ({ + dataPatch, + executionState: { + status: 'completed' as const, + executionId, + jobId: null, + workflowId: 'workflow-1', + error: null, + }, +}) + +async function readCell(rowId: string) { + const [row] = await control< + { out: string; status: string | null; execution_id: string | null }[] + >` + SELECT r.data->>'out' AS out, e.status, e.execution_id + FROM user_table_rows r + LEFT JOIN table_row_executions e ON e.row_id = r.id AND e.group_id = ${groupId} + WHERE r.id = ${rowId}` + return row +} + +beforeAll(async () => { + await control`INSERT INTO "user" (id, name, email, email_verified, created_at, updated_at) + VALUES (${userId}, 'Cell write fixture', ${`${userId}@example.test`}, true, now(), now())` + await control.begin(async (tx) => { + await tx`INSERT INTO project (id, name, owner_id) VALUES (${workspaceId}, 'Fixture project', ${userId})` + await tx`INSERT INTO workspace (id, project_id, name, owner_id, billed_account_user_id) + VALUES (${workspaceId}, ${workspaceId}, 'Cell write fixtures', ${userId}, ${userId})` + }) +}) + +beforeEach(() => { + vi.clearAllMocks() + interception = null +}) + +afterAll(async () => { + await control.begin(async (tx) => { + await tx`DELETE FROM workspace WHERE id = ${workspaceId}` + await tx`DELETE FROM project WHERE id = ${workspaceId}` + await tx`DELETE FROM "user" WHERE id = ${userId}` + }) + await control.end({ timeout: 1 }) + proxy.close() +}) + +describe('a durable cell state write around a lost connection', { timeout: 90_000 }, () => { + it('applies a write whose statement never reached the server', async () => { + const { ctx, rowId } = await createCell() + interception = { match: /commit/i, forward: false } + + await expect(writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }))).resolves.toBe( + 'wrote' + ) + + expect(await readCell(rowId)).toMatchObject({ out: 'result', status: 'completed' }) + expect(tableEventsMockFns.mockAppendTableEvent).toHaveBeenCalledOnce() + }) + + it('recognizes a commit that landed and still tells the client', async () => { + const { ctx, rowId } = await createCell() + interception = LANDED_COMMIT + + await expect(writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }))).resolves.toBe( + 'wrote' + ) + + expect(await readCell(rowId)).toMatchObject({ out: 'result', status: 'completed' }) + expect(tableEventsMockFns.mockAppendTableEvent).toHaveBeenCalledOnce() + expect(tableEventsMockFns.mockAppendTableEvent).toHaveBeenCalledWith( + expect.objectContaining({ status: 'completed', outputs: { out: 'result' } }) + ) + }) + + it('does not undo an edit the user made after the commit landed', async () => { + const { ctx, rowId } = await createCell() + interception = { + ...LANDED_COMMIT, + beforeReset: async () => { + await control`UPDATE user_table_rows SET data = '{"out":"edited"}' WHERE id = ${rowId}` + }, + } + + await writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' })) + + expect((await readCell(rowId)).out).toBe('edited') + }) + + it('does not bring back a cell the user cleared after the commit landed', async () => { + const { ctx, rowId } = await createCell() + interception = { + ...LANDED_COMMIT, + beforeReset: async () => { + await control`DELETE FROM table_row_executions WHERE row_id = ${rowId}` + }, + } + + await expect(writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }))).resolves.toBe( + 'skipped' + ) + + expect((await readCell(rowId)).status).toBeNull() + }) + + it('does not overwrite a Stop that lands during the retry backoff', async () => { + const { ctx, rowId } = await createCell() + interception = { + match: /commit/i, + forward: false, + afterReset: async () => { + await control`UPDATE table_row_executions SET status = 'cancelled' WHERE row_id = ${rowId}` + }, + } + + await expect(writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }))).resolves.toBe( + 'skipped' + ) + + expect((await readCell(rowId)).status).toBe('cancelled') + }) + + it('does not repeat a cancel stamp, which could kill a run the user just started', async () => { + const { ctx, rowId } = await createCell() + interception = { + ...LANDED_COMMIT, + beforeReset: async () => { + await control` + UPDATE table_row_executions SET status = 'pending', execution_id = NULL + WHERE row_id = ${rowId}` + }, + } + + await expect( + writeWorkflowGroupStateDurably(ctx, { + executionState: { + status: 'cancelled', + executionId, + jobId: null, + workflowId: 'workflow-1', + error: 'Cancelled', + }, + }) + ).rejects.toThrow() + + expect(await readCell(rowId)).toMatchObject({ status: 'pending', execution_id: null }) + }) +}) diff --git a/apps/sim/lib/table/cell-write.test.ts b/apps/sim/lib/table/cell-write.test.ts index 8cc1d9d9d08..76b3ce8ebd3 100644 --- a/apps/sim/lib/table/cell-write.test.ts +++ b/apps/sim/lib/table/cell-write.test.ts @@ -9,7 +9,8 @@ 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 { mockWriteExecutionsPatch } = vi.hoisted(() => ({ +const { mockLoadExecutionsForRow, mockWriteExecutionsPatch } = vi.hoisted(() => ({ + mockLoadExecutionsForRow: vi.fn(), mockWriteExecutionsPatch: vi.fn(), })) @@ -18,6 +19,7 @@ vi.mock('@/lib/core/security/encryption', () => encryptionMock) vi.mock('@/lib/table/events', () => tableEventsMock) vi.mock('@/lib/table/rows/executions', () => ({ + loadExecutionsForRow: mockLoadExecutionsForRow, writeExecutionsPatch: mockWriteExecutionsPatch, })) diff --git a/apps/sim/lib/table/cell-write.ts b/apps/sim/lib/table/cell-write.ts index 1acba3c595e..56fb349666d 100644 --- a/apps/sim/lib/table/cell-write.ts +++ b/apps/sim/lib/table/cell-write.ts @@ -14,9 +14,11 @@ import { userTableRows } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { getValueAtPath } from '@sim/utils/object' import { and, eq } from 'drizzle-orm' +import { withIdempotentWriteRetry } from '@/lib/db/read-retry' +import { CELL_DATABASE_RETRY } from '@/lib/table/constants' import { appendTableEvent } from '@/lib/table/events' import { TableRowNotFoundError } from '@/lib/table/rows/errors' -import { writeExecutionsPatch } from '@/lib/table/rows/executions' +import { loadExecutionsForRow, writeExecutionsPatch } from '@/lib/table/rows/executions' import { createTableRowSecretProvenanceFromEncryptedExecution, createUnknownTableRowSecretProvenance, @@ -136,6 +138,15 @@ export async function writeWorkflowGroupState( return 'skipped' } + emitCellStateEvent(ctx, payload) + + return 'wrote' +} + +function emitCellStateEvent( + { tableId, rowId, groupId, table }: WriteWorkflowGroupContext, + payload: WriteWorkflowGroupStatePayload +): void { // The SSE snapshot must carry what was persisted, not the raw block output. // `updateRow` coerces its own merged copy, so the patch object here still // holds the workflow's value — for a `select` column that's the option *name* @@ -145,7 +156,7 @@ export async function writeWorkflowGroupState( // bookkeeping, so it must not be mutated. The `null` policy mirrors what // `updateRow` persists for a computed write, so the snapshot the client sees // and the row on disk agree about a block output its column cannot hold. - const rawEventOutputs = payload.eventOutputs ?? dataPatch + const rawEventOutputs = payload.eventOutputs ?? payload.dataPatch const hasOutputs = rawEventOutputs && Object.keys(rawEventOutputs).length > 0 const eventOutputs = hasOutputs ? { ...rawEventOutputs } : rawEventOutputs if (hasOutputs && eventOutputs) coerceRowValues(eventOutputs, table.schema, 'null') @@ -164,8 +175,44 @@ export async function writeWorkflowGroupState( ...(runningBlockIds && runningBlockIds.length > 0 ? { runningBlockIds } : {}), ...(blockErrors && Object.keys(blockErrors).length > 0 ? { blockErrors } : {}), }) +} - return 'wrote' +/** + * {@link writeWorkflowGroupState} that survives a lost database connection, for writes a cell cannot + * afford to drop: its pickup and terminal states. + * + * A lost connection leaves the outcome unknown, so each repeat first checks what the last attempt + * did. Re-applying a commit that landed could undo what the user did since, such as editing the + * output or stopping a downstream group. A cell whose state already matches only needs its event, + * which the lost attempt never emitted, and a terminal state with no row left was cleared by the + * user, whose write it must not bring back. + * + * A `cancelled` stamp is never repeated. It carries no guard, so a repeat could kill a run the user + * started after it. + */ +export async function writeWorkflowGroupStateDurably( + ctx: WriteWorkflowGroupContext, + payload: WriteWorkflowGroupStatePayload +): Promise<'wrote' | 'skipped'> { + const { status } = payload.executionState + if (status === 'cancelled') return writeWorkflowGroupState(ctx, payload) + + let attempted = false + return withIdempotentWriteRetry( + async () => { + if (attempted) { + const current = (await loadExecutionsForRow(db, ctx.rowId))[ctx.groupId] + if (current?.executionId === ctx.executionId && current.status === status) { + emitCellStateEvent(ctx, payload) + return 'wrote' + } + if (!current && status !== 'running') return 'skipped' + } + attempted = true + return writeWorkflowGroupState(ctx, payload) + }, + { label: 'writeWorkflowGroupState', ...CELL_DATABASE_RETRY } + ) } export interface WorkflowCellProgressWrite { @@ -366,7 +413,7 @@ export async function markWorkflowGroupPickedUp( ctx: WriteWorkflowGroupContext, prev: Pick ): Promise<'wrote' | 'skipped'> { - return writeWorkflowGroupState(ctx, { + return writeWorkflowGroupStateDurably(ctx, { executionState: { status: 'running', executionId: ctx.executionId, diff --git a/apps/sim/lib/table/constants.ts b/apps/sim/lib/table/constants.ts index 5560877e0e5..eef9f2f6af5 100644 --- a/apps/sim/lib/table/constants.ts +++ b/apps/sim/lib/table/constants.ts @@ -518,3 +518,16 @@ export function generateUniqueTableName(existingNames: string[]): string { const suffix = randomInt(100, 1000) return `${adj.toLowerCase()}_${noun.toLowerCase()}_${suffix}` } + +/** + * How a cell rides out lost database connections. A reset takes individual sockets and a fresh one + * works, but resets arrive in bursts of about a second, so backoff totals about four seconds to + * outlast one. A reset can also stall the replacement connection until `connect_timeout`, so + * `maxElapsedMs` admits one more attempt after such a stall and no more, which bounds how long a + * failing cell holds its concurrency slot and cascade lock. + */ +export const CELL_DATABASE_RETRY = { + maxAttempts: 5, + backoff: { baseMs: 250, maxMs: 2_000 }, + maxElapsedMs: 40_000, +} From 8deed030c3af52693ad44fbb73a8e5baa580c9a1 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 4 Nov 2026 21:20:19 -0800 Subject: [PATCH 2/9] fix(tables): use the central test mocks in the cell connection tests --- ...w-column-execution-connection-loss.test.ts | 88 ++++++++++--------- apps/sim/lib/table/cell-write.integration.ts | 1 - 2 files changed, 47 insertions(+), 42 deletions(-) diff --git a/apps/sim/background/workflow-column-execution-connection-loss.test.ts b/apps/sim/background/workflow-column-execution-connection-loss.test.ts index c7642da9887..4b36658bc2c 100644 --- a/apps/sim/background/workflow-column-execution-connection-loss.test.ts +++ b/apps/sim/background/workflow-column-execution-connection-loss.test.ts @@ -10,48 +10,60 @@ * start a workflow twice. What a repeated state write does is proven against real PostgreSQL in * `lib/table/cell-write.integration.ts`. */ +import { dbChainMockFns, resetDbChainMock } from '@sim/testing' +import { + executeWorkflowMock, + executeWorkflowMockFns, +} from '@sim/testing/mocks/execute-workflow.mock' +import { + executionPreprocessingMock, + executionPreprocessingMockFns, +} from '@sim/testing/mocks/execution-preprocessing.mock' +import { + tableRowsServiceMock, + tableRowsServiceMockFns, +} from '@sim/testing/mocks/table-rows-service.mock' +import { tableServiceMock, tableServiceMockFns } from '@sim/testing/mocks/table-service.mock' +import { + tableWorkflowColumnsMock, + tableWorkflowColumnsMockFns, +} from '@sim/testing/mocks/table-workflow-columns.mock' +import { utilsHelpersMock } from '@sim/testing/mocks/utils-helpers.mock' +import { + workflowsPersistenceUtilsMock, + workflowsPersistenceUtilsMockFns, +} from '@sim/testing/mocks/workflows-persistence-utils.mock' import { beforeEach, describe, expect, it, vi } from 'vitest' -const mocks = vi.hoisted(() => ({ - getTableById: vi.fn(), - getRowById: vi.fn(), - getRowSummaryById: vi.fn(), +const local = vi.hoisted(() => ({ readDispatch: vi.fn(), writeWorkflowGroupState: vi.fn(), markWorkflowGroupPickedUp: vi.fn(), - executeWorkflow: vi.fn(), - loadDeployedWorkflowState: vi.fn(), - preprocessExecution: vi.fn(), - pickNextEligibleGroupForRow: vi.fn(), - selectWorkflow: vi.fn(), })) -vi.mock('@sim/utils/helpers', async (importOriginal) => ({ - ...(await importOriginal()), - sleep: vi.fn(async () => {}), -})) -vi.mock('@sim/db', () => ({ - db: { - select: () => ({ from: () => ({ where: () => ({ limit: mocks.selectWorkflow }) }) }), - }, -})) -vi.mock('@/lib/table/service', () => ({ getTableById: mocks.getTableById })) -vi.mock('@/lib/table/rows/service', () => ({ - getRowById: mocks.getRowById, - getRowSummaryById: mocks.getRowSummaryById, -})) +const mocks = { + ...local, + getTableById: tableServiceMockFns.mockGetTableById, + getRowById: tableRowsServiceMockFns.mockGetRowById, + getRowSummaryById: tableRowsServiceMockFns.mockGetRowSummaryById, + executeWorkflow: executeWorkflowMockFns.mockExecuteWorkflow, + loadDeployedWorkflowState: workflowsPersistenceUtilsMockFns.mockLoadDeployedWorkflowState, + preprocessExecution: executionPreprocessingMockFns.mockPreprocessExecution, + pickNextEligibleGroupForRow: tableWorkflowColumnsMockFns.mockPickNextEligibleGroupForRow, + selectWorkflow: dbChainMockFns.limit, +} + +vi.mock('@sim/utils/helpers', () => utilsHelpersMock) +vi.mock('@/lib/table/service', () => tableServiceMock) +vi.mock('@/lib/table/rows/service', () => tableRowsServiceMock) vi.mock('@/lib/table/rows/executions', () => ({ readStampedCapabilitySubject: vi.fn() })) -vi.mock('@/lib/table/dispatcher', () => ({ readDispatch: mocks.readDispatch })) -vi.mock('@/lib/table/workflow-columns', () => ({ - buildWorkflowGroupExecutionCorrelation: vi.fn(), - pickNextEligibleGroupForRow: mocks.pickNextEligibleGroupForRow, - stashCellContextForResume: vi.fn(), -})) +vi.mock('@/lib/table/dispatcher', () => ({ readDispatch: local.readDispatch })) +vi.mock('@/lib/table/workflow-columns', () => tableWorkflowColumnsMock) vi.mock('@/lib/table/cell-write', () => ({ buildCancelledExecution: vi.fn(), writeWorkflowGroupState: vi.fn(), - writeWorkflowGroupStateDurably: mocks.writeWorkflowGroupState, - markWorkflowGroupPickedUp: mocks.markWorkflowGroupPickedUp, + writeWorkflowGroupStateDurably: local.writeWorkflowGroupState, + markWorkflowGroupPickedUp: local.markWorkflowGroupPickedUp, createWorkflowCellProgressWriter: () => ({ onBlockStart: vi.fn(), onBlockComplete: vi.fn(), @@ -62,15 +74,9 @@ vi.mock('@/lib/table/cell-write', () => ({ getBlockErrors: () => ({}), }), })) -vi.mock('@/lib/workflows/executor/execute-workflow', () => ({ - executeWorkflow: mocks.executeWorkflow, -})) -vi.mock('@/lib/workflows/persistence/utils', () => ({ - loadDeployedWorkflowState: mocks.loadDeployedWorkflowState, -})) -vi.mock('@/lib/execution/preprocessing', () => ({ - preprocessExecution: mocks.preprocessExecution, -})) +vi.mock('@/lib/workflows/executor/execute-workflow', () => executeWorkflowMock) +vi.mock('@/lib/workflows/persistence/utils', () => workflowsPersistenceUtilsMock) +vi.mock('@/lib/execution/preprocessing', () => executionPreprocessingMock) vi.mock('@/lib/core/rate-limiter/rate-limiter', () => ({ RateLimiter: class { checkRateLimitWithSubscription = vi.fn(async () => ({ allowed: true })) @@ -138,7 +144,7 @@ const writtenStatuses = () => const lastWrite = () => mocks.writeWorkflowGroupState.mock.calls.at(-1)?.[1] beforeEach(() => { - vi.clearAllMocks() + resetDbChainMock() mocks.getTableById.mockResolvedValue(TABLE) mocks.getRowById.mockResolvedValue(ROW) mocks.getRowSummaryById.mockResolvedValue({ id: 'row-1', data: {} }) diff --git a/apps/sim/lib/table/cell-write.integration.ts b/apps/sim/lib/table/cell-write.integration.ts index 12f9b91b6ce..8b3fc7da4b7 100644 --- a/apps/sim/lib/table/cell-write.integration.ts +++ b/apps/sim/lib/table/cell-write.integration.ts @@ -138,7 +138,6 @@ beforeAll(async () => { }) beforeEach(() => { - vi.clearAllMocks() interception = null }) From 4caa1db91915a27cc04440b52a47f2cc578b4975 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 4 Nov 2026 21:25:05 -0800 Subject: [PATCH 3/9] fix(tables): close the remaining ways a lost connection could leave a cell unfinished --- ...w-column-execution-connection-loss.test.ts | 58 ++++++++++++++++++- .../background/workflow-column-execution.ts | 50 +++++++++++++--- .../sim/lib/db/connection-loss.integration.ts | 5 +- apps/sim/lib/table/cell-write.integration.ts | 28 ++++++++- apps/sim/lib/table/cell-write.ts | 38 ++++++++++-- apps/sim/lib/table/constants.ts | 2 +- 6 files changed, 162 insertions(+), 19 deletions(-) diff --git a/apps/sim/background/workflow-column-execution-connection-loss.test.ts b/apps/sim/background/workflow-column-execution-connection-loss.test.ts index 4b36658bc2c..1ebb4c8d2c3 100644 --- a/apps/sim/background/workflow-column-execution-connection-loss.test.ts +++ b/apps/sim/background/workflow-column-execution-connection-loss.test.ts @@ -86,7 +86,10 @@ vi.mock('@/lib/table/workflow-cell-result', () => ({ classifyWorkflowCellTerminalResult: () => ({ status: 'completed', error: null }), })) -import { runRowCascadeLoop } from '@/background/workflow-column-execution' +import { + executeWorkflowGroupCellJob, + runRowCascadeLoop, +} from '@/background/workflow-column-execution' const GROUP = { id: 'group-1', workflowId: 'workflow-1', outputs: [] } const NEXT_GROUP = { id: 'group-2', workflowId: 'workflow-2', outputs: [] } @@ -215,6 +218,36 @@ describe('a table cell losing a database socket', () => { expect(writtenStatuses()).toEqual(['completed', 'error']) }) + it('fails the run when neither the terminal write nor the error write completes', async () => { + const lost = resetBegin() + mocks.writeWorkflowGroupState.mockRejectedValue(lost) + + await expect(runRowCascadeLoop(PAYLOAD)).rejects.toBe(lost) + + expect(mocks.executeWorkflow).toHaveBeenCalledOnce() + }) + + it('terminalizes the marker when the read that starts a group never recovers, and still fails', async () => { + const failure = resetRead() + mocks.getTableById.mockRejectedValue(failure) + dbChainMockFns.returning.mockResolvedValueOnce([{ rowId: 'row-1' }]) + + await expect(executeWorkflowGroupCellJob(PAYLOAD)).rejects.toBe(failure) + + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ status: 'error', executionId: 'execution-1' }) + ) + expect(mocks.executeWorkflow).not.toHaveBeenCalled() + }) + + it('reports the original failure when the marker cannot be terminalized either', async () => { + const failure = resetRead() + mocks.getTableById.mockRejectedValue(failure) + dbChainMockFns.returning.mockRejectedValue(resetBegin()) + + await expect(executeWorkflowGroupCellJob(PAYLOAD)).rejects.toBe(failure) + }) + it('ends the carrier instead of starting the next group when a Stop owns the failed cell', async () => { mocks.readDispatch.mockRejectedValue(resetRead()) mocks.writeWorkflowGroupState.mockResolvedValue('skipped') @@ -300,6 +333,29 @@ describe('a table cell losing a database socket', () => { expect(stored).not.toMatch(/Failed query|secret-value|again/) }) + it('does not store the text of a bare driver error that is not a connection failure', async () => { + mocks.executeWorkflow.mockRejectedValue( + Object.assign(new Error('duplicate key value violates unique constraint "secret_index"'), { + name: 'PostgresError', + code: '23505', + }) + ) + + await runRowCascadeLoop(PAYLOAD) + + expect(lastWrite()?.executionState.error).not.toMatch(/secret_index/) + }) + + it('keeps the message of a coded failure that did not come from the database', async () => { + mocks.executeWorkflow.mockRejectedValue( + Object.assign(new Error('Could not read the attached file'), { code: 'ENOENT' }) + ) + + await runRowCascadeLoop(PAYLOAD) + + expect(lastWrite()?.executionState.error).toBe('Could not read the attached file') + }) + it('keeps the message of a failure that has nothing to do with the database', async () => { mocks.executeWorkflow.mockRejectedValue(new Error('Workflow has no blocks')) diff --git a/apps/sim/background/workflow-column-execution.ts b/apps/sim/background/workflow-column-execution.ts index 05e5d9fb2f4..4b5d16fc800 100644 --- a/apps/sim/background/workflow-column-execution.ts +++ b/apps/sim/background/workflow-column-execution.ts @@ -25,7 +25,7 @@ import { } from '@/lib/core/execution-limits' import { withResourceOutboundScope } from '@/lib/core/network/resource-scope.server' import { RateLimiter } from '@/lib/core/rate-limiter/rate-limiter' -import { withDatabaseReadRetry } from '@/lib/db/read-retry' +import { withDatabaseReadRetry, withIdempotentWriteRetry } from '@/lib/db/read-retry' import { registerManualExecutionAborter, unregisterManualExecutionAborter, @@ -75,6 +75,8 @@ function readForCell(label: string, read: () => Promise): Promise { return withDatabaseReadRetry(read, { label, ...CELL_DATABASE_RETRY }) } +const CELL_STOPPED_MESSAGE = 'This cell stopped before it finished. Run it again.' + /** * The error a cell shows. Text that would expose a query, or a bare driver error, says what happened * instead, because anyone viewing the table reads it. Anything else keeps its own message, including @@ -84,10 +86,11 @@ function getCellFailureMessage(error: unknown): string { const failure = toError(error) const exposesQuery = failure.message.includes('Failed query: ') if (!exposesQuery && !Object.hasOwn(failure, 'code')) return failure.message - if (classifyDatabaseFailure(failure) === 'connection') { - return 'Lost the database connection. Run this cell again.' - } - return exposesQuery ? 'A database error stopped this cell.' : failure.message + const failureClass = classifyDatabaseFailure(failure) + if (failureClass === 'connection') return 'Lost the database connection. Run this cell again.' + const isDatabaseFailure = + exposesQuery || failureClass !== 'permanent' || failure.name === 'PostgresError' + return isDatabaseFailure ? 'A database error stopped this cell.' : failure.message } /** @@ -198,6 +201,13 @@ export async function terminalizeAbortedQueuedCarrierMarker( timedOut: isTimeoutAbortReason(signal.reason), timeoutMs, }) + return terminalizeQueuedMarker(payload, executionState) +} + +async function terminalizeQueuedMarker( + payload: QueuedWorkflowGroupCellPayload, + executionState: RowExecutionMetadata +): Promise { const [updated] = await db .update(tableRowExecutions) .set({ @@ -322,6 +332,7 @@ export async function executeWorkflowGroupCellJob( ) { const carrierTimeoutController = createWorkflowGroupCarrierTimeoutController(payload, signal) const carrierSignal = carrierTimeoutController.signal + let currentPayload = payload try { const { tableId, rowId, workspaceId } = payload const { getTableById } = await import('@/lib/table/service') @@ -329,7 +340,6 @@ export async function executeWorkflowGroupCellJob( const { pickNextEligibleGroupForRow } = await import('@/lib/table/workflow-columns') const { readStampedCapabilitySubject } = await import('@/lib/table/rows/executions') - let currentPayload = payload while (true) { if (carrierSignal.aborted) { await terminalizeAbortedQueuedCarrierMarker(currentPayload, carrierSignal) @@ -384,6 +394,31 @@ export async function executeWorkflowGroupCellJob( ), } } + } catch (err) { + /** + * A read that ran out of retries outside a group's own handling, such as the table lookup that + * starts a group, leaves this carrier's marker unclaimed with nothing to revisit it. The throw + * still fails the run, because the database stayed unreachable through a full retry budget. + */ + const { executionId, enrichmentId, workflowId } = currentPayload + await withIdempotentWriteRetry( + () => + terminalizeQueuedMarker(currentPayload, { + status: 'error', + executionId, + jobId: null, + workflowId: enrichmentId ?? workflowId, + error: CELL_STOPPED_MESSAGE, + runningBlockIds: [], + }), + { label: 'terminalizeQueuedMarker', ...CELL_DATABASE_RETRY } + ).catch((markerErr) => { + logger.error('Failed to terminalize the queued marker of a failed cell', { + executionId, + cause: describeError(markerErr), + }) + }) + throw err } finally { carrierTimeoutController.cleanup() } @@ -1268,6 +1303,7 @@ async function runWorkflowAndWriteTerminal( cause: describeError(writeErr), retryable: isRetryableInfrastructureError(writeErr), }) + throw writeErr } return 'error' } @@ -1299,7 +1335,7 @@ async function runWorkflowAndWriteTerminal( executionId, jobId: null, workflowId: (group.type === 'enrichment' && group.enrichmentId) || workflowId, - error: 'This cell stopped before it finished. Run it again.', + error: CELL_STOPPED_MESSAGE, }, } ) diff --git a/apps/sim/lib/db/connection-loss.integration.ts b/apps/sim/lib/db/connection-loss.integration.ts index 2d9fa120947..7da697ffc98 100644 --- a/apps/sim/lib/db/connection-loss.integration.ts +++ b/apps/sim/lib/db/connection-loss.integration.ts @@ -5,6 +5,7 @@ */ import net, { type Socket } from 'node:net' import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure' +import { sleep } from '@sim/utils/helpers' import { generateShortId } from '@sim/utils/id' import { sql } from 'drizzle-orm' import { drizzle } from 'drizzle-orm/postgres-js' @@ -28,12 +29,12 @@ const resetEverySocket = () => { } /** Lets the statement reach the server before the socket is reset under it. */ -const shortly = () => new Promise((resolve) => setTimeout(resolve, 75)) +const shortly = () => sleep(75) beforeAll(async () => { const url = new URL(readTestDatabaseUrl()) proxy = net.createServer((downstream) => { - const upstream = net.connect(Number(url.port || 5432), url.hostname) + const upstream = net.connect(Number(url.port || 5432), url.hostname.replace(/^\[|\]$/g, '')) live.add(downstream) downstream.pipe(upstream) upstream.pipe(downstream) diff --git a/apps/sim/lib/table/cell-write.integration.ts b/apps/sim/lib/table/cell-write.integration.ts index 8b3fc7da4b7..b334a42c416 100644 --- a/apps/sim/lib/table/cell-write.integration.ts +++ b/apps/sim/lib/table/cell-write.integration.ts @@ -36,7 +36,7 @@ const control = postgres(realUrl.toString(), { max: 2, onnotice: () => {} }) let interception: Interception | null = null const proxy = net.createServer((client: Socket) => { - const server = net.connect(Number(realUrl.port || 5432), realUrl.hostname) + const server = net.connect(Number(realUrl.port || 5432), realUrl.hostname.replace(/^\[|\]$/g, '')) let replyLost = false client.on('data', (chunk) => { const armed = interception @@ -79,7 +79,7 @@ const userId = generateId() const groupId = 'group-1' const executionId = 'execution-1' -async function createCell() { +async function createCell({ pickedUp = true } = {}) { const tableId = generateId() const rowId = generateId() await db.insert(userTableDefinitions).values({ @@ -99,7 +99,7 @@ async function createCell() { workflowId: 'workflow-1', error: null, } - await writeExecutionsPatch(db, tableId, rowId, { [groupId]: running }) + if (pickedUp) await writeExecutionsPatch(db, tableId, rowId, { [groupId]: running }) const table = await getTableById(tableId) if (!table) throw new Error('Fixture table was not created') return { ctx: { tableId, rowId, workspaceId, groupId, executionId, table }, tableId, rowId } @@ -191,6 +191,9 @@ describe('a durable cell state write around a lost connection', { timeout: 90_00 await writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' })) expect((await readCell(rowId)).out).toBe('edited') + expect(tableEventsMockFns.mockAppendTableEvent).toHaveBeenCalledWith( + expect.objectContaining({ status: 'completed', outputs: { out: 'edited' } }) + ) }) it('does not bring back a cell the user cleared after the commit landed', async () => { @@ -209,6 +212,25 @@ describe('a durable cell state write around a lost connection', { timeout: 90_00 expect((await readCell(rowId)).status).toBeNull() }) + it('writes an error for a cell that never had a state, not mistaking it for a cleared one', async () => { + const { ctx, rowId } = await createCell({ pickedUp: false }) + interception = { match: /commit/i, forward: false } + + await expect( + writeWorkflowGroupStateDurably(ctx, { + executionState: { + status: 'error', + executionId, + jobId: null, + workflowId: 'workflow-1', + error: 'Workflow not found', + }, + }) + ).resolves.toBe('wrote') + + expect((await readCell(rowId)).status).toBe('error') + }) + it('does not overwrite a Stop that lands during the retry backoff', async () => { const { ctx, rowId } = await createCell() interception = { diff --git a/apps/sim/lib/table/cell-write.ts b/apps/sim/lib/table/cell-write.ts index 56fb349666d..64c838315da 100644 --- a/apps/sim/lib/table/cell-write.ts +++ b/apps/sim/lib/table/cell-write.ts @@ -177,15 +177,40 @@ function emitCellStateEvent( }) } +/** The stored values of a write's outputs, which may have changed since the write landed. */ +async function readStoredOutputs( + { rowId, tableId, workspaceId }: WriteWorkflowGroupContext, + payload: WriteWorkflowGroupStatePayload +): Promise { + const [row] = await db + .select({ data: userTableRows.data }) + .from(userTableRows) + .where( + and( + eq(userTableRows.id, rowId), + eq(userTableRows.tableId, tableId), + eq(userTableRows.workspaceId, workspaceId) + ) + ) + .limit(1) + const stored = (row?.data ?? {}) as RowData + const outputs: RowData = {} + for (const columnId of Object.keys(payload.eventOutputs ?? payload.dataPatch ?? {})) { + if (Object.hasOwn(stored, columnId)) outputs[columnId] = stored[columnId] + } + return outputs +} + /** * {@link writeWorkflowGroupState} that survives a lost database connection, for writes a cell cannot * afford to drop: its pickup and terminal states. * * A lost connection leaves the outcome unknown, so each repeat first checks what the last attempt * did. Re-applying a commit that landed could undo what the user did since, such as editing the - * output or stopping a downstream group. A cell whose state already matches only needs its event, - * which the lost attempt never emitted, and a terminal state with no row left was cleared by the - * user, whose write it must not bring back. + * output or stopping a downstream group, so a cell whose state already matches only needs the event + * the lost attempt never emitted, carrying what is stored now. A `completed` cell with no state left + * was cleared by the user, since a run only completes after its pickup wrote one. Any other state + * can legitimately be the first the cell ever had, so it is written. * * A `cancelled` stamp is never repeated. It carries no guard, so a repeat could kill a run the user * started after it. @@ -203,10 +228,13 @@ export async function writeWorkflowGroupStateDurably( if (attempted) { const current = (await loadExecutionsForRow(db, ctx.rowId))[ctx.groupId] if (current?.executionId === ctx.executionId && current.status === status) { - emitCellStateEvent(ctx, payload) + emitCellStateEvent(ctx, { + ...payload, + eventOutputs: await readStoredOutputs(ctx, payload), + }) return 'wrote' } - if (!current && status !== 'running') return 'skipped' + if (!current && status === 'completed') return 'skipped' } attempted = true return writeWorkflowGroupState(ctx, payload) diff --git a/apps/sim/lib/table/constants.ts b/apps/sim/lib/table/constants.ts index eef9f2f6af5..66bacfefa71 100644 --- a/apps/sim/lib/table/constants.ts +++ b/apps/sim/lib/table/constants.ts @@ -530,4 +530,4 @@ export const CELL_DATABASE_RETRY = { maxAttempts: 5, backoff: { baseMs: 250, maxMs: 2_000 }, maxElapsedMs: 40_000, -} +} as const From d6d2e6ccba0edf888615fb06e6fd3a5753ba3536 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 4 Nov 2026 21:32:57 -0800 Subject: [PATCH 4/9] fix(tables): keep a cleared failed cell cleared after a lost connection --- .../background/dispatch-cancel-guard.test.ts | 1 + .../background/drain-governed-subject.test.ts | 1 + .../enrichment-capability-subject.test.ts | 1 + .../workflow-group-governed-subject.test.ts | 1 + apps/sim/lib/table/cell-write.integration.ts | 26 +++++++++++++++++++ apps/sim/lib/table/cell-write.ts | 11 +++++--- 6 files changed, 37 insertions(+), 4 deletions(-) diff --git a/apps/sim/background/dispatch-cancel-guard.test.ts b/apps/sim/background/dispatch-cancel-guard.test.ts index a7ffae490ff..6143e469c62 100644 --- a/apps/sim/background/dispatch-cancel-guard.test.ts +++ b/apps/sim/background/dispatch-cancel-guard.test.ts @@ -45,6 +45,7 @@ vi.mock('@/lib/table/cell-write', () => ({ }), createWorkflowCellProgressWriter: mocks.createWorkflowCellProgressWriter, writeWorkflowGroupState: mocks.writeWorkflowGroupState, + writeWorkflowGroupStateDurably: mocks.writeWorkflowGroupState, markWorkflowGroupPickedUp: mocks.markWorkflowGroupPickedUp, })) vi.mock('@/lib/table/workflow-cell-result', () => ({ diff --git a/apps/sim/background/drain-governed-subject.test.ts b/apps/sim/background/drain-governed-subject.test.ts index 015ab11a686..a8a999ff9cc 100644 --- a/apps/sim/background/drain-governed-subject.test.ts +++ b/apps/sim/background/drain-governed-subject.test.ts @@ -42,6 +42,7 @@ vi.mock('@/lib/table/cell-write', () => ({ buildCancelledExecution: vi.fn(), createWorkflowCellProgressWriter: vi.fn(), writeWorkflowGroupState: hoisted.writeWorkflowGroupState, + writeWorkflowGroupStateDurably: hoisted.writeWorkflowGroupState, markWorkflowGroupPickedUp: hoisted.markWorkflowGroupPickedUp, })) vi.mock('@/lib/table/workflow-cell-result', () => ({ diff --git a/apps/sim/background/enrichment-capability-subject.test.ts b/apps/sim/background/enrichment-capability-subject.test.ts index aab5f966eef..9158471d955 100644 --- a/apps/sim/background/enrichment-capability-subject.test.ts +++ b/apps/sim/background/enrichment-capability-subject.test.ts @@ -49,6 +49,7 @@ vi.mock('@/lib/table/service', () => tableServiceMock) vi.mock('@/lib/table/rows/service', () => tableRowsServiceMock) vi.mock('@/lib/table/cell-write', () => ({ writeWorkflowGroupState: hoisted.writeWorkflowGroupState, + writeWorkflowGroupStateDurably: hoisted.writeWorkflowGroupState, markWorkflowGroupPickedUp: hoisted.markWorkflowGroupPickedUp, createWorkflowCellProgressWriter: hoisted.createWorkflowCellProgressWriter, buildCancelledExecution: hoisted.buildCancelledExecution, diff --git a/apps/sim/background/workflow-group-governed-subject.test.ts b/apps/sim/background/workflow-group-governed-subject.test.ts index 3e9af43acf7..3858d284fcf 100644 --- a/apps/sim/background/workflow-group-governed-subject.test.ts +++ b/apps/sim/background/workflow-group-governed-subject.test.ts @@ -57,6 +57,7 @@ vi.mock('@/lib/table/cell-write', () => ({ }), createWorkflowCellProgressWriter: hoisted.createWorkflowCellProgressWriter, writeWorkflowGroupState: hoisted.writeWorkflowGroupState, + writeWorkflowGroupStateDurably: hoisted.writeWorkflowGroupState, markWorkflowGroupPickedUp: hoisted.markWorkflowGroupPickedUp, })) vi.mock('@/lib/table/workflow-cell-result', () => ({ diff --git a/apps/sim/lib/table/cell-write.integration.ts b/apps/sim/lib/table/cell-write.integration.ts index b334a42c416..0775585994f 100644 --- a/apps/sim/lib/table/cell-write.integration.ts +++ b/apps/sim/lib/table/cell-write.integration.ts @@ -212,6 +212,32 @@ describe('a durable cell state write around a lost connection', { timeout: 90_00 expect((await readCell(rowId)).status).toBeNull() }) + it('does not bring back outputs of a failed run the user cleared after the commit landed', async () => { + const { ctx, rowId } = await createCell() + interception = { + ...LANDED_COMMIT, + beforeReset: async () => { + await control`DELETE FROM table_row_executions WHERE row_id = ${rowId}` + await control`UPDATE user_table_rows SET data = '{}' WHERE id = ${rowId}` + }, + } + + await expect( + writeWorkflowGroupStateDurably(ctx, { + dataPatch: { out: 'partial' }, + executionState: { + status: 'error', + executionId, + jobId: null, + workflowId: 'workflow-1', + error: 'A block failed', + }, + }) + ).resolves.toBe('skipped') + + expect(await readCell(rowId)).toMatchObject({ out: null, status: null }) + }) + it('writes an error for a cell that never had a state, not mistaking it for a cleared one', async () => { const { ctx, rowId } = await createCell({ pickedUp: false }) interception = { match: /commit/i, forward: false } diff --git a/apps/sim/lib/table/cell-write.ts b/apps/sim/lib/table/cell-write.ts index 64c838315da..4527abb34a1 100644 --- a/apps/sim/lib/table/cell-write.ts +++ b/apps/sim/lib/table/cell-write.ts @@ -208,9 +208,10 @@ async function readStoredOutputs( * A lost connection leaves the outcome unknown, so each repeat first checks what the last attempt * did. Re-applying a commit that landed could undo what the user did since, such as editing the * output or stopping a downstream group, so a cell whose state already matches only needs the event - * the lost attempt never emitted, carrying what is stored now. A `completed` cell with no state left - * was cleared by the user, since a run only completes after its pickup wrote one. Any other state - * can legitimately be the first the cell ever had, so it is written. + * the lost attempt never emitted, carrying what is stored now. A completed run or a write carrying + * outputs comes after a pickup wrote the cell's state, so if that state is gone the user cleared it. + * Any other write can be the first the cell ever had, so it is written; the most that can come back + * is an error badge with no outputs, which the user can clear or re-run. * * A `cancelled` stamp is never repeated. It carries no guard, so a repeat could kill a run the user * started after it. @@ -234,7 +235,9 @@ export async function writeWorkflowGroupStateDurably( }) return 'wrote' } - if (!current && status === 'completed') return 'skipped' + const followsPickup = + status === 'completed' || Object.keys(payload.dataPatch ?? {}).length > 0 + if (!current && followsPickup) return 'skipped' } attempted = true return writeWorkflowGroupState(ctx, payload) From 8cedba31e7c3c198a0a1bf2e80a8a20b8567d753 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 4 Nov 2026 21:47:50 -0800 Subject: [PATCH 5/9] fix(tables): finish every unclaimed cell on a failed row and keep committed results --- ...w-column-execution-connection-loss.test.ts | 39 +++++++++-- .../background/workflow-column-execution.ts | 70 +++++++++++-------- apps/sim/lib/table/cell-write.integration.ts | 47 ++++++++++++- apps/sim/lib/table/cell-write.ts | 66 ++++++++++++++++- 4 files changed, 180 insertions(+), 42 deletions(-) diff --git a/apps/sim/background/workflow-column-execution-connection-loss.test.ts b/apps/sim/background/workflow-column-execution-connection-loss.test.ts index 1ebb4c8d2c3..c6087e61005 100644 --- a/apps/sim/background/workflow-column-execution-connection-loss.test.ts +++ b/apps/sim/background/workflow-column-execution-connection-loss.test.ts @@ -39,6 +39,8 @@ const local = vi.hoisted(() => ({ readDispatch: vi.fn(), writeWorkflowGroupState: vi.fn(), markWorkflowGroupPickedUp: vi.fn(), + confirmWorkflowGroupStateCommitted: vi.fn(), + terminalizeUnclaimedRowMarkers: vi.fn(), })) const mocks = { @@ -64,6 +66,8 @@ vi.mock('@/lib/table/cell-write', () => ({ writeWorkflowGroupState: vi.fn(), writeWorkflowGroupStateDurably: local.writeWorkflowGroupState, markWorkflowGroupPickedUp: local.markWorkflowGroupPickedUp, + confirmWorkflowGroupStateCommitted: local.confirmWorkflowGroupStateCommitted, + terminalizeUnclaimedRowMarkers: local.terminalizeUnclaimedRowMarkers, createWorkflowCellProgressWriter: () => ({ onBlockStart: vi.fn(), onBlockComplete: vi.fn(), @@ -154,6 +158,8 @@ beforeEach(() => { mocks.readDispatch.mockResolvedValue({ id: 'dispatch-1', status: 'active' }) mocks.writeWorkflowGroupState.mockResolvedValue('wrote') mocks.markWorkflowGroupPickedUp.mockResolvedValue('wrote') + mocks.confirmWorkflowGroupStateCommitted.mockResolvedValue(false) + mocks.terminalizeUnclaimedRowMarkers.mockResolvedValue(undefined) mocks.executeWorkflow.mockResolvedValue({ status: 'completed', success: true, output: {} }) mocks.loadDeployedWorkflowState.mockResolvedValue({ blocks: { start: { id: 'start', type: 'start_trigger', subBlocks: {}, outputs: {} } }, @@ -218,6 +224,24 @@ describe('a table cell losing a database socket', () => { expect(writtenStatuses()).toEqual(['completed', 'error']) }) + it('does not mark a finished run failed when its terminal write committed despite the error', async () => { + mocks.writeWorkflowGroupState.mockRejectedValueOnce(resetBegin()) + mocks.confirmWorkflowGroupStateCommitted.mockResolvedValue(true) + + await runRowCascadeLoop(PAYLOAD) + + expect(writtenStatuses()).toEqual(['completed']) + }) + + it('only trusts a committed terminal write after a terminal write was attempted', async () => { + mocks.executeWorkflow.mockRejectedValue(resetRead()) + mocks.confirmWorkflowGroupStateCommitted.mockResolvedValue(true) + + await runRowCascadeLoop(PAYLOAD) + + expect(writtenStatuses()).toEqual(['error']) + }) + it('fails the run when neither the terminal write nor the error write completes', async () => { const lost = resetBegin() mocks.writeWorkflowGroupState.mockRejectedValue(lost) @@ -227,23 +251,24 @@ describe('a table cell losing a database socket', () => { expect(mocks.executeWorkflow).toHaveBeenCalledOnce() }) - it('terminalizes the marker when the read that starts a group never recovers, and still fails', async () => { + it('finishes the row markers when a read between groups never recovers, and still fails', async () => { const failure = resetRead() - mocks.getTableById.mockRejectedValue(failure) - dbChainMockFns.returning.mockResolvedValueOnce([{ rowId: 'row-1' }]) + mocks.getRowById.mockResolvedValueOnce(ROW).mockRejectedValue(failure) await expect(executeWorkflowGroupCellJob(PAYLOAD)).rejects.toBe(failure) - expect(dbChainMockFns.set).toHaveBeenCalledWith( - expect.objectContaining({ status: 'error', executionId: 'execution-1' }) + expect(mocks.executeWorkflow).toHaveBeenCalledOnce() + expect(mocks.terminalizeUnclaimedRowMarkers).toHaveBeenCalledWith( + 'table-1', + 'row-1', + expect.any(String) ) - expect(mocks.executeWorkflow).not.toHaveBeenCalled() }) it('reports the original failure when the marker cannot be terminalized either', async () => { const failure = resetRead() mocks.getTableById.mockRejectedValue(failure) - dbChainMockFns.returning.mockRejectedValue(resetBegin()) + mocks.terminalizeUnclaimedRowMarkers.mockRejectedValue(resetBegin()) await expect(executeWorkflowGroupCellJob(PAYLOAD)).rejects.toBe(failure) }) diff --git a/apps/sim/background/workflow-column-execution.ts b/apps/sim/background/workflow-column-execution.ts index 4b5d16fc800..2b7fbaa598e 100644 --- a/apps/sim/background/workflow-column-execution.ts +++ b/apps/sim/background/workflow-column-execution.ts @@ -34,6 +34,7 @@ import { preprocessExecution } from '@/lib/execution/preprocessing' import { retryTableAdmission } from '@/lib/table/admission-retry' import { withCascadeLock } from '@/lib/table/cascade-lock' import { fillMissingColumns, mapInputValues, namedRowMapper } from '@/lib/table/cell-format' +import type { WriteWorkflowGroupStatePayload } from '@/lib/table/cell-write' import { getColumnId } from '@/lib/table/column-keys' import { CELL_DATABASE_RETRY } from '@/lib/table/constants' import { isEmptyCellValue, isExecCancelled } from '@/lib/table/deps' @@ -201,13 +202,6 @@ export async function terminalizeAbortedQueuedCarrierMarker( timedOut: isTimeoutAbortReason(signal.reason), timeoutMs, }) - return terminalizeQueuedMarker(payload, executionState) -} - -async function terminalizeQueuedMarker( - payload: QueuedWorkflowGroupCellPayload, - executionState: RowExecutionMetadata -): Promise { const [updated] = await db .update(tableRowExecutions) .set({ @@ -332,7 +326,6 @@ export async function executeWorkflowGroupCellJob( ) { const carrierTimeoutController = createWorkflowGroupCarrierTimeoutController(payload, signal) const carrierSignal = carrierTimeoutController.signal - let currentPayload = payload try { const { tableId, rowId, workspaceId } = payload const { getTableById } = await import('@/lib/table/service') @@ -340,6 +333,7 @@ export async function executeWorkflowGroupCellJob( const { pickNextEligibleGroupForRow } = await import('@/lib/table/workflow-columns') const { readStampedCapabilitySubject } = await import('@/lib/table/rows/executions') + let currentPayload = payload while (true) { if (carrierSignal.aborted) { await terminalizeAbortedQueuedCarrierMarker(currentPayload, carrierSignal) @@ -396,25 +390,20 @@ export async function executeWorkflowGroupCellJob( } } catch (err) { /** - * A read that ran out of retries outside a group's own handling, such as the table lookup that - * starts a group, leaves this carrier's marker unclaimed with nothing to revisit it. The throw - * still fails the run, because the database stayed unreachable through a full retry budget. + * A read that ran out of retries between groups kills this carrier, the row's only owner. Other + * carriers bail on its lock expecting it to drain their markers, so any left unclaimed would + * stay Queued with nothing to revisit them. The throw still fails the run, because the database + * stayed unreachable through a full retry budget. */ - const { executionId, enrichmentId, workflowId } = currentPayload + const { tableId, rowId } = payload + const { terminalizeUnclaimedRowMarkers } = await import('@/lib/table/cell-write') await withIdempotentWriteRetry( - () => - terminalizeQueuedMarker(currentPayload, { - status: 'error', - executionId, - jobId: null, - workflowId: enrichmentId ?? workflowId, - error: CELL_STOPPED_MESSAGE, - runningBlockIds: [], - }), - { label: 'terminalizeQueuedMarker', ...CELL_DATABASE_RETRY } + () => terminalizeUnclaimedRowMarkers(tableId, rowId, CELL_STOPPED_MESSAGE), + { label: 'terminalizeUnclaimedRowMarkers', ...CELL_DATABASE_RETRY } ).catch((markerErr) => { - logger.error('Failed to terminalize the queued marker of a failed cell', { - executionId, + logger.error('Failed to terminalize the unclaimed markers of a failed row', { + tableId, + rowId, cause: describeError(markerErr), }) }) @@ -864,6 +853,10 @@ async function runWorkflowAndWriteTerminal( } let progressWriter: ReturnType | null = null + let attemptedTerminal: { + write: WriteWorkflowGroupStatePayload + status: 'completed' | 'error' | 'cancelled' + } | null = null try { const [workflowRecord] = await readForCell('selectWorkflow', () => @@ -1269,12 +1262,17 @@ async function runWorkflowAndWriteTerminal( runningBlockIds: [], blockErrors, } - await writeState( - terminalState, - pendingDataPatch, - eventOutputs, - progressWriter.getPendingSecretProvenance() - ) + const secretProvenance = progressWriter.getPendingSecretProvenance() + attemptedTerminal = { + status: terminalResult.status, + write: { + executionState: terminalState, + dataPatch: pendingDataPatch, + eventOutputs, + secretProvenance, + }, + } + await writeState(terminalState, pendingDataPatch, eventOutputs, secretProvenance) return terminalResult.status } catch (err) { logger.error( @@ -1287,6 +1285,18 @@ async function runWorkflowAndWriteTerminal( } ) await progressWriter?.finish() + /** + * A terminal write that lost its connection may still have committed, and an error written + * over it would mark a finished, billed run as failed. + */ + if (attemptedTerminal) { + const { write, status } = attemptedTerminal + const { confirmWorkflowGroupStateCommitted } = await import('@/lib/table/cell-write') + const committed = await readForCell('confirmWorkflowGroupStateCommitted', () => + confirmWorkflowGroupStateCommitted(cellCtx, write) + ).catch(() => false) + if (committed) return status + } try { await writeState({ status: 'error', diff --git a/apps/sim/lib/table/cell-write.integration.ts b/apps/sim/lib/table/cell-write.integration.ts index 0775585994f..68324c5e77c 100644 --- a/apps/sim/lib/table/cell-write.integration.ts +++ b/apps/sim/lib/table/cell-write.integration.ts @@ -13,7 +13,7 @@ import { tableWorkflowColumnsMock } from '@sim/testing/mocks/table-workflow-colu import { sleep } from '@sim/utils/helpers' import { generateId } from '@sim/utils/id' import postgres from 'postgres' -import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest' +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest' vi.mock('@/lib/table/billing', () => tableBillingMock) vi.mock('@/lib/table/events', () => tableEventsMock) @@ -71,7 +71,11 @@ const { db } = await import('@sim/db') const { userTableDefinitions, userTableRows } = await import('@sim/db/schema') const { writeExecutionsPatch } = await import('@/lib/table/rows/executions') const { getTableById } = await import('@/lib/table/service') -const { writeWorkflowGroupStateDurably } = await import('@/lib/table/cell-write') +const { + confirmWorkflowGroupStateCommitted, + terminalizeUnclaimedRowMarkers, + writeWorkflowGroupStateDurably, +} = await import('@/lib/table/cell-write') const LANDED_COMMIT = { match: /commit/i, forward: true } const workspaceId = generateId() @@ -141,6 +145,11 @@ beforeEach(() => { interception = null }) +/** An interception still armed means no reset happened, and the test proved nothing. */ +afterEach(() => { + expect(interception).toBeNull() +}) + afterAll(async () => { await control.begin(async (tx) => { await tx`DELETE FROM workspace WHERE id = ${workspaceId}` @@ -274,6 +283,40 @@ describe('a durable cell state write around a lost connection', { timeout: 90_00 expect((await readCell(rowId)).status).toBe('cancelled') }) + it('confirms a write that committed and tells the client, and only that write', async () => { + const { ctx, rowId } = await createCell() + const write = completed({ out: 'result' }) + + await expect(confirmWorkflowGroupStateCommitted(ctx, write)).resolves.toBe(false) + expect(tableEventsMockFns.mockAppendTableEvent).not.toHaveBeenCalled() + + await control`UPDATE table_row_executions SET status = 'completed' WHERE row_id = ${rowId}` + await expect(confirmWorkflowGroupStateCommitted(ctx, write)).resolves.toBe(true) + expect(tableEventsMockFns.mockAppendTableEvent).toHaveBeenCalledOnce() + }) + + it('fails only the markers on a row that no worker has claimed', async () => { + const { ctx, tableId, rowId } = await createCell() + const pending = { status: 'pending' as const, jobId: null, workflowId: 'w', error: null } + await writeExecutionsPatch(db, tableId, rowId, { + unclaimed: { ...pending, executionId: null }, + paused: { ...pending, executionId: 'paused-run', jobId: 'paused-paused-run' }, + stopped: { ...pending, status: 'cancelled', executionId: null }, + }) + + await terminalizeUnclaimedRowMarkers(tableId, rowId, 'Stopped') + + const rows = await control<{ group_id: string; status: string }[]>` + SELECT group_id, status FROM table_row_executions WHERE row_id = ${rowId}` + expect(Object.fromEntries(rows.map((row) => [row.group_id, row.status]))).toEqual({ + [ctx.groupId]: 'running', + unclaimed: 'error', + paused: 'pending', + stopped: 'cancelled', + }) + expect(tableEventsMockFns.mockAppendTableEvent).toHaveBeenCalledOnce() + }) + it('does not repeat a cancel stamp, which could kill a run the user just started', async () => { const { ctx, rowId } = await createCell() interception = { diff --git a/apps/sim/lib/table/cell-write.ts b/apps/sim/lib/table/cell-write.ts index 4527abb34a1..46f16aaf6aa 100644 --- a/apps/sim/lib/table/cell-write.ts +++ b/apps/sim/lib/table/cell-write.ts @@ -10,10 +10,10 @@ */ import { db } from '@sim/db' -import { userTableRows } from '@sim/db/schema' +import { tableRowExecutions, userTableRows } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { getValueAtPath } from '@sim/utils/object' -import { and, eq } from 'drizzle-orm' +import { and, eq, isNull } from 'drizzle-orm' import { withIdempotentWriteRetry } from '@/lib/db/read-retry' import { CELL_DATABASE_RETRY } from '@/lib/table/constants' import { appendTableEvent } from '@/lib/table/events' @@ -177,6 +177,66 @@ function emitCellStateEvent( }) } +function isCommittedState( + current: RowExecutionMetadata | undefined, + ctx: WriteWorkflowGroupContext, + payload: WriteWorkflowGroupStatePayload +): boolean { + return ( + current?.executionId === ctx.executionId && current.status === payload.executionState.status + ) +} + +/** + * Whether a write that reported failure actually committed. If it did, this sends the event the + * lost attempt never emitted, so a caller can treat the write as done. + */ +export async function confirmWorkflowGroupStateCommitted( + ctx: WriteWorkflowGroupContext, + payload: WriteWorkflowGroupStatePayload +): Promise { + const current = (await loadExecutionsForRow(db, ctx.rowId))[ctx.groupId] + if (!isCommittedState(current, ctx, payload)) return false + emitCellStateEvent(ctx, { ...payload, eventOutputs: await readStoredOutputs(ctx, payload) }) + return true +} + +/** + * Fails every marker on a row that no worker has claimed. `executionId` stays null, so a carrier + * still queued for one of them can claim it and run. Claimed, cancelled, and finished cells are + * untouched, and so is a paused one, which holds an execution id. + */ +export async function terminalizeUnclaimedRowMarkers( + tableId: string, + rowId: string, + error: string +): Promise { + const failed = await db + .update(tableRowExecutions) + .set({ status: 'error', jobId: null, error, runningBlockIds: [], updatedAt: new Date() }) + .where( + and( + eq(tableRowExecutions.tableId, tableId), + eq(tableRowExecutions.rowId, rowId), + eq(tableRowExecutions.status, 'pending'), + isNull(tableRowExecutions.executionId) + ) + ) + .returning({ groupId: tableRowExecutions.groupId }) + for (const { groupId } of failed) { + void appendTableEvent({ + kind: 'cell', + tableId, + rowId, + groupId, + status: 'error', + executionId: null, + jobId: null, + error, + }) + } +} + /** The stored values of a write's outputs, which may have changed since the write landed. */ async function readStoredOutputs( { rowId, tableId, workspaceId }: WriteWorkflowGroupContext, @@ -228,7 +288,7 @@ export async function writeWorkflowGroupStateDurably( async () => { if (attempted) { const current = (await loadExecutionsForRow(db, ctx.rowId))[ctx.groupId] - if (current?.executionId === ctx.executionId && current.status === status) { + if (isCommittedState(current, ctx, payload)) { emitCellStateEvent(ctx, { ...payload, eventOutputs: await readStoredOutputs(ctx, payload), From 45b009426c9a2e37fcf9d4517e5de36c6603bde9 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 4 Nov 2026 22:02:57 -0800 Subject: [PATCH 6/9] fix(tables): clean up a failed row under its lock and respect newer runs --- ...w-column-execution-connection-loss.test.ts | 14 ++++++++ .../background/workflow-column-execution.ts | 34 ++++++++++++------- .../sim/lib/db/connection-loss.integration.ts | 3 +- apps/sim/lib/table/cell-write.integration.ts | 26 +++++++++++++- apps/sim/lib/table/cell-write.ts | 10 +++--- 5 files changed, 69 insertions(+), 18 deletions(-) diff --git a/apps/sim/background/workflow-column-execution-connection-loss.test.ts b/apps/sim/background/workflow-column-execution-connection-loss.test.ts index c6087e61005..a7d5e2c1bd1 100644 --- a/apps/sim/background/workflow-column-execution-connection-loss.test.ts +++ b/apps/sim/background/workflow-column-execution-connection-loss.test.ts @@ -19,6 +19,7 @@ import { executionPreprocessingMock, executionPreprocessingMockFns, } from '@sim/testing/mocks/execution-preprocessing.mock' +import { redisConfigMockFns } from '@sim/testing/mocks/redis-config.mock' import { tableRowsServiceMock, tableRowsServiceMockFns, @@ -265,6 +266,19 @@ describe('a table cell losing a database socket', () => { ) }) + it('finishes the row markers before giving up the row lock', async () => { + const releasesBeforeCleanup: number[] = [] + mocks.getTableById.mockRejectedValue(resetRead()) + mocks.terminalizeUnclaimedRowMarkers.mockImplementation(async () => { + releasesBeforeCleanup.push(redisConfigMockFns.mockReleaseLock.mock.calls.length) + }) + + await expect(executeWorkflowGroupCellJob(PAYLOAD)).rejects.toThrow() + + expect(releasesBeforeCleanup).toEqual([0]) + expect(redisConfigMockFns.mockReleaseLock).toHaveBeenCalledOnce() + }) + it('reports the original failure when the marker cannot be terminalized either', async () => { const failure = resetRead() mocks.getTableById.mockRejectedValue(failure) diff --git a/apps/sim/background/workflow-column-execution.ts b/apps/sim/background/workflow-column-execution.ts index 2b7fbaa598e..0923eeb801b 100644 --- a/apps/sim/background/workflow-column-execution.ts +++ b/apps/sim/background/workflow-column-execution.ts @@ -388,13 +388,28 @@ export async function executeWorkflowGroupCellJob( ), } } + } finally { + carrierTimeoutController.cleanup() + } +} + +/** Re-fetches the table schema each iteration so groups added DURING the + * cascade become visible to the eligibility check. The resume worker must + * already hold the row's cascade lock before calling. + * + * A read that runs out of retries between groups kills this carrier, the row's + * only owner. Other carriers bail on its lock expecting it to drain their + * markers, so any left unclaimed would stay Queued with nothing to revisit + * them. It fails them while it still holds the lock, so it cannot touch work a + * later owner has taken on, and rethrows because the database stayed + * unreachable through a full retry budget. */ +export async function runRowCascadeLoop( + payload: QueuedWorkflowGroupCellPayload, + signal?: AbortSignal +): Promise<'blocked' | undefined> { + try { + return await driveRowCascade(payload, signal) } catch (err) { - /** - * A read that ran out of retries between groups kills this carrier, the row's only owner. Other - * carriers bail on its lock expecting it to drain their markers, so any left unclaimed would - * stay Queued with nothing to revisit them. The throw still fails the run, because the database - * stayed unreachable through a full retry budget. - */ const { tableId, rowId } = payload const { terminalizeUnclaimedRowMarkers } = await import('@/lib/table/cell-write') await withIdempotentWriteRetry( @@ -408,15 +423,10 @@ export async function executeWorkflowGroupCellJob( }) }) throw err - } finally { - carrierTimeoutController.cleanup() } } -/** Re-fetches the table schema each iteration so groups added DURING the - * cascade become visible to the eligibility check. The resume worker must - * already hold the row's cascade lock before calling. */ -export async function runRowCascadeLoop( +async function driveRowCascade( payload: QueuedWorkflowGroupCellPayload, signal?: AbortSignal ): Promise<'blocked' | undefined> { diff --git a/apps/sim/lib/db/connection-loss.integration.ts b/apps/sim/lib/db/connection-loss.integration.ts index 7da697ffc98..a28f4fade2b 100644 --- a/apps/sim/lib/db/connection-loss.integration.ts +++ b/apps/sim/lib/db/connection-loss.integration.ts @@ -68,7 +68,8 @@ afterAll(async () => { await new Promise((resolve) => proxy.close(resolve)) }) -describe('real connection resets', () => { +/** A reset can stall the replacement connection until `connect_timeout` before it rejects. */ +describe('real connection resets', { timeout: 90_000 }, () => { it('recovers a read reset while its statement is running', async () => { let attempts = 0 const rows = await withDatabaseReadRetry(async () => { diff --git a/apps/sim/lib/table/cell-write.integration.ts b/apps/sim/lib/table/cell-write.integration.ts index 68324c5e77c..b557247fc37 100644 --- a/apps/sim/lib/table/cell-write.integration.ts +++ b/apps/sim/lib/table/cell-write.integration.ts @@ -156,8 +156,9 @@ afterAll(async () => { await tx`DELETE FROM project WHERE id = ${workspaceId}` await tx`DELETE FROM "user" WHERE id = ${userId}` }) + await db.$client.end({ timeout: 1 }) await control.end({ timeout: 1 }) - proxy.close() + await new Promise((resolve) => proxy.close(resolve)) }) describe('a durable cell state write around a lost connection', { timeout: 90_000 }, () => { @@ -266,6 +267,29 @@ describe('a durable cell state write around a lost connection', { timeout: 90_00 expect((await readCell(rowId)).status).toBe('error') }) + it('does not overwrite the marker of a newer run requested during the retry backoff', async () => { + const { ctx, rowId } = await createCell() + interception = { + match: /commit/i, + forward: false, + afterReset: async () => { + await control` + UPDATE table_row_executions SET status = 'pending', execution_id = NULL + WHERE row_id = ${rowId}` + }, + } + + await expect(writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }))).resolves.toBe( + 'skipped' + ) + + expect(await readCell(rowId)).toMatchObject({ + out: 'before', + status: 'pending', + execution_id: null, + }) + }) + it('does not overwrite a Stop that lands during the retry backoff', async () => { const { ctx, rowId } = await createCell() interception = { diff --git a/apps/sim/lib/table/cell-write.ts b/apps/sim/lib/table/cell-write.ts index 46f16aaf6aa..88480b43c4c 100644 --- a/apps/sim/lib/table/cell-write.ts +++ b/apps/sim/lib/table/cell-write.ts @@ -269,9 +269,10 @@ async function readStoredOutputs( * did. Re-applying a commit that landed could undo what the user did since, such as editing the * output or stopping a downstream group, so a cell whose state already matches only needs the event * the lost attempt never emitted, carrying what is stored now. A completed run or a write carrying - * outputs comes after a pickup wrote the cell's state, so if that state is gone the user cleared it. - * Any other write can be the first the cell ever had, so it is written; the most that can come back - * is an error badge with no outputs, which the user can clear or re-run. + * outputs comes after a pickup wrote the cell's state, so if that state is gone the user cleared it, + * and an unclaimed marker in its place belongs to a newer run. Any other write can be the first the + * cell ever had, or can be settling its own marker, so it is written; the most that can come back is + * an error badge with no outputs, which the user can clear or re-run. * * A `cancelled` stamp is never repeated. It carries no guard, so a repeat could kill a run the user * started after it. @@ -297,7 +298,8 @@ export async function writeWorkflowGroupStateDurably( } const followsPickup = status === 'completed' || Object.keys(payload.dataPatch ?? {}).length > 0 - if (!current && followsPickup) return 'skipped' + const isNewerMarker = current?.status === 'pending' && current.executionId == null + if (followsPickup && (!current || isNewerMarker)) return 'skipped' } attempted = true return writeWorkflowGroupState(ctx, payload) From 17658414bf9c560da25c5318d62805732b8d1065 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 4 Nov 2026 22:14:42 -0800 Subject: [PATCH 7/9] fix(tables): keep driver text out of wrapped cell errors --- ...w-column-execution-connection-loss.test.ts | 27 +++++++++++++++++++ .../background/workflow-column-execution.ts | 21 +++++++++++---- 2 files changed, 43 insertions(+), 5 deletions(-) diff --git a/apps/sim/background/workflow-column-execution-connection-loss.test.ts b/apps/sim/background/workflow-column-execution-connection-loss.test.ts index a7d5e2c1bd1..0b47d15e97f 100644 --- a/apps/sim/background/workflow-column-execution-connection-loss.test.ts +++ b/apps/sim/background/workflow-column-execution-connection-loss.test.ts @@ -348,6 +348,33 @@ describe('a table cell losing a database socket', () => { expect(stored).not.toMatch(/ECONNRESET|Failed query|select|secret-value/) }) + it('does not store driver text a wrapper copied into its own message', async () => { + const driverError = Object.assign( + new Error('duplicate key value violates unique constraint "secret_index"'), + { name: 'PostgresError', code: '23505' } + ) + mocks.executeWorkflow.mockRejectedValue( + new Error(`Failed to save: ${driverError.message}`, { cause: driverError }) + ) + + await runRowCascadeLoop(PAYLOAD) + + expect(lastWrite()?.executionState.error).not.toMatch(/secret_index/) + }) + + it('keeps a wrapper message that repeats a cause from outside the database', async () => { + const message = 'Failed to upload: bucket not found' + mocks.executeWorkflow.mockRejectedValue( + new Error(message, { + cause: Object.assign(new Error('bucket not found'), { code: 'NoSuchBucket' }), + }) + ) + + await runRowCascadeLoop(PAYLOAD) + + expect(lastWrite()?.executionState.error).toBe(message) + }) + it('keeps a block error that names the block over a database cause', async () => { const message = 'Agent 1: An internal error occurred while executing the block.' mocks.executeWorkflow.mockRejectedValue(new Error(message, { cause: resetRead() })) diff --git a/apps/sim/background/workflow-column-execution.ts b/apps/sim/background/workflow-column-execution.ts index 0923eeb801b..eb3f7b9a604 100644 --- a/apps/sim/background/workflow-column-execution.ts +++ b/apps/sim/background/workflow-column-execution.ts @@ -1,7 +1,7 @@ import { db } from '@sim/db' import { tableRowExecutions, workflow as workflowTable } from '@sim/db/schema' import { createLogger, runWithRequestContext } from '@sim/logger' -import { classifyDatabaseFailure, describeError, toError } from '@sim/utils/errors' +import { classifyDatabaseFailure, describeError, findCause, toError } from '@sim/utils/errors' import { sleep } from '@sim/utils/helpers' import { generateId } from '@sim/utils/id' import { backoffWithJitter } from '@sim/utils/retry' @@ -79,13 +79,24 @@ function readForCell(label: string, read: () => Promise): Promise { const CELL_STOPPED_MESSAGE = 'This cell stopped before it finished. Run it again.' /** - * The error a cell shows. Text that would expose a query, or a bare driver error, says what happened - * instead, because anyone viewing the table reads it. Anything else keeps its own message, including - * a block's already-safe error over a database cause, which names the block that failed. + * The error a cell shows. Text that would expose a query, a bare driver error, or a wrapper repeating + * a driver's text says what happened instead, because anyone viewing the table reads it. Anything + * else keeps its own message, including a block's already-safe error over a database cause, which + * names the block that failed. */ function getCellFailureMessage(error: unknown): string { const failure = toError(error) - const exposesQuery = failure.message.includes('Failed query: ') + const exposesQuery = + failure.message.includes('Failed query: ') || + findCause( + failure, + (link): link is Error => + link instanceof Error && + link !== failure && + link.message.length > 0 && + failure.message.includes(link.message) && + (link.name === 'PostgresError' || classifyDatabaseFailure(link) !== 'permanent') + ) !== undefined if (!exposesQuery && !Object.hasOwn(failure, 'code')) return failure.message const failureClass = classifyDatabaseFailure(failure) if (failureClass === 'connection') return 'Lost the database connection. Run this cell again.' From 9353f5881b28683bec9e6663522328c689033e96 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 5 Nov 2026 01:05:53 -0800 Subject: [PATCH 8/9] fix(tables): keep a newer run alive when a failed cell retries its write Track whether a carrier's own pickup landed instead of inferring it from the write. After a pickup, an unclaimed marker belongs to a newer run, so the write is dropped and the cascade drains it. Before a pickup, an error over a marker leaves the execution id null so a newer carrier can still claim it. --- .../background/dispatch-cancel-guard.test.ts | 3 +- ...w-column-execution-connection-loss.test.ts | 54 ++++- .../background/workflow-column-execution.ts | 38 ++-- apps/sim/lib/table/cell-write.integration.ts | 194 +++++++++++++----- apps/sim/lib/table/cell-write.ts | 64 +++--- apps/sim/lib/table/constants.ts | 4 +- 6 files changed, 254 insertions(+), 103 deletions(-) diff --git a/apps/sim/background/dispatch-cancel-guard.test.ts b/apps/sim/background/dispatch-cancel-guard.test.ts index 6143e469c62..88eb6d4c21d 100644 --- a/apps/sim/background/dispatch-cancel-guard.test.ts +++ b/apps/sim/background/dispatch-cancel-guard.test.ts @@ -122,7 +122,8 @@ describe('the cell guard on its owning dispatch', () => { expect.anything(), expect.objectContaining({ executionState: expect.objectContaining({ status: 'cancelled' }), - }) + }), + { claimed: false } ) }) }) diff --git a/apps/sim/background/workflow-column-execution-connection-loss.test.ts b/apps/sim/background/workflow-column-execution-connection-loss.test.ts index 0b47d15e97f..8ac32ae6605 100644 --- a/apps/sim/background/workflow-column-execution-connection-loss.test.ts +++ b/apps/sim/background/workflow-column-execution-connection-loss.test.ts @@ -95,6 +95,7 @@ import { executeWorkflowGroupCellJob, runRowCascadeLoop, } from '@/background/workflow-column-execution' +import { buildBlockExecutionError } from '@/executor/utils/errors' const GROUP = { id: 'group-1', workflowId: 'workflow-1', outputs: [] } const NEXT_GROUP = { id: 'group-2', workflowId: 'workflow-2', outputs: [] } @@ -375,13 +376,58 @@ describe('a table cell losing a database socket', () => { expect(lastWrite()?.executionState.error).toBe(message) }) - it('keeps a block error that names the block over a database cause', async () => { - const message = 'Agent 1: An internal error occurred while executing the block.' - mocks.executeWorkflow.mockRejectedValue(new Error(message, { cause: resetRead() })) + /** The chain the block executor builds around a query failure, with the block's generic text. */ + function blockErrorOver(queryError: Error): Error { + const internal = 'An internal error occurred while executing the block. Please try again.' + return buildBlockExecutionError({ + block: { id: 'agent-1', metadata: { id: 'agent', name: 'Agent 1' } } as never, + error: new Error(internal, { cause: queryError }), + }) + } + + it('says a block lost the connection, since its generic text would hide that a re-run can work', async () => { + mocks.executeWorkflow.mockRejectedValue(blockErrorOver(resetRead())) await runRowCascadeLoop(PAYLOAD) - expect(lastWrite()?.executionState.error).toBe(message) + expect(lastWrite()?.executionState.error).toBe( + 'Lost the database connection. Run this cell again.' + ) + }) + + it('keeps the error of a block whose query the database rejected, which names the block', async () => { + const failure = blockErrorOver( + Object.assign(new Error('Failed query: update "x"\nparams: secret-value'), { + query: 'update "x"', + params: ['secret-value'], + cause: Object.assign(new Error('duplicate key'), { name: 'PostgresError', code: '23505' }), + }) + ) + mocks.executeWorkflow.mockRejectedValue(failure) + + await runRowCascadeLoop(PAYLOAD) + + expect(lastWrite()?.executionState.error).toBe(failure.message) + }) + + it('leaves a cell it never claimed open for a newer carrier when it fails before pickup', async () => { + mocks.readDispatch.mockRejectedValue(resetRead()) + + await runRowCascadeLoop(PAYLOAD) + + const [, write, options] = mocks.writeWorkflowGroupState.mock.calls.at(-1) ?? [] + expect(write.executionState).toMatchObject({ status: 'error', executionId: null }) + expect(options).toEqual({ claimed: false }) + }) + + it('reports a failure after pickup under its own claim', async () => { + mocks.executeWorkflow.mockRejectedValue(resetRead()) + + await runRowCascadeLoop(PAYLOAD) + + const [, write, options] = mocks.writeWorkflowGroupState.mock.calls.at(-1) ?? [] + expect(write.executionState).toMatchObject({ status: 'error', executionId: 'execution-1' }) + expect(options).toEqual({ claimed: true }) }) it('does not tell the user to re-run a query that will fail the same way', async () => { diff --git a/apps/sim/background/workflow-column-execution.ts b/apps/sim/background/workflow-column-execution.ts index eb3f7b9a604..21e43528566 100644 --- a/apps/sim/background/workflow-column-execution.ts +++ b/apps/sim/background/workflow-column-execution.ts @@ -80,9 +80,9 @@ const CELL_STOPPED_MESSAGE = 'This cell stopped before it finished. Run it again /** * The error a cell shows. Text that would expose a query, a bare driver error, or a wrapper repeating - * a driver's text says what happened instead, because anyone viewing the table reads it. Anything - * else keeps its own message, including a block's already-safe error over a database cause, which - * names the block that failed. + * a driver's text says what happened instead, because anyone viewing the table reads it. A block's + * error over a lost connection is one of those, since its generic text would hide that a re-run can + * work. Anything else keeps its own message, including a block's error over a rejected query. */ function getCellFailureMessage(error: unknown): string { const failure = toError(error) @@ -408,12 +408,11 @@ export async function executeWorkflowGroupCellJob( * cascade become visible to the eligibility check. The resume worker must * already hold the row's cascade lock before calling. * - * A read that runs out of retries between groups kills this carrier, the row's - * only owner. Other carriers bail on its lock expecting it to drain their - * markers, so any left unclaimed would stay Queued with nothing to revisit - * them. It fails them while it still holds the lock, so it cannot touch work a - * later owner has taken on, and rethrows because the database stayed - * unreachable through a full retry budget. */ + * Anything thrown here, usually a read out of retries between groups, kills this + * carrier, the row's only owner. Other carriers bail on its lock expecting it to + * drain their markers, so any left unclaimed would stay Queued with nothing to + * revisit them. It fails them while it still holds the lock, so it cannot touch + * work a later owner has taken on, and rethrows so the run fails visibly. */ export async function runRowCascadeLoop( payload: QueuedWorkflowGroupCellPayload, signal?: AbortSignal @@ -538,6 +537,7 @@ async function runWorkflowAndWriteTerminal( const timeoutController = createWorkflowGroupAttemptTimeoutController(payload, signal) const attemptSignal = timeoutController.signal const requestId = `wfgrp-${executionId}` + let claimed = false try { return await runWithRequestContext({ requestId }, async () => { @@ -563,12 +563,11 @@ async function runWorkflowAndWriteTerminal( eventOutputs?: RowData, secretProvenance?: TableRowSecretProvenanceWrite ) => - writeWorkflowGroupStateDurably(cellCtx, { - executionState, - dataPatch, - eventOutputs, - secretProvenance, - }) + writeWorkflowGroupStateDurably( + cellCtx, + { executionState, dataPatch, eventOutputs, secretProvenance }, + { claimed } + ) /** * Dispatch-level cancellation guard. @@ -691,6 +690,7 @@ async function runWorkflowAndWriteTerminal( jobId: null, }) if (pickedUp === 'skipped') return 'error' + claimed = true try { // Map table columns → enrichment input ids (skip this group's own outputs). @@ -1090,6 +1090,7 @@ async function runWorkflowAndWriteTerminal( // check and pickup. const pickedUp = await markWorkflowGroupPickedUp(cellCtx, { workflowId, jobId: null }) if (pickedUp === 'skipped') return 'error' + claimed = true // Output columns produced by THIS group are skipped on input — they're // populated by the run we're starting. Other group's outputs ARE @@ -1345,6 +1346,8 @@ async function runWorkflowAndWriteTerminal( * dispatcher has moved on by then and no sweep revisits the cell, so a `pending` left behind * renders as Queued for good. Reporting it as an error lets the cascade continue and the user * re-run it. The text is fixed because anything can be thrown here, and the log holds the rest. + * A carrier that never claimed the cell leaves the execution id null, so a newer run's carrier + * can still claim it. * * A rejected write means a Stop or a newer run owns the cell, so this carrier ends rather than * start the next group. If the write fails too the throw fails the run, the alert to act on. @@ -1363,12 +1366,13 @@ async function runWorkflowAndWriteTerminal( { executionState: { status: 'error', - executionId, + executionId: claimed ? executionId : null, jobId: null, workflowId: (group.type === 'enrichment' && group.enrichmentId) || workflowId, error: CELL_STOPPED_MESSAGE, }, - } + }, + { claimed } ) return written === 'wrote' ? 'error' : 'cancelled' } finally { diff --git a/apps/sim/lib/table/cell-write.integration.ts b/apps/sim/lib/table/cell-write.integration.ts index b557247fc37..a967155eff5 100644 --- a/apps/sim/lib/table/cell-write.integration.ts +++ b/apps/sim/lib/table/cell-write.integration.ts @@ -73,11 +73,15 @@ const { writeExecutionsPatch } = await import('@/lib/table/rows/executions') const { getTableById } = await import('@/lib/table/service') const { confirmWorkflowGroupStateCommitted, + markWorkflowGroupPickedUp, terminalizeUnclaimedRowMarkers, writeWorkflowGroupStateDurably, } = await import('@/lib/table/cell-write') const LANDED_COMMIT = { match: /commit/i, forward: true } +const LOST_COMMIT = { match: /commit/i, forward: false } +const CLAIMED = { claimed: true } +const UNCLAIMED = { claimed: false } const workspaceId = generateId() const userId = generateId() const groupId = 'group-1' @@ -109,6 +113,16 @@ async function createCell({ pickedUp = true } = {}) { return { ctx: { tableId, rowId, workspaceId, groupId, executionId, table }, tableId, rowId } } +const failed = (error = 'A block failed') => ({ + executionState: { + status: 'error' as const, + executionId, + jobId: null, + workflowId: 'workflow-1', + error, + }, +}) + const completed = (dataPatch: { out: string }) => ({ dataPatch, executionState: { @@ -164,11 +178,11 @@ afterAll(async () => { describe('a durable cell state write around a lost connection', { timeout: 90_000 }, () => { it('applies a write whose statement never reached the server', async () => { const { ctx, rowId } = await createCell() - interception = { match: /commit/i, forward: false } + interception = LOST_COMMIT - await expect(writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }))).resolves.toBe( - 'wrote' - ) + await expect( + writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }), CLAIMED) + ).resolves.toBe('wrote') expect(await readCell(rowId)).toMatchObject({ out: 'result', status: 'completed' }) expect(tableEventsMockFns.mockAppendTableEvent).toHaveBeenCalledOnce() @@ -178,9 +192,9 @@ describe('a durable cell state write around a lost connection', { timeout: 90_00 const { ctx, rowId } = await createCell() interception = LANDED_COMMIT - await expect(writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }))).resolves.toBe( - 'wrote' - ) + await expect( + writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }), CLAIMED) + ).resolves.toBe('wrote') expect(await readCell(rowId)).toMatchObject({ out: 'result', status: 'completed' }) expect(tableEventsMockFns.mockAppendTableEvent).toHaveBeenCalledOnce() @@ -198,7 +212,7 @@ describe('a durable cell state write around a lost connection', { timeout: 90_00 }, } - await writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' })) + await writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }), CLAIMED) expect((await readCell(rowId)).out).toBe('edited') expect(tableEventsMockFns.mockAppendTableEvent).toHaveBeenCalledWith( @@ -215,9 +229,9 @@ describe('a durable cell state write around a lost connection', { timeout: 90_00 }, } - await expect(writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }))).resolves.toBe( - 'skipped' - ) + await expect( + writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }), CLAIMED) + ).resolves.toBe('skipped') expect((await readCell(rowId)).status).toBeNull() }) @@ -233,16 +247,7 @@ describe('a durable cell state write around a lost connection', { timeout: 90_00 } await expect( - writeWorkflowGroupStateDurably(ctx, { - dataPatch: { out: 'partial' }, - executionState: { - status: 'error', - executionId, - jobId: null, - workflowId: 'workflow-1', - error: 'A block failed', - }, - }) + writeWorkflowGroupStateDurably(ctx, { ...failed(), dataPatch: { out: 'partial' } }, CLAIMED) ).resolves.toBe('skipped') expect(await readCell(rowId)).toMatchObject({ out: null, status: null }) @@ -250,28 +255,116 @@ describe('a durable cell state write around a lost connection', { timeout: 90_00 it('writes an error for a cell that never had a state, not mistaking it for a cleared one', async () => { const { ctx, rowId } = await createCell({ pickedUp: false }) - interception = { match: /commit/i, forward: false } + interception = LOST_COMMIT await expect( - writeWorkflowGroupStateDurably(ctx, { - executionState: { - status: 'error', - executionId, - jobId: null, - workflowId: 'workflow-1', - error: 'Workflow not found', - }, - }) + writeWorkflowGroupStateDurably(ctx, failed('Workflow not found'), UNCLAIMED) ).resolves.toBe('wrote') expect((await readCell(rowId)).status).toBe('error') }) + it('does not bring back a cell the user cleared before a claimed error write landed', async () => { + const { ctx, rowId } = await createCell() + interception = { + ...LOST_COMMIT, + afterReset: async () => { + await control`DELETE FROM table_row_executions WHERE row_id = ${rowId}` + }, + } + + await expect(writeWorkflowGroupStateDurably(ctx, failed(), CLAIMED)).resolves.toBe('skipped') + + expect((await readCell(rowId)).status).toBeNull() + }) + + it('leaves a newer run queued when a claimed error write retries, so the cascade can drain it', async () => { + const { ctx, rowId } = await createCell() + interception = { + ...LOST_COMMIT, + afterReset: async () => { + await control` + UPDATE table_row_executions SET status = 'pending', execution_id = NULL + WHERE row_id = ${rowId}` + }, + } + + await expect(writeWorkflowGroupStateDurably(ctx, failed(), CLAIMED)).resolves.toBe('skipped') + + expect(await readCell(rowId)).toMatchObject({ status: 'pending', execution_id: null }) + }) + + it('lets a newer carrier claim a cell whose unclaimed error write retried over a marker', async () => { + const { ctx, tableId, rowId } = await createCell({ pickedUp: false }) + await writeExecutionsPatch(db, tableId, rowId, { + [groupId]: { + status: 'pending', + executionId: null, + jobId: null, + workflowId: 'workflow-1', + error: null, + }, + }) + interception = LOST_COMMIT + + await expect(writeWorkflowGroupStateDurably(ctx, failed(), UNCLAIMED)).resolves.toBe('wrote') + + expect(await readCell(rowId)).toMatchObject({ status: 'error', execution_id: null }) + await expect( + markWorkflowGroupPickedUp( + { ...ctx, executionId: 'newer-execution' }, + { workflowId: 'workflow-1', jobId: null } + ) + ).resolves.toBe('wrote') + expect(await readCell(rowId)).toMatchObject({ + status: 'running', + execution_id: 'newer-execution', + }) + }) + + it('claims the cell when a pickup retries over a marker, which keeps its execution id', async () => { + const { ctx, tableId, rowId } = await createCell({ pickedUp: false }) + await writeExecutionsPatch(db, tableId, rowId, { + [groupId]: { + status: 'pending', + executionId: null, + jobId: null, + workflowId: 'workflow-1', + error: null, + }, + }) + interception = LOST_COMMIT + + await expect( + markWorkflowGroupPickedUp(ctx, { workflowId: 'workflow-1', jobId: null }) + ).resolves.toBe('wrote') + + expect(await readCell(rowId)).toMatchObject({ status: 'running', execution_id: executionId }) + }) + + it('does not take a finished newer run for its own lost write', async () => { + const { ctx, rowId } = await createCell() + interception = { + ...LOST_COMMIT, + afterReset: async () => { + await control` + UPDATE table_row_executions SET status = 'completed', execution_id = 'newer-execution' + WHERE row_id = ${rowId}` + }, + } + + await expect( + writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }), CLAIMED) + ).resolves.toBe('skipped') + + expect(tableEventsMockFns.mockAppendTableEvent).not.toHaveBeenCalled() + expect(await readCell(rowId)).toMatchObject({ out: 'before', execution_id: 'newer-execution' }) + }) + it('does not overwrite the marker of a newer run requested during the retry backoff', async () => { const { ctx, rowId } = await createCell() interception = { - match: /commit/i, - forward: false, + ...LOST_COMMIT, afterReset: async () => { await control` UPDATE table_row_executions SET status = 'pending', execution_id = NULL @@ -279,9 +372,9 @@ describe('a durable cell state write around a lost connection', { timeout: 90_00 }, } - await expect(writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }))).resolves.toBe( - 'skipped' - ) + await expect( + writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }), CLAIMED) + ).resolves.toBe('skipped') expect(await readCell(rowId)).toMatchObject({ out: 'before', @@ -293,16 +386,15 @@ describe('a durable cell state write around a lost connection', { timeout: 90_00 it('does not overwrite a Stop that lands during the retry backoff', async () => { const { ctx, rowId } = await createCell() interception = { - match: /commit/i, - forward: false, + ...LOST_COMMIT, afterReset: async () => { await control`UPDATE table_row_executions SET status = 'cancelled' WHERE row_id = ${rowId}` }, } - await expect(writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }))).resolves.toBe( - 'skipped' - ) + await expect( + writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }), CLAIMED) + ).resolves.toBe('skipped') expect((await readCell(rowId)).status).toBe('cancelled') }) @@ -353,15 +445,19 @@ describe('a durable cell state write around a lost connection', { timeout: 90_00 } await expect( - writeWorkflowGroupStateDurably(ctx, { - executionState: { - status: 'cancelled', - executionId, - jobId: null, - workflowId: 'workflow-1', - error: 'Cancelled', + writeWorkflowGroupStateDurably( + ctx, + { + executionState: { + status: 'cancelled', + executionId, + jobId: null, + workflowId: 'workflow-1', + error: 'Cancelled', + }, }, - }) + CLAIMED + ) ).rejects.toThrow() expect(await readCell(rowId)).toMatchObject({ status: 'pending', execution_id: null }) diff --git a/apps/sim/lib/table/cell-write.ts b/apps/sim/lib/table/cell-write.ts index 88480b43c4c..5084bfd1580 100644 --- a/apps/sim/lib/table/cell-write.ts +++ b/apps/sim/lib/table/cell-write.ts @@ -179,11 +179,10 @@ function emitCellStateEvent( function isCommittedState( current: RowExecutionMetadata | undefined, - ctx: WriteWorkflowGroupContext, - payload: WriteWorkflowGroupStatePayload + { executionState }: WriteWorkflowGroupStatePayload ): boolean { return ( - current?.executionId === ctx.executionId && current.status === payload.executionState.status + current?.executionId === executionState.executionId && current.status === executionState.status ) } @@ -196,7 +195,7 @@ export async function confirmWorkflowGroupStateCommitted( payload: WriteWorkflowGroupStatePayload ): Promise { const current = (await loadExecutionsForRow(db, ctx.rowId))[ctx.groupId] - if (!isCommittedState(current, ctx, payload)) return false + if (!isCommittedState(current, payload)) return false emitCellStateEvent(ctx, { ...payload, eventOutputs: await readStoredOutputs(ctx, payload) }) return true } @@ -268,41 +267,42 @@ async function readStoredOutputs( * A lost connection leaves the outcome unknown, so each repeat first checks what the last attempt * did. Re-applying a commit that landed could undo what the user did since, such as editing the * output or stopping a downstream group, so a cell whose state already matches only needs the event - * the lost attempt never emitted, carrying what is stored now. A completed run or a write carrying - * outputs comes after a pickup wrote the cell's state, so if that state is gone the user cleared it, - * and an unclaimed marker in its place belongs to a newer run. Any other write can be the first the - * cell ever had, or can be settling its own marker, so it is written; the most that can come back is - * an error badge with no outputs, which the user can clear or re-run. + * the lost attempt never emitted, carrying what is stored now. + * + * Otherwise `claimed` says what a changed cell means. After this carrier's pickup landed, a missing + * state means the user cleared the cell and an unclaimed marker belongs to a newer run, so the write + * is dropped and the cascade drains that marker. Before a pickup, either can be the cell's own, so + * the write goes ahead, except that an error over a marker leaves the execution id null: a newer + * run's carrier can still claim the cell. * * A `cancelled` stamp is never repeated. It carries no guard, so a repeat could kill a run the user * started after it. */ export async function writeWorkflowGroupStateDurably( ctx: WriteWorkflowGroupContext, - payload: WriteWorkflowGroupStatePayload + payload: WriteWorkflowGroupStatePayload, + { claimed }: { claimed: boolean } ): Promise<'wrote' | 'skipped'> { - const { status } = payload.executionState - if (status === 'cancelled') return writeWorkflowGroupState(ctx, payload) + if (payload.executionState.status === 'cancelled') return writeWorkflowGroupState(ctx, payload) + let write = payload let attempted = false return withIdempotentWriteRetry( async () => { if (attempted) { const current = (await loadExecutionsForRow(db, ctx.rowId))[ctx.groupId] - if (isCommittedState(current, ctx, payload)) { - emitCellStateEvent(ctx, { - ...payload, - eventOutputs: await readStoredOutputs(ctx, payload), - }) + if (isCommittedState(current, write)) { + emitCellStateEvent(ctx, { ...write, eventOutputs: await readStoredOutputs(ctx, write) }) return 'wrote' } - const followsPickup = - status === 'completed' || Object.keys(payload.dataPatch ?? {}).length > 0 - const isNewerMarker = current?.status === 'pending' && current.executionId == null - if (followsPickup && (!current || isNewerMarker)) return 'skipped' + const isUnclaimedMarker = current?.status === 'pending' && current.executionId == null + if (claimed && (!current || isUnclaimedMarker)) return 'skipped' + if (isUnclaimedMarker && write.executionState.status === 'error') { + write = { ...write, executionState: { ...write.executionState, executionId: null } } + } } attempted = true - return writeWorkflowGroupState(ctx, payload) + return writeWorkflowGroupState(ctx, write) }, { label: 'writeWorkflowGroupState', ...CELL_DATABASE_RETRY } ) @@ -506,15 +506,19 @@ export async function markWorkflowGroupPickedUp( ctx: WriteWorkflowGroupContext, prev: Pick ): Promise<'wrote' | 'skipped'> { - return writeWorkflowGroupStateDurably(ctx, { - executionState: { - status: 'running', - executionId: ctx.executionId, - jobId: prev.jobId, - workflowId: prev.workflowId, - error: null, + return writeWorkflowGroupStateDurably( + ctx, + { + executionState: { + status: 'running', + executionId: ctx.executionId, + jobId: prev.jobId, + workflowId: prev.workflowId, + error: null, + }, }, - }) + { claimed: false } + ) } /** Builds the canonical `cancelled` execution state used by every cancel path. diff --git a/apps/sim/lib/table/constants.ts b/apps/sim/lib/table/constants.ts index 66bacfefa71..ccc69255d71 100644 --- a/apps/sim/lib/table/constants.ts +++ b/apps/sim/lib/table/constants.ts @@ -523,8 +523,8 @@ export function generateUniqueTableName(existingNames: string[]): string { * How a cell rides out lost database connections. A reset takes individual sockets and a fresh one * works, but resets arrive in bursts of about a second, so backoff totals about four seconds to * outlast one. A reset can also stall the replacement connection until `connect_timeout`, so - * `maxElapsedMs` admits one more attempt after such a stall and no more, which bounds how long a - * failing cell holds its concurrency slot and cascade lock. + * `maxElapsedMs` admits one more attempt after such a stall and no more, which bounds how long each + * failing operation holds the cell's concurrency slot and cascade lock. */ export const CELL_DATABASE_RETRY = { maxAttempts: 5, From 283f404d455f595da3ee819cdc06d455ccc6904f Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 5 Nov 2026 01:23:42 -0800 Subject: [PATCH 9/9] fix(tables): refresh open tables when a marker cleanup loses its reply A repeat after a lost reply finds nothing left to update, so it cannot name the cells the lost attempt failed. The cleanup now owns its retry and signals a refetch. Correct the retry budget comment: maxElapsedMs only decides whether another attempt starts. --- .../background/workflow-column-execution.ts | 21 ++++++------ apps/sim/lib/table/cell-write.integration.ts | 27 +++++++++++++++ apps/sim/lib/table/cell-write.ts | 34 ++++++++++++------- apps/sim/lib/table/constants.ts | 5 +-- 4 files changed, 61 insertions(+), 26 deletions(-) diff --git a/apps/sim/background/workflow-column-execution.ts b/apps/sim/background/workflow-column-execution.ts index 21e43528566..d3c346423d7 100644 --- a/apps/sim/background/workflow-column-execution.ts +++ b/apps/sim/background/workflow-column-execution.ts @@ -25,7 +25,7 @@ import { } from '@/lib/core/execution-limits' import { withResourceOutboundScope } from '@/lib/core/network/resource-scope.server' import { RateLimiter } from '@/lib/core/rate-limiter/rate-limiter' -import { withDatabaseReadRetry, withIdempotentWriteRetry } from '@/lib/db/read-retry' +import { withDatabaseReadRetry } from '@/lib/db/read-retry' import { registerManualExecutionAborter, unregisterManualExecutionAborter, @@ -422,16 +422,15 @@ export async function runRowCascadeLoop( } catch (err) { const { tableId, rowId } = payload const { terminalizeUnclaimedRowMarkers } = await import('@/lib/table/cell-write') - await withIdempotentWriteRetry( - () => terminalizeUnclaimedRowMarkers(tableId, rowId, CELL_STOPPED_MESSAGE), - { label: 'terminalizeUnclaimedRowMarkers', ...CELL_DATABASE_RETRY } - ).catch((markerErr) => { - logger.error('Failed to terminalize the unclaimed markers of a failed row', { - tableId, - rowId, - cause: describeError(markerErr), - }) - }) + await terminalizeUnclaimedRowMarkers(tableId, rowId, CELL_STOPPED_MESSAGE).catch( + (markerErr) => { + logger.error('Failed to terminalize the unclaimed markers of a failed row', { + tableId, + rowId, + cause: describeError(markerErr), + }) + } + ) throw err } } diff --git a/apps/sim/lib/table/cell-write.integration.ts b/apps/sim/lib/table/cell-write.integration.ts index a967155eff5..2c36d679007 100644 --- a/apps/sim/lib/table/cell-write.integration.ts +++ b/apps/sim/lib/table/cell-write.integration.ts @@ -433,6 +433,33 @@ describe('a durable cell state write around a lost connection', { timeout: 90_00 expect(tableEventsMockFns.mockAppendTableEvent).toHaveBeenCalledOnce() }) + it('tells open tables to refetch when the reply to failing the markers is lost', async () => { + const { tableId, rowId } = await createCell({ pickedUp: false }) + await writeExecutionsPatch(db, tableId, rowId, { + [groupId]: { + status: 'pending', + executionId: null, + jobId: null, + workflowId: 'workflow-1', + error: null, + }, + }) + /** An unprepared statement waits on a reply before it executes, so the commit is applied here. */ + interception = { + match: /update "table_row_executions"/i, + forward: false, + beforeReset: async () => { + await control`UPDATE table_row_executions SET status = 'error' WHERE row_id = ${rowId}` + }, + } + + await terminalizeUnclaimedRowMarkers(tableId, rowId, 'Stopped') + + expect((await readCell(rowId)).status).toBe('error') + expect(tableEventsMockFns.mockAppendTableEvent).not.toHaveBeenCalled() + expect(tableEventsMockFns.mockSignalTableRowsChanged).toHaveBeenCalledWith(tableId) + }) + it('does not repeat a cancel stamp, which could kill a run the user just started', async () => { const { ctx, rowId } = await createCell() interception = { diff --git a/apps/sim/lib/table/cell-write.ts b/apps/sim/lib/table/cell-write.ts index 5084bfd1580..fe7618c547a 100644 --- a/apps/sim/lib/table/cell-write.ts +++ b/apps/sim/lib/table/cell-write.ts @@ -16,7 +16,7 @@ import { getValueAtPath } from '@sim/utils/object' import { and, eq, isNull } from 'drizzle-orm' import { withIdempotentWriteRetry } from '@/lib/db/read-retry' import { CELL_DATABASE_RETRY } from '@/lib/table/constants' -import { appendTableEvent } from '@/lib/table/events' +import { appendTableEvent, signalTableRowsChanged } from '@/lib/table/events' import { TableRowNotFoundError } from '@/lib/table/rows/errors' import { loadExecutionsForRow, writeExecutionsPatch } from '@/lib/table/rows/executions' import { @@ -204,24 +204,32 @@ export async function confirmWorkflowGroupStateCommitted( * Fails every marker on a row that no worker has claimed. `executionId` stays null, so a carrier * still queued for one of them can claim it and run. Claimed, cancelled, and finished cells are * untouched, and so is a paused one, which holds an execution id. + * + * Open tables are told to refetch too: a repeat after a lost reply finds nothing left to update, so + * it cannot name the cells the lost attempt failed. */ export async function terminalizeUnclaimedRowMarkers( tableId: string, rowId: string, error: string ): Promise { - const failed = await db - .update(tableRowExecutions) - .set({ status: 'error', jobId: null, error, runningBlockIds: [], updatedAt: new Date() }) - .where( - and( - eq(tableRowExecutions.tableId, tableId), - eq(tableRowExecutions.rowId, rowId), - eq(tableRowExecutions.status, 'pending'), - isNull(tableRowExecutions.executionId) - ) - ) - .returning({ groupId: tableRowExecutions.groupId }) + const failed = await withIdempotentWriteRetry( + () => + db + .update(tableRowExecutions) + .set({ status: 'error', jobId: null, error, runningBlockIds: [], updatedAt: new Date() }) + .where( + and( + eq(tableRowExecutions.tableId, tableId), + eq(tableRowExecutions.rowId, rowId), + eq(tableRowExecutions.status, 'pending'), + isNull(tableRowExecutions.executionId) + ) + ) + .returning({ groupId: tableRowExecutions.groupId }), + { label: 'terminalizeUnclaimedRowMarkers', ...CELL_DATABASE_RETRY } + ) + signalTableRowsChanged(tableId) for (const { groupId } of failed) { void appendTableEvent({ kind: 'cell', diff --git a/apps/sim/lib/table/constants.ts b/apps/sim/lib/table/constants.ts index ccc69255d71..cd13ea40de3 100644 --- a/apps/sim/lib/table/constants.ts +++ b/apps/sim/lib/table/constants.ts @@ -523,8 +523,9 @@ export function generateUniqueTableName(existingNames: string[]): string { * How a cell rides out lost database connections. A reset takes individual sockets and a fresh one * works, but resets arrive in bursts of about a second, so backoff totals about four seconds to * outlast one. A reset can also stall the replacement connection until `connect_timeout`, so - * `maxElapsedMs` admits one more attempt after such a stall and no more, which bounds how long each - * failing operation holds the cell's concurrency slot and cascade lock. + * `maxElapsedMs` leaves room to try again after such a stall. It only decides whether another attempt + * starts, so one operation can hold the cell's concurrency slot and cascade lock for up to + * `maxElapsedMs` plus a final stall, about 70 seconds. */ export const CELL_DATABASE_RETRY = { maxAttempts: 5,