Skip to content

Commit 161f18d

Browse files
j15zj15z
committed
fix(workflows): reconcile replacement resume cancellation
1 parent a752aab commit 161f18d

2 files changed

Lines changed: 138 additions & 34 deletions

File tree

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

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -150,6 +150,12 @@ const ACTIVE_RESUME_TARGET = {
150150
resumeExecutionId: 'resume-ex-1',
151151
}
152152

153+
const REPLACEMENT_ACTIVE_RESUME_TARGET = {
154+
...ACTIVE_RESUME_TARGET,
155+
resumeEntryId: 'resume-entry-2',
156+
resumeExecutionId: 'resume-ex-2',
157+
}
158+
153159
describe('cancelWorkflowExecution', () => {
154160
beforeEach(() => {
155161
vi.clearAllMocks()
@@ -785,6 +791,85 @@ describe('cancelWorkflowExecution', () => {
785791
expect(mockClearExecutionCancellation).not.toHaveBeenCalled()
786792
})
787793

794+
it('finishes cancellation when a failed active-resume rollback detects a replacement', async () => {
795+
mockStagePausedCancellation
796+
.mockResolvedValueOnce({ kind: 'active_resume', target: ACTIVE_RESUME_TARGET })
797+
.mockResolvedValueOnce({
798+
kind: 'active_resume',
799+
target: REPLACEMENT_ACTIVE_RESUME_TARGET,
800+
})
801+
mockGetActiveResumeCancellationTarget.mockResolvedValueOnce(REPLACEMENT_ACTIVE_RESUME_TARGET)
802+
mockRollbackActiveResumeCancellation.mockResolvedValueOnce(false)
803+
mockMarkExecutionCancelled
804+
.mockResolvedValueOnce({ durablyRecorded: false, reason: 'redis_unavailable' })
805+
.mockResolvedValueOnce({ durablyRecorded: true, reason: 'recorded' })
806+
mockCompletePausedCancellation.mockResolvedValueOnce(true)
807+
808+
const response = await POST(makeRequest(), makeParams())
809+
810+
expect(response.status).toBe(200)
811+
await expect(response.json()).resolves.toMatchObject({
812+
success: true,
813+
durablyRecorded: true,
814+
pausedCancelled: true,
815+
reason: 'recorded',
816+
})
817+
expect(mockRollbackActiveResumeCancellation).toHaveBeenCalledWith(
818+
'ex-1',
819+
'wf-1',
820+
'resume-entry-1'
821+
)
822+
expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(1, 'resume-ex-1', {
823+
executionDeadlineAt: null,
824+
})
825+
expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(2, 'resume-ex-2', {
826+
executionDeadlineAt: null,
827+
})
828+
expect(mockCompletePausedCancellation).toHaveBeenCalledWith('ex-1', 'wf-1')
829+
})
830+
831+
it('finishes late pause cancellation when rollback detects a replacement resume', async () => {
832+
mockStagePausedCancellation
833+
.mockResolvedValueOnce({ kind: 'not_paused' })
834+
.mockResolvedValueOnce({ kind: 'active_resume', target: ACTIVE_RESUME_TARGET })
835+
.mockResolvedValueOnce({
836+
kind: 'active_resume',
837+
target: REPLACEMENT_ACTIVE_RESUME_TARGET,
838+
})
839+
mockGetActiveResumeCancellationTarget.mockResolvedValueOnce(REPLACEMENT_ACTIVE_RESUME_TARGET)
840+
mockRollbackActiveResumeCancellation.mockResolvedValueOnce(false)
841+
mockMarkExecutionCancelled
842+
.mockResolvedValueOnce({ durablyRecorded: false, reason: 'redis_unavailable' })
843+
.mockResolvedValueOnce({ durablyRecorded: false, reason: 'redis_unavailable' })
844+
.mockResolvedValueOnce({ durablyRecorded: true, reason: 'recorded' })
845+
mockCompletePausedCancellation.mockResolvedValueOnce(true)
846+
847+
const response = await POST(makeRequest(), makeParams())
848+
849+
expect(response.status).toBe(200)
850+
await expect(response.json()).resolves.toMatchObject({
851+
success: true,
852+
durablyRecorded: true,
853+
pausedCancelled: true,
854+
reason: 'recorded',
855+
})
856+
expect(mockRollbackActiveResumeCancellation).toHaveBeenCalledWith(
857+
'ex-1',
858+
'wf-1',
859+
'resume-entry-1'
860+
)
861+
expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(1, 'ex-1', {
862+
executionDeadlineAt: null,
863+
})
864+
expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(2, 'resume-ex-1', {
865+
executionDeadlineAt: null,
866+
})
867+
expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(3, 'resume-ex-2', {
868+
executionDeadlineAt: null,
869+
})
870+
expect(mockCompletePausedCancellation).toHaveBeenCalledWith('ex-1', 'wf-1')
871+
})
872+
788873
it('returns success when a paused HITL execution is cancelled directly in the database', async () => {
789874
mockStagePausedCancellation.mockResolvedValue({ kind: 'idle' })
790875
mockCompletePausedCancellation.mockResolvedValue(true)

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

Lines changed: 53 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -392,6 +392,49 @@ async function rollbackPausedCancellationAfterAbort(args: {
392392
return true
393393
}
394394

395+
async function rollbackActiveResumeAfterFailedSignal(args: {
396+
executionId: string
397+
workflowId: string
398+
resumeEntryId: string
399+
}): Promise<boolean> {
400+
try {
401+
const rolledBack = await PauseResumeManager.rollbackActiveResumeCancellation(
402+
args.executionId,
403+
args.workflowId,
404+
args.resumeEntryId
405+
)
406+
if (!rolledBack) {
407+
logger.warn('Active resume cancellation could not be rolled back; completing cancellation', {
408+
executionId: args.executionId,
409+
activeResumeEntryId: args.resumeEntryId,
410+
})
411+
}
412+
return rolledBack
413+
} catch (error) {
414+
logger.warn('Failed to roll back active resume cancellation; completing cancellation', {
415+
executionId: args.executionId,
416+
activeResumeEntryId: args.resumeEntryId,
417+
error: toError(error).message,
418+
})
419+
return false
420+
}
421+
}
422+
423+
function activeResumeSignalFailureResult(
424+
executionId: string,
425+
stopSummary: ExecutionStopSummary
426+
): CancelWorkflowExecutionResult {
427+
return {
428+
success: false,
429+
executionId,
430+
redisAvailable: stopSummary.cancellation.reason !== 'redis_unavailable',
431+
durablyRecorded: stopSummary.cancellation.durablyRecorded,
432+
locallyAborted: stopSummary.locallyAborted,
433+
pausedCancelled: false,
434+
reason: 'active_resume_signal_failed',
435+
}
436+
}
437+
395438
function resolveCancellationReason(args: {
396439
activeResumeSignalFailed: boolean
397440
pauseReconciliationFailed: boolean
@@ -676,26 +719,14 @@ export async function cancelWorkflowExecution({
676719

677720
if (!activeResumeSignalAccepted) {
678721
const failedResumeEntryId = activeResumeTarget.resumeEntryId
679-
await PauseResumeManager.rollbackActiveResumeCancellation(
722+
const rolledBack = await rollbackActiveResumeAfterFailedSignal({
680723
executionId,
681724
workflowId,
682-
failedResumeEntryId
683-
).catch((error) => {
684-
logger.warn('Failed to roll back active resume cancellation intent', {
685-
executionId,
686-
activeResumeEntryId: failedResumeEntryId,
687-
error: toError(error).message,
688-
})
725+
resumeEntryId: failedResumeEntryId,
689726
})
690-
await clearStopSignalMarkers(stopSummary)
691-
return {
692-
success: false,
693-
executionId,
694-
redisAvailable: stopSummary.cancellation.reason !== 'redis_unavailable',
695-
durablyRecorded: stopSummary.cancellation.durablyRecorded,
696-
locallyAborted: stopSummary.locallyAborted,
697-
pausedCancelled: false,
698-
reason: 'active_resume_signal_failed',
727+
if (rolledBack) {
728+
await clearStopSignalMarkers(stopSummary)
729+
return activeResumeSignalFailureResult(executionId, stopSummary)
699730
}
700731
}
701732
} else if (!effectivePausedCancellationPath && !isWorkflowGroupExecution) {
@@ -738,26 +769,14 @@ export async function cancelWorkflowExecution({
738769
})
739770
if (!activeResumeSignalAccepted) {
740771
const failedResumeEntryId = activeResumeTarget.resumeEntryId
741-
await PauseResumeManager.rollbackActiveResumeCancellation(
772+
const rolledBack = await rollbackActiveResumeAfterFailedSignal({
742773
executionId,
743774
workflowId,
744-
failedResumeEntryId
745-
).catch((error) => {
746-
logger.warn('Failed to roll back late active resume cancellation intent', {
747-
executionId,
748-
activeResumeEntryId: failedResumeEntryId,
749-
error: toError(error).message,
750-
})
775+
resumeEntryId: failedResumeEntryId,
751776
})
752-
await clearStopSignalMarkers(stopSummary)
753-
return {
754-
success: false,
755-
executionId,
756-
redisAvailable: stopSummary.cancellation.reason !== 'redis_unavailable',
757-
durablyRecorded: stopSummary.cancellation.durablyRecorded,
758-
locallyAborted: stopSummary.locallyAborted,
759-
pausedCancelled: false,
760-
reason: 'active_resume_signal_failed',
777+
if (rolledBack) {
778+
await clearStopSignalMarkers(stopSummary)
779+
return activeResumeSignalFailureResult(executionId, stopSummary)
761780
}
762781
}
763782
} else if (!effectivePausedCancellationPath) {

0 commit comments

Comments
 (0)