Skip to content

Commit e6c8777

Browse files
j15zj15z
committed
improvement(copilot): simplify cancellation lifecycle
1 parent 1f5bdab commit e6c8777

1 file changed

Lines changed: 59 additions & 95 deletions

File tree

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

Lines changed: 59 additions & 95 deletions
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,10 @@ interface ExecutionStopSummary extends ExecutionStopSignalResult {
8787
signalledExecutionIds: Set<string>
8888
}
8989

90+
type ActiveResumeCancellationTarget = NonNullable<
91+
Awaited<ReturnType<typeof PauseResumeManager.getActiveResumeCancellationTarget>>
92+
>
93+
9094
function createExecutionStopSummary(): ExecutionStopSummary {
9195
return {
9296
cancellation: { durablyRecorded: false, reason: 'redis_unavailable' },
@@ -142,35 +146,28 @@ async function signalExecutionStop(args: {
142146
}
143147
}
144148

145-
async function signalActiveResumeStop(args: {
149+
async function signalAndRecordActiveResumeStop(args: {
146150
workflowId: string
147151
executionId: string
148152
executionDeadlineAt: Date | null
149-
target: NonNullable<
150-
Awaited<ReturnType<typeof PauseResumeManager.getActiveResumeCancellationTarget>>
151-
>
152-
}): Promise<{
153-
target: NonNullable<
154-
Awaited<ReturnType<typeof PauseResumeManager.getActiveResumeCancellationTarget>>
155-
>
156-
signal: ExecutionStopSignalResult
157-
}> {
153+
target: ActiveResumeCancellationTarget
154+
summary: ExecutionStopSummary
155+
}): Promise<boolean> {
158156
const signal = await signalExecutionStop({
159157
workflowId: args.workflowId,
160158
signalExecutionId: args.target.resumeExecutionId,
161159
queueBindingExecutionId: args.executionId,
162160
executionDeadlineAt: args.executionDeadlineAt,
163161
queueScope: 'resume',
164162
})
165-
return { target: args.target, signal }
163+
mergeExecutionStopSignal(args.summary, args.target.resumeExecutionId, signal)
164+
return didActiveResumeStop(args.executionId, args.workflowId, args.target, signal)
166165
}
167166

168167
async function didActiveResumeStop(
169168
executionId: string,
170169
workflowId: string,
171-
target: NonNullable<
172-
Awaited<ReturnType<typeof PauseResumeManager.getActiveResumeCancellationTarget>>
173-
>,
170+
target: ActiveResumeCancellationTarget,
174171
signal: ExecutionStopSignalResult
175172
): Promise<boolean> {
176173
if (signal.accepted) return true
@@ -375,7 +372,33 @@ async function rollbackPausedCancellationAfterAbort(args: {
375372
return abortResponse
376373
}
377374

378-
/** Runs the existing workflow cancellation lifecycle after surface authentication. */
375+
function resolveCancellationReason(args: {
376+
activeResumeSignalFailed: boolean
377+
pauseReconciliationFailed: boolean
378+
effectivePausedCancellationPath: boolean
379+
cancellationEventPublished: boolean
380+
pausedCancelled: boolean
381+
stopSummary: ExecutionStopSummary
382+
}): NonNullable<CancelWorkflowExecutionResponse['reason']> {
383+
if (args.activeResumeSignalFailed) return 'active_resume_signal_failed'
384+
if (args.pauseReconciliationFailed) return 'paused_database_cancel_failed'
385+
if (args.effectivePausedCancellationPath && !args.cancellationEventPublished) {
386+
return 'paused_event_publish_failed'
387+
}
388+
if (args.effectivePausedCancellationPath && !args.pausedCancelled) {
389+
return 'paused_database_cancel_failed'
390+
}
391+
if (args.effectivePausedCancellationPath) return 'recorded'
392+
if (
393+
args.stopSummary.queueJobsCancelled > 0 &&
394+
!args.stopSummary.cancellation.durablyRecorded &&
395+
!args.stopSummary.locallyAborted
396+
) {
397+
return 'queue_cancelled'
398+
}
399+
return args.stopSummary.cancellation.reason
400+
}
401+
379402
export async function cancelWorkflowExecutionPostAuth({
380403
workflowId,
381404
executionId,
@@ -544,23 +567,13 @@ export async function cancelWorkflowExecutionPostAuth({
544567
let activeResumeSignalFailed = false
545568
let exactStopSatisfied = true
546569
if (pausedCancellationStage.kind === 'active_resume') {
547-
const activeResumeStop = await signalActiveResumeStop({
570+
exactStopSatisfied = await signalAndRecordActiveResumeStop({
548571
workflowId,
549572
executionId,
550573
executionDeadlineAt: execution.executionDeadlineAt,
551574
target: pausedCancellationStage.target,
575+
summary: stopSummary,
552576
})
553-
mergeExecutionStopSignal(
554-
stopSummary,
555-
activeResumeStop.target.resumeExecutionId,
556-
activeResumeStop.signal
557-
)
558-
exactStopSatisfied = await didActiveResumeStop(
559-
executionId,
560-
workflowId,
561-
activeResumeStop.target,
562-
activeResumeStop.signal
563-
)
564577
activeResumeSignalFailed = !exactStopSatisfied
565578
} else if (isWorkflowGroupExecution && !hasPausedCancellation) {
566579
const retrySignal = await signalExecutionStop({
@@ -657,23 +670,13 @@ export async function cancelWorkflowExecutionPostAuth({
657670
let activeResumeSignalAccepted = false
658671

659672
if (activeResumeTarget && !isWorkflowGroupExecution) {
660-
const activeResumeStop = await signalActiveResumeStop({
673+
activeResumeSignalAccepted = await signalAndRecordActiveResumeStop({
661674
workflowId,
662675
executionId,
663676
executionDeadlineAt: execution.executionDeadlineAt,
664677
target: activeResumeTarget,
678+
summary: stopSummary,
665679
})
666-
mergeExecutionStopSignal(
667-
stopSummary,
668-
activeResumeStop.target.resumeExecutionId,
669-
activeResumeStop.signal
670-
)
671-
activeResumeSignalAccepted = await didActiveResumeStop(
672-
executionId,
673-
workflowId,
674-
activeResumeStop.target,
675-
activeResumeStop.signal
676-
)
677680

678681
if (!activeResumeSignalAccepted) {
679682
const failedResumeEntryId = activeResumeTarget.resumeEntryId
@@ -730,23 +733,13 @@ export async function cancelWorkflowExecutionPostAuth({
730733
activeResumeEntryId = activeResumeTarget?.resumeEntryId ?? null
731734

732735
if (activeResumeTarget) {
733-
const activeResumeStop = await signalActiveResumeStop({
736+
activeResumeSignalAccepted = await signalAndRecordActiveResumeStop({
734737
workflowId,
735738
executionId,
736739
executionDeadlineAt: execution.executionDeadlineAt,
737740
target: activeResumeTarget,
741+
summary: stopSummary,
738742
})
739-
mergeExecutionStopSignal(
740-
stopSummary,
741-
activeResumeStop.target.resumeExecutionId,
742-
activeResumeStop.signal
743-
)
744-
activeResumeSignalAccepted = await didActiveResumeStop(
745-
executionId,
746-
workflowId,
747-
activeResumeStop.target,
748-
activeResumeStop.signal
749-
)
750743
if (!activeResumeSignalAccepted) {
751744
const failedResumeEntryId = activeResumeTarget.resumeEntryId
752745
await PauseResumeManager.rollbackActiveResumeCancellation(
@@ -919,23 +912,13 @@ export async function cancelWorkflowExecutionPostAuth({
919912
let activeResumeSignalFailed = false
920913
if (isWorkflowGroupExecution) {
921914
if (activeResumeTarget) {
922-
const activeResumeStop = await signalActiveResumeStop({
915+
activeResumeSignalAccepted = await signalAndRecordActiveResumeStop({
923916
workflowId,
924917
executionId,
925918
executionDeadlineAt: execution.executionDeadlineAt,
926919
target: activeResumeTarget,
920+
summary: stopSummary,
927921
})
928-
mergeExecutionStopSignal(
929-
stopSummary,
930-
activeResumeStop.target.resumeExecutionId,
931-
activeResumeStop.signal
932-
)
933-
activeResumeSignalAccepted = await didActiveResumeStop(
934-
executionId,
935-
workflowId,
936-
activeResumeStop.target,
937-
activeResumeStop.signal
938-
)
939922
activeResumeSignalFailed = !activeResumeSignalAccepted
940923
} else if (!effectivePausedCancellationPath) {
941924
const groupSignal = await signalExecutionStop({
@@ -954,32 +937,19 @@ export async function cancelWorkflowExecutionPostAuth({
954937
)
955938
if (isPausedCancellationStage(postClaimPausedCancellationStage)) {
956939
effectivePausedCancellationPath = true
957-
pausedCancellationStage = postClaimPausedCancellationStage
958940
if (postClaimPausedCancellationStage.kind === 'active_resume') {
959941
const currentActiveResume = postClaimPausedCancellationStage.target
960942
const alreadyAttemptedCurrentResume =
961943
currentActiveResume.resumeEntryId === activeResumeEntryId &&
962944
stopSummary.signalledExecutionIds.has(currentActiveResume.resumeExecutionId)
963945
if (!alreadyAttemptedCurrentResume) {
964-
const activeResumeStop = await signalActiveResumeStop({
946+
activeResumeSignalAccepted = await signalAndRecordActiveResumeStop({
965947
workflowId,
966948
executionId,
967949
executionDeadlineAt: execution.executionDeadlineAt,
968950
target: currentActiveResume,
951+
summary: stopSummary,
969952
})
970-
activeResumeEntryId = activeResumeStop.target.resumeEntryId
971-
activeResumeTarget = activeResumeStop.target
972-
mergeExecutionStopSignal(
973-
stopSummary,
974-
activeResumeStop.target.resumeExecutionId,
975-
activeResumeStop.signal
976-
)
977-
activeResumeSignalAccepted = await didActiveResumeStop(
978-
executionId,
979-
workflowId,
980-
activeResumeStop.target,
981-
activeResumeStop.signal
982-
)
983953
}
984954
activeResumeSignalFailed = !activeResumeSignalAccepted
985955
} else {
@@ -1046,21 +1016,14 @@ export async function cancelWorkflowExecutionPostAuth({
10461016
const durablyRecorded = effectivePausedCancellationPath
10471017
? true
10481018
: stopSummary.cancellation.durablyRecorded
1049-
const reason = activeResumeSignalFailed
1050-
? 'active_resume_signal_failed'
1051-
: pauseReconciliationFailed
1052-
? 'paused_database_cancel_failed'
1053-
: effectivePausedCancellationPath && !cancellationEventPublished
1054-
? 'paused_event_publish_failed'
1055-
: effectivePausedCancellationPath && !pausedCancelled
1056-
? 'paused_database_cancel_failed'
1057-
: effectivePausedCancellationPath
1058-
? 'recorded'
1059-
: stopSummary.queueJobsCancelled > 0 &&
1060-
!stopSummary.cancellation.durablyRecorded &&
1061-
!stopSummary.locallyAborted
1062-
? 'queue_cancelled'
1063-
: stopSummary.cancellation.reason
1019+
const reason = resolveCancellationReason({
1020+
activeResumeSignalFailed,
1021+
pauseReconciliationFailed,
1022+
effectivePausedCancellationPath,
1023+
cancellationEventPublished,
1024+
pausedCancelled,
1025+
stopSummary,
1026+
})
10641027

10651028
return cancellationOutcome({
10661029
success,
@@ -1075,13 +1038,14 @@ export async function cancelWorkflowExecutionPostAuth({
10751038
reason,
10761039
})
10771040
} catch (error) {
1041+
const normalizedError = toError(error)
10781042
logger.error('Failed to cancel execution', {
10791043
workflowId,
10801044
executionId,
1081-
error: toError(error).message,
1045+
error: normalizedError.message,
10821046
})
10831047
return NextResponse.json(
1084-
{ error: toError(error).message || 'Failed to cancel execution' },
1048+
{ error: normalizedError.message || 'Failed to cancel execution' },
10851049
{ status: 500 }
10861050
)
10871051
}

0 commit comments

Comments
 (0)