diff --git a/apps/sim/background/dispatch-cancel-guard.test.ts b/apps/sim/background/dispatch-cancel-guard.test.ts index a7ffae490ff..88eb6d4c21d 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', () => ({ @@ -121,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/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-column-execution-connection-loss.test.ts b/apps/sim/background/workflow-column-execution-connection-loss.test.ts new file mode 100644 index 00000000000..8ac32ae6605 --- /dev/null +++ b/apps/sim/background/workflow-column-execution-connection-loss.test.ts @@ -0,0 +1,478 @@ +/** + * 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 { 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 { redisConfigMockFns } from '@sim/testing/mocks/redis-config.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 local = vi.hoisted(() => ({ + readDispatch: vi.fn(), + writeWorkflowGroupState: vi.fn(), + markWorkflowGroupPickedUp: vi.fn(), + confirmWorkflowGroupStateCommitted: vi.fn(), + terminalizeUnclaimedRowMarkers: vi.fn(), +})) + +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: local.readDispatch })) +vi.mock('@/lib/table/workflow-columns', () => tableWorkflowColumnsMock) +vi.mock('@/lib/table/cell-write', () => ({ + buildCancelledExecution: vi.fn(), + writeWorkflowGroupState: vi.fn(), + writeWorkflowGroupStateDurably: local.writeWorkflowGroupState, + markWorkflowGroupPickedUp: local.markWorkflowGroupPickedUp, + confirmWorkflowGroupStateCommitted: local.confirmWorkflowGroupStateCommitted, + terminalizeUnclaimedRowMarkers: local.terminalizeUnclaimedRowMarkers, + createWorkflowCellProgressWriter: () => ({ + onBlockStart: vi.fn(), + onBlockComplete: vi.fn(), + finish: vi.fn(async () => {}), + getEventOutputs: () => ({}), + getPendingDataPatch: () => ({ out: 'value' }), + getPendingSecretProvenance: () => ({}), + getBlockErrors: () => ({}), + }), +})) +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 })) + }, +})) +vi.mock('@/lib/table/workflow-cell-result', () => ({ + classifyWorkflowCellTerminalResult: () => ({ status: 'completed', error: null }), +})) + +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: [] } +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(() => { + resetDbChainMock() + 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.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: {} } }, + 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('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) + + await expect(runRowCascadeLoop(PAYLOAD)).rejects.toBe(lost) + + expect(mocks.executeWorkflow).toHaveBeenCalledOnce() + }) + + it('finishes the row markers when a read between groups never recovers, and still fails', async () => { + const failure = resetRead() + mocks.getRowById.mockResolvedValueOnce(ROW).mockRejectedValue(failure) + + await expect(executeWorkflowGroupCellJob(PAYLOAD)).rejects.toBe(failure) + + expect(mocks.executeWorkflow).toHaveBeenCalledOnce() + expect(mocks.terminalizeUnclaimedRowMarkers).toHaveBeenCalledWith( + 'table-1', + 'row-1', + expect.any(String) + ) + }) + + 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) + mocks.terminalizeUnclaimedRowMarkers.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') + + 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('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) + }) + + /** 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( + '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 () => { + 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('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')) + + 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..d3c346423d7 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, findCause, 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, @@ -33,7 +34,9 @@ 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' import { getMaxTableDispatchConcurrency } from '@/lib/table/dispatch-concurrency' import { appendTableEvent } from '@/lib/table/events' @@ -65,6 +68,43 @@ 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 }) +} + +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. 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) + 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.' + const isDatabaseFailure = + exposesQuery || failureClass !== 'permanent' || failure.name === 'PostgresError' + return isDatabaseFailure ? '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 +364,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 +394,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 { @@ -362,10 +406,38 @@ export async function executeWorkflowGroupCellJob( /** 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. */ + * already hold the row's cascade lock before calling. + * + * 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 +): Promise<'blocked' | undefined> { + try { + return await driveRowCascade(payload, signal) + } catch (err) { + const { tableId, rowId } = payload + const { terminalizeUnclaimedRowMarkers } = await import('@/lib/table/cell-write') + 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 + } +} + +async function driveRowCascade( + payload: QueuedWorkflowGroupCellPayload, + signal?: AbortSignal ): Promise<'blocked' | undefined> { const { tableId, rowId, workspaceId } = payload const { getTableById } = await import('@/lib/table/service') @@ -395,7 +467,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 +497,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 +510,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 @@ -462,6 +536,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 () => { @@ -472,6 +547,7 @@ async function runWorkflowAndWriteTerminal( buildCancelledExecution, createWorkflowCellProgressWriter, writeWorkflowGroupState, + writeWorkflowGroupStateDurably, markWorkflowGroupPickedUp, } = await import('@/lib/table/cell-write') const { classifyWorkflowCellTerminalResult } = await import( @@ -486,12 +562,11 @@ async function runWorkflowAndWriteTerminal( eventOutputs?: RowData, secretProvenance?: TableRowSecretProvenanceWrite ) => - writeWorkflowGroupState(cellCtx, { - executionState, - dataPatch, - eventOutputs, - secretProvenance, - }) + writeWorkflowGroupStateDurably( + cellCtx, + { executionState, dataPatch, eventOutputs, secretProvenance }, + { claimed } + ) /** * Dispatch-level cancellation guard. @@ -515,7 +590,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 +636,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 +650,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})` @@ -612,6 +689,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). @@ -624,7 +702,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,20 +866,22 @@ async function runWorkflowAndWriteTerminal( executionId, jobId: null, workflowId: statusId, - error: toError(err).message, + error: getCellFailureMessage(err), }) return 'error' } } let progressWriter: ReturnType | null = null + let attemptedTerminal: { + write: WriteWorkflowGroupStatePayload + status: 'completed' | 'error' | 'cancelled' + } | 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 +896,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 +906,7 @@ async function runWorkflowAndWriteTerminal( executionId, jobId: null, workflowId, - error: toError(err).message, + error: getCellFailureMessage(err), }) return 'error' } @@ -838,7 +922,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 +1077,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,11 +1087,9 @@ 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' + 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 @@ -1028,7 +1112,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 +1154,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 +1162,8 @@ async function runWorkflowAndWriteTerminal( runningBlockIds, blockErrors, }) => - writeState( - { + writeWorkflowGroupState(cellCtx, { + executionState: { status: 'running', executionId, jobId: null, @@ -1087,8 +1174,8 @@ async function runWorkflowAndWriteTerminal( }, dataPatch, eventOutputs, - secretProvenance - ), + secretProvenance, + }), onWriteError: (err) => { logger.warn( `Per-block partial write failed (table=${tableId} row=${rowId} group=${groupId})`, @@ -1196,32 +1283,48 @@ 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) { - 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), } ) 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', executionId, jobId: null, workflowId, - error: message, + error: getCellFailureMessage(err), runningBlockIds: [], blockErrors: progressWriter?.getBlockErrors() ?? {}, }) @@ -1231,10 +1334,46 @@ async function runWorkflowAndWriteTerminal( cause: describeError(writeErr), retryable: isRetryableInfrastructureError(writeErr), }) + throw writeErr } 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 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. + */ + 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: claimed ? executionId : null, + jobId: null, + workflowId: (group.type === 'enrichment' && group.enrichmentId) || workflowId, + error: CELL_STOPPED_MESSAGE, + }, + }, + { claimed } + ) + return written === 'wrote' ? 'error' : 'cancelled' } finally { timeoutController.cleanup() } 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/db/connection-loss.integration.ts b/apps/sim/lib/db/connection-loss.integration.ts new file mode 100644 index 00000000000..a28f4fade2b --- /dev/null +++ b/apps/sim/lib/db/connection-loss.integration.ts @@ -0,0 +1,140 @@ +/** + * 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 { sleep } from '@sim/utils/helpers' +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 = () => sleep(75) + +beforeAll(async () => { + const url = new URL(readTestDatabaseUrl()) + proxy = net.createServer((downstream) => { + const upstream = net.connect(Number(url.port || 5432), url.hostname.replace(/^\[|\]$/g, '')) + 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)) +}) + +/** 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 () => { + 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..2c36d679007 --- /dev/null +++ b/apps/sim/lib/table/cell-write.integration.ts @@ -0,0 +1,492 @@ +/** + * 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, afterEach, 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.replace(/^\[|\]$/g, '')) + 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 { + 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' +const executionId = 'execution-1' + +async function createCell({ pickedUp = true } = {}) { + 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, + } + 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 } +} + +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: { + 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(() => { + 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}` + 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 }) + await new Promise((resolve) => proxy.close(resolve)) +}) + +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 = LOST_COMMIT + + await expect( + writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }), CLAIMED) + ).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' }), CLAIMED) + ).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' }), CLAIMED) + + 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 () => { + 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' }), CLAIMED) + ).resolves.toBe('skipped') + + 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, { ...failed(), dataPatch: { out: 'partial' } }, CLAIMED) + ).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 = LOST_COMMIT + + await expect( + 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 = { + ...LOST_COMMIT, + 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' }), CLAIMED) + ).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 = { + ...LOST_COMMIT, + afterReset: async () => { + await control`UPDATE table_row_executions SET status = 'cancelled' WHERE row_id = ${rowId}` + }, + } + + await expect( + writeWorkflowGroupStateDurably(ctx, completed({ out: 'result' }), CLAIMED) + ).resolves.toBe('skipped') + + 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('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 = { + ...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', + }, + }, + CLAIMED + ) + ).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..fe7618c547a 100644 --- a/apps/sim/lib/table/cell-write.ts +++ b/apps/sim/lib/table/cell-write.ts @@ -10,13 +10,15 @@ */ 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 { appendTableEvent } from '@/lib/table/events' +import { and, eq, isNull } from 'drizzle-orm' +import { withIdempotentWriteRetry } from '@/lib/db/read-retry' +import { CELL_DATABASE_RETRY } from '@/lib/table/constants' +import { appendTableEvent, signalTableRowsChanged } 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,145 @@ export async function writeWorkflowGroupState( ...(runningBlockIds && runningBlockIds.length > 0 ? { runningBlockIds } : {}), ...(blockErrors && Object.keys(blockErrors).length > 0 ? { blockErrors } : {}), }) +} - return 'wrote' +function isCommittedState( + current: RowExecutionMetadata | undefined, + { executionState }: WriteWorkflowGroupStatePayload +): boolean { + return ( + current?.executionId === executionState.executionId && current.status === 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, 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. + * + * 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 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', + 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, + 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, so a cell whose state already matches only needs the event + * 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, + { claimed }: { claimed: boolean } +): Promise<'wrote' | 'skipped'> { + 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, write)) { + emitCellStateEvent(ctx, { ...write, eventOutputs: await readStoredOutputs(ctx, write) }) + return 'wrote' + } + 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, write) + }, + { label: 'writeWorkflowGroupState', ...CELL_DATABASE_RETRY } + ) } export interface WorkflowCellProgressWrite { @@ -366,15 +514,19 @@ export async function markWorkflowGroupPickedUp( ctx: WriteWorkflowGroupContext, prev: Pick ): Promise<'wrote' | 'skipped'> { - return writeWorkflowGroupState(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 5560877e0e5..cd13ea40de3 100644 --- a/apps/sim/lib/table/constants.ts +++ b/apps/sim/lib/table/constants.ts @@ -518,3 +518,17 @@ 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` 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, + backoff: { baseMs: 250, maxMs: 2_000 }, + maxElapsedMs: 40_000, +} as const