Skip to content

Commit df3a592

Browse files
icecrasher321icecrasher321
committed
fix(workflows): preserve terminal runs during cancellation
1 parent 086f1bf commit df3a592

4 files changed

Lines changed: 210 additions & 18 deletions

File tree

‎apps/sim/lib/execution/cancel-workflow-execution.test.ts‎

Lines changed: 39 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ const {
1414
mockBlockQueuedResumesForCancellation,
1515
mockClearPausedCancellationIntent,
1616
mockCompletePausedCancellation,
17+
mockFinalizePausedCancellationForTerminalRun,
1718
mockGetPausedCancellationStatus,
1819
mockGetActiveResumeCancellationTarget,
1920
mockRollbackActiveResumeCancellation,
@@ -35,6 +36,7 @@ const {
3536
mockBlockQueuedResumesForCancellation: vi.fn(),
3637
mockClearPausedCancellationIntent: vi.fn(),
3738
mockCompletePausedCancellation: vi.fn(),
39+
mockFinalizePausedCancellationForTerminalRun: vi.fn(),
3840
mockGetPausedCancellationStatus: vi.fn(),
3941
mockGetActiveResumeCancellationTarget: vi.fn(),
4042
mockRollbackActiveResumeCancellation: vi.fn(),
@@ -75,6 +77,8 @@ vi.mock('@/lib/workflows/executor/human-in-the-loop-manager', () => ({
7577
clearPausedCancellationIntent: (...args: unknown[]) =>
7678
mockClearPausedCancellationIntent(...args),
7779
completePausedCancellation: (...args: unknown[]) => mockCompletePausedCancellation(...args),
80+
finalizePausedCancellationForTerminalRun: (...args: unknown[]) =>
81+
mockFinalizePausedCancellationForTerminalRun(...args),
7882
getPausedCancellationStatus: (...args: unknown[]) => mockGetPausedCancellationStatus(...args),
7983
getActiveResumeCancellationTarget: (...args: unknown[]) =>
8084
mockGetActiveResumeCancellationTarget(...args),
@@ -184,6 +188,7 @@ describe('cancelWorkflowExecution', () => {
184188
mockBlockQueuedResumesForCancellation.mockReset().mockResolvedValue(false)
185189
mockClearPausedCancellationIntent.mockReset().mockResolvedValue(undefined)
186190
mockCompletePausedCancellation.mockReset().mockResolvedValue(false)
191+
mockFinalizePausedCancellationForTerminalRun.mockReset().mockResolvedValue(true)
187192
mockGetPausedCancellationStatus.mockReset().mockResolvedValue(null)
188193
mockGetActiveResumeCancellationTarget.mockReset().mockResolvedValue(null)
189194
mockRollbackActiveResumeCancellation.mockReset().mockResolvedValue(true)
@@ -1626,7 +1631,7 @@ describe('cancelWorkflowExecution', () => {
16261631
expect(mockReleaseExecutionSlot).not.toHaveBeenCalled()
16271632
})
16281633

1629-
it('rolls back paused cancellation intent when resume completion wins the log claim', async () => {
1634+
it('finalizes paused cancellation state when resume completion wins the log claim', async () => {
16301635
mockStagePausedCancellation.mockResolvedValue({ kind: 'idle' })
16311636
const returning = vi.fn().mockResolvedValue([])
16321637
const where = vi.fn(() => ({ returning }))
@@ -1645,12 +1650,14 @@ describe('cancelWorkflowExecution', () => {
16451650
})
16461651
expect(mockWriteTerminalEvent).not.toHaveBeenCalled()
16471652
expect(mockCompletePausedCancellation).not.toHaveBeenCalled()
1648-
expect(mockClearPausedCancellationIntent).toHaveBeenCalledWith('ex-1', 'wf-1')
1653+
expect(mockFinalizePausedCancellationForTerminalRun).toHaveBeenCalledWith('ex-1', 'wf-1')
16491654
})
16501655

1651-
it('retries paused cancellation cleanup before returning a terminal-race conflict', async () => {
1656+
it('retries paused cancellation finalization before returning a terminal-race conflict', async () => {
16521657
mockStagePausedCancellation.mockResolvedValue({ kind: 'idle' })
1653-
mockClearPausedCancellationIntent.mockRejectedValueOnce(new Error('database unavailable'))
1658+
mockFinalizePausedCancellationForTerminalRun.mockRejectedValueOnce(
1659+
new Error('database unavailable')
1660+
)
16541661
const returning = vi.fn().mockResolvedValue([])
16551662
const where = vi.fn(() => ({ returning }))
16561663
databaseMock.db.update.mockReturnValueOnce({ set: vi.fn(() => ({ where })) })
@@ -1666,10 +1673,37 @@ describe('cancelWorkflowExecution', () => {
16661673
await expect(response.json()).resolves.toEqual({
16671674
error: 'Execution cannot be cancelled while completed',
16681675
})
1669-
expect(mockClearPausedCancellationIntent).toHaveBeenCalledTimes(2)
1676+
expect(mockFinalizePausedCancellationForTerminalRun).toHaveBeenCalledTimes(2)
16701677
expect(mockWriteTerminalEvent).not.toHaveBeenCalled()
16711678
})
16721679

1680+
it('keeps the active-resume stop marker when a terminal parent wins the log claim', async () => {
1681+
mockStagePausedCancellation.mockResolvedValue({
1682+
kind: 'active_resume',
1683+
target: ACTIVE_RESUME_TARGET,
1684+
})
1685+
mockMarkExecutionCancelled.mockResolvedValue({ durablyRecorded: true, reason: 'recorded' })
1686+
const returning = vi.fn().mockResolvedValue([])
1687+
const where = vi.fn(() => ({ returning }))
1688+
databaseMock.db.update.mockReturnValueOnce({ set: vi.fn(() => ({ where })) })
1689+
dbChainMockFns.limit
1690+
.mockResolvedValueOnce([
1691+
{ executionDeadlineAt: null, status: 'running', workspaceId: 'workspace-1' },
1692+
])
1693+
.mockResolvedValueOnce([{ status: 'completed' }])
1694+
1695+
const response = await POST(makeRequest(), makeParams())
1696+
1697+
expect(response.status).toBe(409)
1698+
await expect(response.json()).resolves.toEqual({
1699+
error: 'Execution cannot be cancelled while completed',
1700+
})
1701+
expect(mockFinalizePausedCancellationForTerminalRun).toHaveBeenCalledWith('ex-1', 'wf-1')
1702+
expect(mockRollbackActiveResumeCancellation).not.toHaveBeenCalled()
1703+
expect(mockClearPausedCancellationIntent).not.toHaveBeenCalled()
1704+
expect(mockClearExecutionCancellation).not.toHaveBeenCalled()
1705+
})
1706+
16731707
it('treats a concurrent cancellation as an idempotent success', async () => {
16741708
mockMarkExecutionCancelled.mockResolvedValue({ durablyRecorded: true, reason: 'recorded' })
16751709
dbChainMockFns.limit

‎apps/sim/lib/execution/cancel-workflow-execution.ts‎

Lines changed: 52 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -314,6 +314,35 @@ async function clearPausedCancellationIntentWithRetry(
314314
return false
315315
}
316316

317+
async function finalizePausedCancellationForTerminalRunWithRetry(
318+
executionId: string,
319+
workflowId: string
320+
): Promise<boolean> {
321+
for (let attempt = 1; attempt <= PAUSED_CANCELLATION_DB_ATTEMPTS; attempt++) {
322+
try {
323+
const finalized = await PauseResumeManager.finalizePausedCancellationForTerminalRun(
324+
executionId,
325+
workflowId
326+
)
327+
if (finalized) return true
328+
logger.warn('Paused cancellation terminal cleanup was rejected', {
329+
executionId,
330+
attempt,
331+
})
332+
} catch (error) {
333+
logger.warn('Failed to finalize paused cancellation after terminal race', {
334+
executionId,
335+
attempt,
336+
error: toError(error).message,
337+
})
338+
}
339+
if (attempt < PAUSED_CANCELLATION_DB_ATTEMPTS) {
340+
await sleep(PAUSED_CANCELLATION_DB_RETRY_MS)
341+
}
342+
}
343+
return false
344+
}
345+
317346
async function restorePausedCancellationAfterRejectedCommit(args: {
318347
executionId: string
319348
workflowId: string
@@ -346,10 +375,10 @@ async function restorePausedCancellationAfterRejectedCommit(args: {
346375
return clearPausedCancellationIntentWithRetry(args.executionId, args.workflowId)
347376
}
348377

349-
function throwPausedCancellationRestoreFailed(): never {
378+
function throwPausedCancellationReconciliationFailed(): never {
350379
throw new OrchestrationError(
351380
'internal',
352-
'Failed to restore paused execution after cancellation was rejected'
381+
'Failed to reconcile paused execution after cancellation was rejected'
353382
)
354383
}
355384

@@ -742,11 +771,11 @@ export async function cancelWorkflowExecution({
742771
}
743772

744773
if (execution.status !== 'running' && execution.status !== 'pending') {
745-
const pausedCancellationRestored = await clearPausedCancellationIntentWithRetry(
774+
const pausedCancellationFinalized = await finalizePausedCancellationForTerminalRunWithRetry(
746775
executionId,
747776
workflowId
748777
)
749-
if (!pausedCancellationRestored) throwPausedCancellationRestoreFailed()
778+
if (!pausedCancellationFinalized) throwPausedCancellationReconciliationFailed()
750779

751780
if (!isWorkflowGroupExecution && isWorkflowRunAlreadyTerminalStatus(execution.status)) {
752781
throw new WorkflowRunAlreadyTerminalError({
@@ -926,22 +955,32 @@ export async function cancelWorkflowExecution({
926955
effectivePausedCancellationPath,
927956
activeResumeEntryId,
928957
})
929-
if (!pausedCancellationRestored) throwPausedCancellationRestoreFailed()
958+
if (!pausedCancellationRestored) throwPausedCancellationReconciliationFailed()
930959
throw new OrchestrationError(
931960
'conflict',
932961
'Workflow group execution is no longer the active table execution'
933962
)
934963
}
935964

936965
if (competingTerminalStatus) {
937-
await clearStopSignalMarkers(stopSummary)
938-
const pausedCancellationRestored = await restorePausedCancellationAfterRejectedCommit({
939-
executionId,
940-
workflowId,
941-
effectivePausedCancellationPath,
942-
activeResumeEntryId,
943-
})
944-
if (!pausedCancellationRestored) throwPausedCancellationRestoreFailed()
966+
if (isWorkflowGroupExecution) {
967+
await clearStopSignalMarkers(stopSummary)
968+
const pausedCancellationRestored = await restorePausedCancellationAfterRejectedCommit({
969+
executionId,
970+
workflowId,
971+
effectivePausedCancellationPath,
972+
activeResumeEntryId,
973+
})
974+
if (!pausedCancellationRestored) throwPausedCancellationReconciliationFailed()
975+
} else if (effectivePausedCancellationPath) {
976+
const pausedCancellationFinalized = await finalizePausedCancellationForTerminalRunWithRetry(
977+
executionId,
978+
workflowId
979+
)
980+
if (!pausedCancellationFinalized) throwPausedCancellationReconciliationFailed()
981+
} else {
982+
await clearStopSignalMarkers(stopSummary)
983+
}
945984
if (
946985
!isWorkflowGroupExecution &&
947986
isWorkflowRunAlreadyTerminalStatus(competingTerminalStatus)

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

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1282,6 +1282,29 @@ describe('PauseResumeManager paused cancellation after pause release', () => {
12821282
expect(dbChainMockFns.set).not.toHaveBeenCalled()
12831283
})
12841284

1285+
it('finalizes staged pause state without mutating a terminal parent log', async () => {
1286+
queueTableRows(workflowExecutionLogs, [{ status: 'completed' }])
1287+
queueTableRows(pausedExecutions, [{ id: 'paused-exec-1', status: 'cancelling' }])
1288+
queueTableRows(resumeQueue, [{ id: 'resume-entry-1' }])
1289+
1290+
await expect(
1291+
PauseResumeManager.finalizePausedCancellationForTerminalRun('execution-1', 'workflow-1')
1292+
).resolves.toBe(true)
1293+
1294+
expect(dbChainMockFns.set).toHaveBeenNthCalledWith(1, {
1295+
status: 'cancelled',
1296+
updatedAt: expect.any(Date),
1297+
nextResumeAt: null,
1298+
})
1299+
expect(dbChainMockFns.set).toHaveBeenNthCalledWith(2, {
1300+
status: 'failed',
1301+
completedAt: expect.any(Date),
1302+
failureReason: 'Paused execution cancelled',
1303+
})
1304+
expect(dbChainMockFns.update).not.toHaveBeenCalledWith(workflowExecutionLogs)
1305+
expect(mockReleaseExecutionSlot).toHaveBeenCalledWith('resume-entry-1')
1306+
})
1307+
12851308
it('restores only cancellation-staged queue entries while the workflow remains active', async () => {
12861309
queueTableRows(workflowExecutionLogs, [{ status: 'running' }])
12871310
queueTableRows(pausedExecutions, [{ id: 'paused-exec-1' }])

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

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2732,6 +2732,102 @@ export class PauseResumeManager {
27322732
return transition.cancelled
27332733
}
27342734

2735+
/**
2736+
* Finalizes only pause and resume state when a non-cancellation terminal
2737+
* transition wins the parent execution race. The parent log is locked and
2738+
* inspected but never mutated, so a late claimed resume cannot revive it.
2739+
*/
2740+
static async finalizePausedCancellationForTerminalRun(
2741+
executionId: string,
2742+
workflowId: string
2743+
): Promise<boolean> {
2744+
const now = new Date()
2745+
2746+
const transition = await execDb.transaction(async (tx) => {
2747+
const executionLog = await tx
2748+
.select({ status: workflowExecutionLogs.status })
2749+
.from(workflowExecutionLogs)
2750+
.where(
2751+
and(
2752+
eq(workflowExecutionLogs.executionId, executionId),
2753+
eq(workflowExecutionLogs.workflowId, workflowId)
2754+
)
2755+
)
2756+
.for('update')
2757+
.limit(1)
2758+
.then((rows) => rows[0])
2759+
2760+
if (executionLog?.status === 'running' || executionLog?.status === 'pending') {
2761+
return { finalized: false, claimedResumeEntryIds: [] as string[] }
2762+
}
2763+
2764+
const pausedExecution = await tx
2765+
.select({ id: pausedExecutions.id, status: pausedExecutions.status })
2766+
.from(pausedExecutions)
2767+
.where(
2768+
and(
2769+
eq(pausedExecutions.executionId, executionId),
2770+
eq(pausedExecutions.workflowId, workflowId),
2771+
inArray(pausedExecutions.status, ['cancelling', 'cancelled'])
2772+
)
2773+
)
2774+
.for('update')
2775+
.limit(1)
2776+
.then((rows) => rows[0])
2777+
2778+
if (!pausedExecution) {
2779+
return { finalized: true, claimedResumeEntryIds: [] as string[] }
2780+
}
2781+
2782+
const claimedResumeEntries = await tx
2783+
.select({ id: resumeQueue.id })
2784+
.from(resumeQueue)
2785+
.where(
2786+
and(
2787+
eq(resumeQueue.parentExecutionId, executionId),
2788+
eq(resumeQueue.pausedExecutionId, pausedExecution.id),
2789+
eq(resumeQueue.status, 'claimed')
2790+
)
2791+
)
2792+
.for('update')
2793+
2794+
if (pausedExecution.status !== 'cancelled') {
2795+
await tx
2796+
.update(pausedExecutions)
2797+
.set({ status: 'cancelled', updatedAt: now, nextResumeAt: null })
2798+
.where(
2799+
and(
2800+
eq(pausedExecutions.id, pausedExecution.id),
2801+
eq(pausedExecutions.status, 'cancelling')
2802+
)
2803+
)
2804+
}
2805+
2806+
await tx
2807+
.update(resumeQueue)
2808+
.set({
2809+
status: 'failed',
2810+
completedAt: now,
2811+
failureReason: 'Paused execution cancelled',
2812+
})
2813+
.where(
2814+
and(
2815+
eq(resumeQueue.parentExecutionId, executionId),
2816+
eq(resumeQueue.pausedExecutionId, pausedExecution.id),
2817+
inArray(resumeQueue.status, ['pending', 'claimed'])
2818+
)
2819+
)
2820+
2821+
return {
2822+
finalized: true,
2823+
claimedResumeEntryIds: claimedResumeEntries.map((entry) => entry.id),
2824+
}
2825+
})
2826+
2827+
await releaseCancelledResumeReservations(transition.claimedResumeEntryIds)
2828+
return transition.finalized
2829+
}
2830+
27352831
static async blockQueuedResumesForCancellation(
27362832
executionId: string,
27372833
workflowId: string

0 commit comments

Comments
 (0)