Skip to content

Commit 464449f

Browse files
authored
fix: address PR #180 review feedback (#188)
- executor: stop treating a missing job row as a cancellation in the watcher. Jobs are not deleted in normal operation; the integration test's mid-run truncate is handled by the resilient terminal transition instead. The test's long job is shortened so it cannot outlive teardown. - executor: on a hard-kill, default to a retry and only map to canceled when the abort reason is clearly a JobCanceled (failsafe for unknown abort reasons). - runner-pool: if the signal is already aborted on entry, reject without submitting the job to the pool instead of running it anyway. - docs: remove the duplicated crash-isolation and inline-abort danger callouts from the execution-modes page (the tables already convey them).
1 parent 8470de5 commit 464449f

6 files changed

Lines changed: 28 additions & 49 deletions

File tree

packages/docs/production/execution-modes.md

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -36,10 +36,6 @@ Use `fork: false` when:
3636
- You're running an **integration test** and want to avoid IPC and process teardown flakiness.
3737
- Your jobs need access to **live, in-process state** that can't cross a process boundary, for example a dependency-injection container.
3838

39-
::: danger No crash isolation with `fork: false`
40-
With the default `fork: true`, a job that throws an unhandled exception or calls `process.exit()` only takes down the engine fork, and Sidequest restarts it. With `fork: false`, the engine shares your application's process: **a misbehaving job can crash your whole app.** Only use it when you understand and accept that.
41-
:::
42-
4339
## `runner`: thread pool vs inline
4440

