Skip to content

Commit e6fb280

Browse files
committed
fix(execution): abort local runs before the removal signal deadline can skip them
The signal deadline bounds only the Redis flag writes. Aborting a run this process holds needs no Redis, so every execution is aborted first, even after the deadline has passed. Corrects the ban's note on why a failed stop record is logged: banning again records the stops again.
1 parent cd54396 commit e6fb280

3 files changed

Lines changed: 48 additions & 12 deletions

File tree

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
/**
2+
* An in-process run is stopped by aborting it here, which needs no Redis. Failure mode: the
3+
* signal deadline passes before the Redis writes start, and the local runs are left going.
4+
*/
5+
import { describe, expect, it, vi } from 'vitest'
6+
import {
7+
registerManualExecutionAborter,
8+
unregisterManualExecutionAborter,
9+
} from '@/lib/execution/manual-cancellation'
10+
import { signalExecutionStops } from '@/lib/execution/removal-stop-records'
11+
12+
describe('signalling removed executions', () => {
13+
it('aborts the runs this process holds even after the signal deadline has passed', async () => {
14+
const abort = vi.fn()
15+
registerManualExecutionAborter('execution-local', abort)
16+
const deadline = new AbortController()
17+
deadline.abort()
18+
19+
try {
20+
await expect(
21+
signalExecutionStops(
22+
[{ executionId: 'execution-local', executionDeadlineAt: null }],
23+
deadline.signal
24+
)
25+
).resolves.toBe(1)
26+
} finally {
27+
unregisterManualExecutionAborter('execution-local', abort)
28+
}
29+
30+
expect(abort).toHaveBeenCalledOnce()
31+
})
32+
})

‎apps/sim/lib/execution/removal-stop-records.ts‎

Lines changed: 13 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -193,8 +193,9 @@ export async function recordExecutionStops(
193193
* Sets the cancellation flag each execution's engine watches, and aborts the ones this process
194194
* runs, without the queue sweep or the retry the outbox event adds. Callers whose module graph
195195
* cannot load the delivery stop in-flight runs with it at once and leave the rest to the outbox
196-
* worker. Returns how many flags Redis refused or the deadline (`signal`, else
197-
* {@link SIGNAL_DEADLINE_MS}) left unsent; never throws.
196+
* worker. The deadline (`signal`, else {@link SIGNAL_DEADLINE_MS}) bounds only the Redis writes;
197+
* every execution this process runs is aborted first. Returns how many flags Redis refused or
198+
* the deadline left unsent; never throws.
198199
*
199200
* Without Redis at all the flag is process-local by design (`markExecutionCancelled`) and a
200201
* retry from any process could reach no further, so that outcome does not count as refused.
@@ -203,6 +204,16 @@ export async function signalExecutionStops(
203204
executions: readonly Pick<PlannedExecutionStop, 'executionId' | 'executionDeadlineAt'>[],
204205
signal?: AbortSignal
205206
): Promise<number> {
207+
for (const { executionId } of executions) {
208+
try {
209+
abortManualExecution(executionId)
210+
} catch (error) {
211+
logger.warn('Failed to abort a local execution being stopped', {
212+
executionId,
213+
error: describeError(error),
214+
})
215+
}
216+
}
206217
const deadline = signal ?? AbortSignal.timeout(SIGNAL_DEADLINE_MS)
207218
let refused = 0
208219
await mapWithConcurrency(executions, SIGNAL_CONCURRENCY, async (execution) => {
@@ -222,14 +233,6 @@ export async function signalExecutionStops(
222233
error: describeError(error),
223234
})
224235
}
225-
try {
226-
abortManualExecution(execution.executionId)
227-
} catch (error) {
228-
logger.warn('Failed to abort a local execution being stopped', {
229-
executionId: execution.executionId,
230-
error: describeError(error),
231-
})
232-
}
233236
})
234237
return refused
235238
}

‎apps/sim/lib/workflows/lifecycle.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -324,8 +324,9 @@ export async function disableUserResources(userId: string): Promise<void> {
324324
/**
325325
* Records the workspaces' stops and signals their in-flight runs; the recorded stop's queue
326326
* sweep and retries run in the outbox worker, since this module sits in page graphs that must
327-
* not load that delivery. Logged, not thrown: the ban and archives have committed, and a retry
328-
* would find no unarchived workspace to stop.
327+
* not load that delivery. Logged, not thrown: the ban and archives have committed, and banning
328+
* the user again records the stops again, since every owned workspace is stopped, archived or
329+
* not.
329330
*/
330331
let stopFailed = false
331332
const stopRuns = async (workspaceIds: string[]): Promise<void> => {

0 commit comments

Comments
 (0)