Skip to content

Commit 61a5169

Browse files
authored
fix(tables): write the cell state when a resumed run throws (#8401)
* fix(tables): write the cell state when a resumed run throws runResumeAndCellTerminal only wrote the cell terminal after a resume returned, so a resume that threw left the cell on its last partial running state, where it could not even be cancelled. The resume manager now records, on the error it rethrows, when the attempt kept its pause resumable (admission refused, run buffer unavailable). The resume job mirrors that onto the cell: back to paused when the pause was kept, failed otherwise. A failed cell write is logged without masking the resume error, which is still rethrown. * fix(tables): mirror what a failed resume did instead of inferring it from the error An admission refusal can also mean the execution already finished, so treating every refusal as a kept pause wrote paused over a completed cell. markResumeAttemptFailed and markResumeFailed now return what their transaction actually did (pause still resumable / execution failed), the manager records that outcome on the rethrown error, and the resume job writes paused, error, or nothing. * fix(tables): type the terminal execution log statuses against the persisted vocabulary * fix(tables): report a failed resume's outcome through a hook instead of the error The outcome rode on the thrown error through a module WeakMap, so it was lost when draining queued resumes threw after the settle. startResumeExecution now takes onAttemptFailed, called right after the settle transaction; a hook failure is logged and never replaces the attempt's error.
1 parent bb4d256 commit 61a5169

4 files changed

Lines changed: 295 additions & 13 deletions

File tree

‎apps/sim/background/resume-execution.ts‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import { createLogger } from '@sim/logger'
2+
import { getErrorMessage } from '@sim/utils/errors'
23
import { generateId } from '@sim/utils/id'
34
import { task, timeout } from '@trigger.dev/sdk'
45
import {
@@ -21,6 +22,7 @@ import { classifyWorkflowCellTerminalResult } from '@/lib/table/workflow-cell-re
2122
import type { CellResumeContext } from '@/lib/table/workflow-columns'
2223
import {
2324
createResumeAttemptTimeoutController,
25+
type FailedResumeOutcome,
2426
PauseResumeManager,
2527
} from '@/lib/workflows/executor/human-in-the-loop-manager'
2628
import { RESUME_EXECUTION_CONCURRENCY_LIMIT } from '@/background/concurrency-limits'
@@ -359,6 +361,25 @@ async function buildResumeCellWriters(
359361
return { cellOnBlockComplete, writeCellTerminal }
360362
}
361363

364+
/**
365+
* A resume that throws never reaches the terminal write in
366+
* {@link runResumeAndCellTerminal}, which would leave the cell showing its last
367+
* partial `running` state. Mirror what the failed attempt did to the execution:
368+
* a pause that stayed resumable goes back to paused, and a failed execution
369+
* fails the cell.
370+
*/
371+
async function writeFailedResumeCellTerminal(
372+
writers: CellWriters,
373+
outcome: FailedResumeOutcome,
374+
error: unknown
375+
): Promise<void> {
376+
if (outcome === 'pause_retained') {
377+
await writers.writeCellTerminal('paused', null)
378+
} else {
379+
await writers.writeCellTerminal('error', getErrorMessage(error, 'Resume execution failed'))
380+
}
381+
}
382+
362383
async function runResumeAndCellTerminal(
363384
payload: ResumeExecutionPayload,
364385
pausedExecution: Awaited<ReturnType<typeof PauseResumeManager.getPausedExecutionById>>,
@@ -375,6 +396,7 @@ async function runResumeAndCellTerminal(
375396
resumeInput: payload.resumeInput,
376397
userId: payload.userId,
377398
onBlockComplete: writers.cellOnBlockComplete,
399+
onAttemptFailed: (outcome, error) => writeFailedResumeCellTerminal(writers, outcome, error),
378400
abortSignal: signal,
379401
})
380402

‎apps/sim/background/resume-governed-subject.test.ts‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ vi.mock('@/executor/execution/snapshot', () => ({
5151
ExecutionSnapshot: { fromJSON: hoisted.snapshotFromJson },
5252
}))
5353

54+
import type { FailedResumeOutcome } from '@/lib/workflows/executor/human-in-the-loop-manager'
5455
import { executeResumeJob, type ResumeExecutionPayload } from '@/background/resume-execution'
5556

5657
const mocks = {
@@ -217,4 +218,52 @@ describe('resuming a paused table cell', () => {
217218
const [cascadePayload] = mocks.runRowCascadeLoop.mock.calls[0]
218219
expect(cascadePayload.capabilityGovernedUserId).toBe('requesting-member')
219220
}, 20_000)
221+
222+
describe('when the resume throws', () => {
223+
/** The execution state the last cell write persisted. */
224+
function lastCellExecutionState() {
225+
const [, payload] = mocks.writeWorkflowGroupState.mock.calls.at(-1) ?? []
226+
return payload?.executionState
227+
}
228+
229+
/** Fails the resume the way the manager does: settle, report the outcome, rethrow. */
230+
function failResume(outcome: FailedResumeOutcome, error: Error) {
231+
mocks.startResumeExecution.mockImplementationOnce(
232+
async ({
233+
onAttemptFailed,
234+
}: {
235+
onAttemptFailed?: (outcome: FailedResumeOutcome, error: unknown) => Promise<void>
236+
}) => {
237+
await onAttemptFailed?.(outcome, error)
238+
throw error
239+
}
240+
)
241+
}
242+
243+
it('marks the cell failed when the resume failed the execution', async () => {
244+
const runFailure = new Error('writeLedger: Unique constraint violation')
245+
failResume('execution_failed', runFailure)
246+
247+
await expect(executeResumeJob(PAYLOAD)).rejects.toBe(runFailure)
248+
249+
expect(lastCellExecutionState()).toMatchObject({
250+
status: 'error',
251+
executionId: 'parent-execution-1',
252+
error: 'writeLedger: Unique constraint violation',
253+
})
254+
}, 20_000)
255+
256+
it('puts the cell back to paused when the pause stayed resumable', async () => {
257+
const admissionRefusal = new Error('Execution can no longer be resumed')
258+
failResume('pause_retained', admissionRefusal)
259+
260+
await expect(executeResumeJob(PAYLOAD)).rejects.toBe(admissionRefusal)
261+
262+
expect(lastCellExecutionState()).toMatchObject({
263+
status: 'pending',
264+
executionId: 'parent-execution-1',
265+
jobId: 'paused-parent-execution-1',
266+
})
267+
}, 20_000)
268+
})
220269
})

‎apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts‎

Lines changed: 163 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ import {
1818
largeValueMetadataMock,
1919
largeValueMetadataMockFns,
2020
} from '@sim/testing/mocks/large-value-metadata.mock'
21-
import { beforeEach, describe, expect, it, vi } from 'vitest'
21+
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
2222
import { createTimeoutAbortController, getExecutionDeadlineAt } from '@/lib/core/execution-limits'
2323
import { abortManualExecution } from '@/lib/execution/manual-cancellation'
2424
import { terminalExecutionLogFields } from '@/lib/logs/execution/cancellation'
@@ -61,6 +61,7 @@ vi.mock('@/lib/execution/payloads/large-value-metadata', () => largeValueMetadat
6161

6262
import {
6363
createResumeAttemptTimeoutController,
64+
type FailedResumeOutcome,
6465
PauseResumeManager,
6566
requireResumeDeploymentVersion,
6667
updateResumeOutputInAggregationBuffers,
@@ -84,7 +85,7 @@ if (!humanInTheLoopLogger) {
8485
}
8586

8687
interface PauseResumeManagerInternals {
87-
markResumeFailed: (...args: unknown[]) => Promise<void>
88+
markResumeFailed: (...args: unknown[]) => Promise<boolean>
8889
runResumeExecution: (...args: unknown[]) => Promise<unknown>
8990
}
9091

@@ -210,6 +211,166 @@ describe('queued resume attempt deadlines', () => {
210211
})
211212
})
212213

214+
describe('what a failed resume did to its paused execution', () => {
215+
type StartResumeArgs = Parameters<typeof PauseResumeManager.startResumeExecution>[0]
216+
const pausedExecution = {
217+
id: 'paused-execution-1',
218+
workflowId: 'workflow-1',
219+
executionId: 'parent-execution-1',
220+
pausePoints: {
221+
'context-1': { contextId: 'context-1', blockId: 'hitl-1' },
222+
},
223+
executionSnapshot: createSnapshotSeed(),
224+
metadata: {},
225+
} as StartResumeArgs['pausedExecution']
226+
const resumeArgs: StartResumeArgs = {
227+
resumeEntryId: 'resume-entry-1',
228+
resumeExecutionId: 'resume-execution-1',
229+
pausedExecution,
230+
contextId: 'context-1',
231+
resumeInput: { approved: true },
232+
userId: 'user-1',
233+
}
234+
const attemptArgs = {
235+
resumeEntryId: 'resume-entry-1',
236+
pausedExecutionId: 'paused-execution-1',
237+
parentExecutionId: 'parent-execution-1',
238+
contextId: 'context-1',
239+
failureReason: 'Execution can no longer be resumed',
240+
preserveForRetry: true,
241+
}
242+
243+
beforeEach(() => {
244+
resetDbChainMock()
245+
})
246+
247+
it('reports the pause still resumable when a refused attempt leaves it paused', async () => {
248+
queueTableRows(workflowExecutionLogs, [{ status: 'paused' }])
249+
queueTableRows(pausedExecutions, [{ automaticResumeRetryCount: 0, status: 'paused' }])
250+
251+
await expect(PauseResumeManager.markResumeAttemptFailed(attemptArgs)).resolves.toBe(true)
252+
})
253+
254+
it.each(['completed', 'failed', 'cancelled'])(
255+
'reports the pause not resumable when the execution is already %s',
256+
async (logStatus) => {
257+
queueTableRows(workflowExecutionLogs, [{ status: logStatus }])
258+
queueTableRows(pausedExecutions, [{ automaticResumeRetryCount: 0, status: 'paused' }])
259+
260+
await expect(PauseResumeManager.markResumeAttemptFailed(attemptArgs)).resolves.toBe(false)
261+
}
262+
)
263+
264+
it.each([
265+
{ logStatus: 'running', pauseStatus: 'paused', executionFailed: true },
266+
{ logStatus: 'cancelled', pauseStatus: 'paused', executionFailed: false },
267+
{ logStatus: 'running', pauseStatus: 'cancelling', executionFailed: false },
268+
])(
269+
'reports the execution failed: $executionFailed for a $logStatus log and $pauseStatus pause',
270+
async ({ logStatus, pauseStatus, executionFailed }) => {
271+
queueTableRows(workflowExecutionLogs, [{ status: logStatus }])
272+
queueTableRows(pausedExecutions, [{ status: pauseStatus }])
273+
const managerInternals = PauseResumeManager as unknown as PauseResumeManagerInternals
274+
275+
await expect(managerInternals.markResumeFailed(attemptArgs)).resolves.toBe(executionFailed)
276+
}
277+
)
278+
279+
/** Resume args that collect every outcome the manager reports. */
280+
function argsReportingOutcomes(onAttemptFailed?: () => Promise<void>) {
281+
const outcomes: FailedResumeOutcome[] = []
282+
return {
283+
outcomes,
284+
args: {
285+
...resumeArgs,
286+
onAttemptFailed: async (outcome: FailedResumeOutcome) => {
287+
outcomes.push(outcome)
288+
await onAttemptFailed?.()
289+
},
290+
},
291+
}
292+
}
293+
294+
it.each([
295+
{ stillResumable: true, reported: ['pause_retained'] },
296+
{ stillResumable: false, reported: [] },
297+
])(
298+
'reports a refused attempt as $reported when the pause is resumable: $stillResumable',
299+
async ({ stillResumable, reported }) => {
300+
const markResumeAttemptFailedSpy = vi
301+
.spyOn(PauseResumeManager, 'markResumeAttemptFailed')
302+
.mockResolvedValueOnce(stillResumable)
303+
const { outcomes, args } = argsReportingOutcomes()
304+
305+
try {
306+
await expect(PauseResumeManager.startResumeExecution(args)).rejects.toMatchObject({
307+
name: 'ResumeAdmissionError',
308+
})
309+
expect(outcomes).toEqual(reported)
310+
} finally {
311+
markResumeAttemptFailedSpy.mockRestore()
312+
}
313+
}
314+
)
315+
316+
describe('when the resumed run fails', () => {
317+
const rawError = new Error('Block failed')
318+
const spies: { mockRestore: () => void }[] = []
319+
320+
function failRun(options: { executionFailed: boolean; queuedResumesError?: Error }) {
321+
const managerInternals = PauseResumeManager as unknown as PauseResumeManagerInternals
322+
spies.push(
323+
vi.spyOn(managerInternals, 'runResumeExecution').mockRejectedValueOnce(rawError),
324+
vi
325+
.spyOn(managerInternals, 'markResumeFailed')
326+
.mockResolvedValueOnce(options.executionFailed),
327+
options.queuedResumesError
328+
? vi
329+
.spyOn(PauseResumeManager, 'processQueuedResumes')
330+
.mockRejectedValueOnce(options.queuedResumesError)
331+
: vi.spyOn(PauseResumeManager, 'processQueuedResumes').mockResolvedValueOnce()
332+
)
333+
}
334+
335+
afterEach(() => {
336+
for (const spy of spies.splice(0)) spy.mockRestore()
337+
})
338+
339+
it.each([
340+
{ executionFailed: true, reported: ['execution_failed'] },
341+
{ executionFailed: false, reported: [] },
342+
])(
343+
'reports $reported when it failed the execution: $executionFailed',
344+
async ({ executionFailed, reported }) => {
345+
failRun({ executionFailed })
346+
const { outcomes, args } = argsReportingOutcomes()
347+
348+
await expect(PauseResumeManager.startResumeExecution(args)).rejects.toBe(rawError)
349+
expect(outcomes).toEqual(reported)
350+
}
351+
)
352+
353+
it('reports the outcome before draining queued resumes, which may throw', async () => {
354+
failRun({ executionFailed: true, queuedResumesError: new Error('queue drain failed') })
355+
const { outcomes, args } = argsReportingOutcomes()
356+
357+
await expect(PauseResumeManager.startResumeExecution(args)).rejects.toThrow(
358+
'queue drain failed'
359+
)
360+
expect(outcomes).toEqual(['execution_failed'])
361+
})
362+
363+
it('rethrows the run failure when the outcome handler fails', async () => {
364+
failRun({ executionFailed: true })
365+
const { args } = argsReportingOutcomes(async () => {
366+
throw new Error('Database unavailable')
367+
})
368+
369+
await expect(PauseResumeManager.startResumeExecution(args)).rejects.toBe(rawError)
370+
})
371+
})
372+
})
373+
213374
describe('resume failure diagnostic projection', () => {
214375
beforeEach(() => {
215376
resetDbChainMock()

0 commit comments

Comments
 (0)