4541
```typescript
@@ -159,9 +155,7 @@ this.abortSignal.addEventListener("abort", () => {
159155
| `runner: "thread"`, `abortGracePeriodMs: 0` (default) | No (worker is killed immediately) | Killed right away. |
160156
| `runner: "thread"`, `abortGracePeriodMs > 0` | Yes, for the grace window | Killed after the grace period. |
161157

162-
::: danger Inline timeout/cancel only work if your job honors the signal
163-
In `runner: "inline"` there is no way to forcibly stop a job. If your job does not pass `this.abortSignal` to its async work or check `this.abortSignal.aborted` / `throwIfAborted()`, then **timeouts and cancellation have no effect**: the job keeps running until it returns on its own. Treat `this.abortSignal` as mandatory for any long-running inline job.
164-
:::
158+
So in inline mode `this.abortSignal` is effectively mandatory for any long-running job: a job that does not honor it keeps running until it returns on its own (timeouts and cancellation cannot stop it).
165159

166160
### `abortGracePeriodMs`: graceful kill for thread jobs
167161

packages/engine/src/execution/executor-manager.test.ts

Lines changed: 0 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -236,31 +236,6 @@ describe("ExecutorManager", () => {
236236
await executorManager.destroy();
237237
});
238238

239-
sidequestTest("aborts a running job when its row no longer exists", async ({ backend, config }) => {
240-
await backend.updateJob({ ...jobData, state: "claimed", claimed_at: new Date() });
241-
242-
const queryConfig = await grantQueueConfig(backend, { name: "default", concurrency: 1 });
243-
const executorManager = new ExecutorManager(backend, config);
244-
245-
// Simulate the row being deleted/truncated while the job runs: the watcher must still stop it.
246-
const getJobSpy = vi.spyOn(backend, "getJob").mockResolvedValue(undefined);
247-
runMock.mockImplementationOnce(
248-
(_job: JobData, signal: AbortSignal) =>
249-
new Promise((_, reject) => {
250-
signal.addEventListener("abort", () => reject(new Error("The task has been aborted")));
251-
}),
252-
);
253-
254-
await executorManager.execute(queryConfig, jobData);
255-
256-
// eslint-disable-next-line @typescript-eslint/unbound-method
257-
expect(JobTransitioner.apply).toHaveBeenCalledWith(backend, jobData, expect.any(CancelTransition));
258-
expect(executorManager.totalActiveWorkers()).toBe(0);
259-
260-
await executorManager.destroy();
261-
getJobSpy.mockRestore();
262-
});
263-
264239
sidequestTest("does not crash when the final transition fails (job row gone)", async ({ backend, config }) => {
265240
const queryConfig = await grantQueueConfig(backend, { name: "default", concurrency: 1 });
266241
const executorManager = new ExecutorManager(backend, config);

packages/engine/src/execution/executor-manager.ts

Lines changed: 9 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -115,13 +115,8 @@ export class ExecutorManager {
115115
const cancellationCheck = async () => {
116116
while (isRunning) {
117117
const watchedJob = await this.backend.getJob(job.id);
118-
// Abort when the job was canceled, and also when its row no longer exists (deleted or
119-
// truncated): the record is gone, so the run must be stopped rather than left to block
120-
// shutdown forever.
121-
if (!watchedJob || watchedJob.state === "canceled") {
122-
logger("Executor Manager").debug(
123-
`Aborting job ${job.id}: ${watchedJob ? "canceled" : "row no longer exists"}`,
124-
);
118+
if (watchedJob?.state === "canceled") {
119+
logger("Executor Manager").debug(`Aborting job ${job.id}: canceled`);
125120
controller.abort(new JobCanceled());
126121
isRunning = false;
127122
return;
@@ -157,12 +152,15 @@ export class ExecutorManager {
157152
isRunning = false;
158153
const err = error as Error;
159154
if (controller.signal.aborted) {
160-
// The run produced no result because the worker was hard-killed (thread). The abort reason
161-
// decides the terminal state: a timeout becomes a retry, anything else (cancellation) becomes
162-
// canceled. The rejection is logged so a real error during the abort is not lost.
155+
// The run produced no result because the worker was hard-killed (thread). Only a clear
156+
// cancellation maps to canceled; every other abort reason (timeout, or anything else) defaults
157+
// to a retry as a failsafe. The rejection is logged so a real error during the abort is kept.
163158
const reason: unknown = controller.signal.reason;
164159
logger("Executor Manager").debug(`Job ${job.id} was hard-killed (${String(reason)}): ${err.message}`);
165-
const transition = reason instanceof JobTimeout ? new RetryTransition(reason) : new CancelTransition();
160+
const transition =
161+
reason instanceof JobCanceled
162+
? new CancelTransition()
163+
: new RetryTransition(reason instanceof Error ? reason : new Error(`Job aborted: ${String(reason)}`));
166164
await this.applyTerminalTransition(job, transition);
167165
} else {
168166
logger("Executor Manager").error(`Unhandled error while executing job ${job.id}: ${err.message}`);

packages/engine/src/shared-runner/runner-pool.test.ts

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,14 @@ describe("RunnerPool", () => {
4747
expect(piscinaMockInstance.run).toHaveBeenCalledWith({ jobData, config }, { signal });
4848
});
4949

50+
sidequestTest("rejects without running the job when the signal is already aborted", async () => {
51+
const controller = new AbortController();
52+
controller.abort(new Error("already gone"));
53+
54+
await expect(pool.run(jobData, controller.signal)).rejects.toThrow("already gone");
55+
expect(piscinaMockInstance.run).not.toHaveBeenCalled();
56+
});
57+
5058
sidequestTest(
5159
"with a grace period, delivers the abort over a port and hard-kills after the grace",
5260
async ({ config }) => {

packages/engine/src/shared-runner/runner-pool.ts

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,12 @@ export class RunnerPool implements JobRunner {
4242
*/
4343
run(job: JobData, signal?: AbortSignal): Promise<JobResult> {
4444
logger("RunnerPool").debug(`Running job ${job.id} in pool`);
45+
46+
// Already aborted before we could start (e.g. canceled between claim and dispatch): don't run it.
47+
if (signal?.aborted) {
48+
return Promise.reject(signal.reason instanceof Error ? signal.reason : new Error("Job aborted before execution"));
49+
}
50+
4551
const grace = this.nonNullConfig.abortGracePeriodMs;
4652

4753
if (!signal || grace <= 0) {
@@ -59,11 +65,7 @@ export class RunnerPool implements JobRunner {
5965
graceTimer = setTimeout(() => hardKill.abort(), grace);
6066
};
6167

62-
if (signal.aborted) {
63-
onAbort();
64-
} else {
65-
signal.addEventListener("abort", onAbort, { once: true });
66-
}
68+
signal.addEventListener("abort", onAbort, { once: true });
6769

6870
return this.pool
6971
.run(

tests/integration/shared-test-suite.js

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -310,7 +310,9 @@ export function createIntegrationTestSuite(Sidequest, jobs, moduleType = "ESM")
310310
queues: [{ name: "default" }],
311311
});
312312

313-
const jobData = await Sidequest.build(TimeoutJob).enqueue(1000000);
313+
// A few seconds: long enough to reliably cancel while running, short enough that it cannot
314+
// outlive the suite's teardown (which truncates the table) if the cancel watcher is mid-poll.
315+
const jobData = await Sidequest.build(TimeoutJob).enqueue(3000);
314316

315317
await vi.waitUntil(() => Sidequest.job.get(jobData.id).then((job) => job?.state === "running"), 5000);
316318
// Cancel the job while it's running

0 commit comments

Comments
 (0)