fix: unify timeout/cancel handling; decide terminal state when the run settles (#184)

merencia · web-flow · commit a8a71b432a41 · 2026-06-19T16:22:23.000-03:00
* fix: in inline mode, let the job's result decide on timeout/cancel

An inline run cannot be force-stopped, so eagerly applying a RetryTransition
when the timeout fires (while the job keeps running) re-queued the job and could
run a second copy concurrently. In inline mode the timeout/cancel now only
signal via this.abortSignal; the terminal state is decided when the job actually
settles: a job that honors the signal returns its own result, one that ignores
it simply completes. Thread mode is unchanged (the worker is terminated, so the
abort still decides).

Also capture the timeout handle and clearTimeout on settle so a job that
finishes before its timeout no longer leaves a dangling timer (it kept the event
loop alive and retained job/backend refs until it fired).

* refactor: address code-review follow-ups (robustness + cleanup)

- executor-manager: guard the cancellation poll against a missing job row
  (watchedJob?.state) and attach a .catch so the polling loop can't die on an
  unhandled rejection; log the underlying rejection when a job is aborted so a
  real post-abort error is not silently lost.
- core: move the abort-reason wire format and its (de)serialization into
  @sidequest/core (AbortReasonMessage + serializeAbortReason/deserializeAbortReason)
  so the encode (runner-pool) and decode (runner) no longer duplicate the mapping.
- runner: replace the nested ternary for picking the abort signal with an
  explicit if-chain (the two inputs are mutually exclusive by construction).

* refactor: unify timeout/cancel handling around the run's settle

Replace the inline-vs-thread special-casing with a single rule that holds for
every runner/grace combination:

- The terminal transition is only applied once the run actually ends, so a
  still-running job is never re-queued underneath itself (fixes the double
  execution in inline and in thread with abortGracePeriodMs &gt; 0).
- If the job returned a state (resolved), respect it and transition to that
  state, even if a timeout/cancel was signaled.
- If the run was hard-killed (rejected, no result), the abort reason decides:
  timeout -&gt; retry, cancellation -&gt; canceled.
- A real error -&gt; retry, as before.

The timeout timer now only signals the abort; it no longer eagerly applies a
retry. Removes the `inline` branch in execute().
diff --git a/packages/core/src/job/abort-reason.test.ts b/packages/core/src/job/abort-reason.test.ts
@@ -0,0 +1,24 @@
+import { describe, expect, it } from "vitest";
+import { deserializeAbortReason, JobCanceled, JobTimeout, serializeAbortReason } from "./abort-reason";
+
+describe("abort-reason", () => {
+  it("round-trips a JobTimeout reason through the wire form", () => {
+    const message = serializeAbortReason(new JobTimeout(1500));
+    expect(message).toEqual({ kind: "timeout", timeoutMs: 1500 });
+
+    const reason = deserializeAbortReason(message);
+    expect(reason).toBeInstanceOf(JobTimeout);
+    expect((reason as JobTimeout).timeoutMs).toBe(1500);
+  });
+
+  it("round-trips a JobCanceled reason through the wire form", () => {
+    const message = serializeAbortReason(new JobCanceled());
+    expect(message).toEqual({ kind: "canceled" });
+    expect(deserializeAbortReason(message)).toBeInstanceOf(JobCanceled);
+  });
+
+  it("treats any non-timeout reason as canceled", () => {
+    expect(serializeAbortReason(new Error("boom"))).toEqual({ kind: "canceled" });
+    expect(serializeAbortReason(undefined)).toEqual({ kind: "canceled" });
+  });
+});
diff --git a/packages/core/src/job/abort-reason.ts b/packages/core/src/job/abort-reason.ts
@@ -29,3 +29,26 @@ export class JobCanceled extends Error {
     this.name = "JobCanceled";
   }
 }
+
+/**
+ * Structured-clone-safe wire form of an {@link AbortReason}, used to convey the reason to a job
+ * running in a worker thread (a live {@link AbortSignal} cannot cross the thread boundary).
+ */
+export type AbortReasonMessage = { kind: "timeout"; timeoutMs: number } | { kind: "canceled" };
+
+/**
+ * Encodes an abort reason into its wire form. Anything that is not a {@link JobTimeout} is treated
+ * as a cancellation.
+ * @param reason The abort reason (typically `signal.reason`).
+ */
+export function serializeAbortReason(reason: unknown): AbortReasonMessage {
+  return reason instanceof JobTimeout ? { kind: "timeout", timeoutMs: reason.timeoutMs } : { kind: "canceled" };
+}
+
+/**
+ * Rebuilds the proper {@link AbortReason} instance from its wire form.
+ * @param message The wire-form message.
+ */
+export function deserializeAbortReason(message: AbortReasonMessage): AbortReason {
+  return message.kind === "timeout" ? new JobTimeout(message.timeoutMs) : new JobCanceled();
+}
diff --git a/packages/engine/src/execution/executor-manager.test.ts b/packages/engine/src/execution/executor-manager.test.ts
@@ -1,6 +1,13 @@
 import { sidequestTest, SidequestTestFixture } from "@/tests/fixture";
 import { Backend } from "@sidequest/backend";
-import { CompletedResult, CompleteTransition, JobData, RetryTransition, RunTransition } from "@sidequest/core";
+import {
+  CancelTransition,
+  CompletedResult,
+  CompleteTransition,
+  JobData,
+  RetryTransition,
+  RunTransition,
+} from "@sidequest/core";
 import { JobTransitioner } from "../job/job-transitioner";
 import { grantQueueConfig } from "../queue/grant-queue-config";
 import { DummyJob } from "../test-jobs/dummy-job";
@@ -94,6 +101,54 @@ describe("ExecutorManager", () => {
       inlineRunMock.mockReset();
     });
 
+    sidequestTest(
+      "inline: a job that ignores the timeout completes instead of being retried",
+      async ({ backend, config }) => {
+        jobData = await backend.updateJob({ ...jobData, state: "claimed", claimed_at: new Date(), timeout: 20 });
+        const queryConfig = await grantQueueConfig(backend, { name: "default", concurrency: 1 });
+        const executorManager = new ExecutorManager(backend, { ...config, runner: "inline" });
+
+        // The job ignores the abort signal and completes after the timeout has already fired.
+        inlineRunMock.mockImplementationOnce(
+          () =>
+            new Promise((resolve) =>
+              setTimeout(() => resolve({ __is_job_transition__: true, type: "completed", result: "done" }), 60),
+            ),
+        );
+
+        await executorManager.execute(queryConfig, jobData);
+
+        // Timeout fired (signal aborted) but inline applies the job's own result: completed, not a retry.
+        // eslint-disable-next-line @typescript-eslint/unbound-method
+        expect(JobTransitioner.apply).toHaveBeenCalledWith(backend, jobData, expect.any(CompleteTransition));
+        // eslint-disable-next-line @typescript-eslint/unbound-method
+        expect(JobTransitioner.apply).not.toHaveBeenCalledWith(backend, jobData, expect.any(RetryTransition));
+        await executorManager.destroy();
+        inlineRunMock.mockReset();
+      },
+    );
+
+    sidequestTest(
+      "inline: a job that ignores cancellation completes (its result wins)",
+      async ({ backend, config }) => {
+        // Pre-cancel in the DB so the first cancellation poll observes it immediately. JobTransitioner is
+        // mocked here, so the RunTransition does not overwrite the persisted state.
+        jobData = await backend.updateJob({ ...jobData, state: "canceled" });
+        const queryConfig = await grantQueueConfig(backend, { name: "default", concurrency: 1 });
+        const executorManager = new ExecutorManager(backend, { ...config, runner: "inline" });
+
+        inlineRunMock.mockResolvedValue({ __is_job_transition__: true, type: "completed", result: "done" });
+
+        await executorManager.execute(queryConfig, jobData);
+
+        // Inline cannot force-stop the job; it completed, so its result is applied.
+        // eslint-disable-next-line @typescript-eslint/unbound-method
+        expect(JobTransitioner.apply).toHaveBeenCalledWith(backend, jobData, expect.any(CompleteTransition));
+        await executorManager.destroy();
+        inlineRunMock.mockReset();
+      },
+    );
+
     sidequestTest("should abort job execution on job cancel", async ({ backend, config }) => {
       await backend.updateJob({ ...jobData, state: "claimed", claimed_at: new Date() });
 
@@ -132,36 +187,54 @@ describe("ExecutorManager", () => {
       await executorManager.destroy();
     });
 
-    sidequestTest(
-      "does not overwrite the canceled state when an aborted job ignores the signal",
-      async ({ backend, config }) => {
-        await backend.updateJob({ ...jobData, state: "claimed", claimed_at: new Date() });
+    sidequestTest("respects a job's returned result even after a cancel was signaled", async ({ backend, config }) => {
+      await backend.updateJob({ ...jobData, state: "claimed", claimed_at: new Date() });
 
-        const queryConfig = await grantQueueConfig(backend, { name: "default", concurrency: 1 });
-        const executorManager = new ExecutorManager(backend, config);
-
-        // The job is canceled mid-run but ignores the abort signal and runs to completion.
-        runMock.mockImplementationOnce(async (job: JobData, signal: AbortSignal) => {
-          await backend.updateJob({ ...job, state: "canceled" });
-          while (!signal.aborted) {
-            await new Promise((r) => setTimeout(r, 50));
-          }
-          return { __is_job_transition__: true, type: "completed", result: "result" } as CompletedResult;
+      const queryConfig = await grantQueueConfig(backend, { name: "default", concurrency: 1 });
+      const executorManager = new ExecutorManager(backend, config);
+
+      // The job is canceled mid-run but ignores the abort signal and returns a completed result.
+      runMock.mockImplementationOnce(async (job: JobData, signal: AbortSignal) => {
+        await backend.updateJob({ ...job, state: "canceled" });
+        while (!signal.aborted) {
+          await new Promise((r) => setTimeout(r, 50));
+        }
+        return { __is_job_transition__: true, type: "completed", result: "result" } as CompletedResult;
+      });
+
+      await executorManager.execute(queryConfig, jobData);
+
+      // The job returned a state, so it is respected: the completion is applied.
+      // eslint-disable-next-line @typescript-eslint/unbound-method
+      expect(JobTransitioner.apply).toHaveBeenCalledWith(backend, jobData, expect.any(CompleteTransition));
+
+      await executorManager.destroy();
+    });
+
+    sidequestTest("a hard-killed canceled job transitions to canceled", async ({ backend, config }) => {
+      await backend.updateJob({ ...jobData, state: "claimed", claimed_at: new Date() });
+
+      const queryConfig = await grantQueueConfig(backend, { name: "default", concurrency: 1 });
+      const executorManager = new ExecutorManager(backend, config);
+
+      // The job is canceled and never produces a result (the worker is terminated -> the run rejects).
+      runMock.mockImplementationOnce(async (job: JobData, signal: AbortSignal) => {
+        await backend.updateJob({ ...job, state: "canceled" });
+        return new Promise((_, reject) => {
+          signal.addEventListener("abort", () => reject(new Error("The task has been aborted")));
         });
+      });
 
-        await executorManager.execute(queryConfig, jobData);
+      await executorManager.execute(queryConfig, jobData);
 
-        // The terminal completion transition must be skipped so the canceled state is preserved.
-        // eslint-disable-next-line @typescript-eslint/unbound-method
-        expect(JobTransitioner.apply).not.toHaveBeenCalledWith(
-          backend,
-          expect.anything(),
-          expect.any(CompleteTransition),
-        );
+      // No result: the abort reason (canceled) decides the terminal state.
+      // eslint-disable-next-line @typescript-eslint/unbound-method
+      expect(JobTransitioner.apply).toHaveBeenCalledWith(backend, jobData, expect.any(CancelTransition));
+      // eslint-disable-next-line @typescript-eslint/unbound-method
+      expect(JobTransitioner.apply).not.toHaveBeenCalledWith(backend, jobData, expect.any(CompleteTransition));
 
-        await executorManager.destroy();
-      },
-    );
+      await executorManager.destroy();
+    });
 
     sidequestTest("should abort job execution on timeout", async ({ backend, config }) => {
       jobData = await backend.updateJob({ ...jobData, state: "claimed", claimed_at: new Date(), timeout: 100 });
diff --git a/packages/engine/src/execution/executor-manager.ts b/packages/engine/src/execution/executor-manager.ts
@@ -1,5 +1,6 @@
 import { Backend } from "@sidequest/backend";
 import {
+  CancelTransition,
   JobCanceled,
   JobData,
   JobTimeout,
@@ -100,6 +101,7 @@ export class ExecutorManager {
   async execute(queueConfig: QueueConfig, job: JobData): Promise<void> {
     let isRunning = false;
     const controller = new AbortController();
+    let timeoutHandle: ReturnType<typeof setTimeout> | undefined;
     try {
       logger("Executor Manager").debug(`Submitting job ${job.id} for execution in queue ${queueConfig.name}`);
       // We call prepareJob here again to make sure the jobs are in the queues.
@@ -111,8 +113,10 @@ export class ExecutorManager {
       isRunning = true;
       const cancellationCheck = async () => {
         while (isRunning) {
+          // The row can be missing transiently or if it was deleted; treat that as "not canceled"
+          // rather than dereferencing undefined and crashing the polling loop.
           const watchedJob = await this.backend.getJob(job.id);
-          if (watchedJob!.state === "canceled") {
+          if (watchedJob?.state === "canceled") {
             logger("Executor Manager").debug(`Aborting job ${job.id}: canceled`);
             controller.abort(new JobCanceled());
             isRunning = false;
@@ -121,47 +125,50 @@ export class ExecutorManager {
           await new Promise((r) => setTimeout(r, 1000));
         }
       };
-      void cancellationCheck();
+      void cancellationCheck().catch((error) => {
+        logger("Executor Manager").error(`Cancellation watcher for job ${job.id} failed:`, error);
+      });
 
       logger("Executor Manager").debug(`Running job ${job.id} in queue ${queueConfig.name}`);
 
       const runPromise = this.jobRunner.run(job, controller.signal);
 
       if (job.timeout) {
-        void new Promise(() => {
-          setTimeout(() => {
-            logger("Executor Manager").debug(`Job ${job.id} timed out after ${job.timeout}ms, aborting.`);
-            controller.abort(new JobTimeout(job.timeout!));
-            void JobTransitioner.apply(this.backend, job, new RetryTransition(`Job timed out after ${job.timeout}ms`));
-          }, job.timeout!);
-        });
+        // Only signal the abort here. The terminal transition is decided when the run actually ends
+        // (resolve or reject) so a still-running job is never re-queued underneath itself.
+        timeoutHandle = setTimeout(() => {
+          logger("Executor Manager").debug(`Job ${job.id} timed out after ${job.timeout}ms, aborting.`);
+          controller.abort(new JobTimeout(job.timeout!));
+        }, job.timeout);
       }
 
       const result = await runPromise;
 
       isRunning = false;
-      // If the job was aborted (canceled or timed out) the terminal state was already decided by that
-      // path. Skip the terminal transition so a job that ignored the signal and ran to completion does
-      // not overwrite the canceled/retry state.
-      if (!controller.signal.aborted) {
-        logger("Executor Manager").debug(`Job ${job.id} completed with result: ${inspect(result)}`);
-        const transition = JobTransitionFactory.create(result);
-        await JobTransitioner.apply(this.backend, job, transition);
-      }
+      // The job ran to a conclusion and returned a state (even if a timeout/cancel was signaled);
+      // respect it and transition accordingly.
+      logger("Executor Manager").debug(`Job ${job.id} settled with result: ${inspect(result)}`);
+      await JobTransitioner.apply(this.backend, job, JobTransitionFactory.create(result));
     } catch (error: unknown) {
       isRunning = false;
       const err = error as Error;
-      // The thread runner rejects the run when its worker is aborted. Detect that via the signal
-      // rather than the rejection message (which varies). The terminal state was already set by the
-      // timeout (retry) or cancellation (canceled) path, so there is nothing more to do here.
       if (controller.signal.aborted) {
-        logger("Executor Manager").debug(`Job ${job.id} was aborted: ${String(controller.signal.reason)}`);
+        // The run produced no result because the worker was hard-killed (thread). The abort reason
+        // decides the terminal state: a timeout becomes a retry, anything else (cancellation) becomes
+        // canceled. The rejection is logged so a real error during the abort is not lost.
+        const reason: unknown = controller.signal.reason;
+        logger("Executor Manager").debug(`Job ${job.id} was hard-killed (${String(reason)}): ${err.message}`);
+        const transition = reason instanceof JobTimeout ? new RetryTransition(reason) : new CancelTransition();
+        await JobTransitioner.apply(this.backend, job, transition);
       } else {
         logger("Executor Manager").error(`Unhandled error while executing job ${job.id}: ${err.message}`);
         await JobTransitioner.apply(this.backend, job, new RetryTransition(err));
       }
     } finally {
       isRunning = false;
+      if (timeoutHandle) {
+        clearTimeout(timeoutHandle);
+      }
       this.activeByQueue[queueConfig.name].delete(job.id);
       this.activeJobs.delete(job.id);
     }
diff --git a/packages/engine/src/shared-runner/runner-pool.test.ts b/packages/engine/src/shared-runner/runner-pool.test.ts
@@ -1,9 +1,8 @@
 import { sidequestTest, SidequestTestFixture } from "@/tests/fixture";
-import { JobData, JobTimeout } from "@sidequest/core";
+import { AbortReasonMessage, JobData, JobTimeout } from "@sidequest/core";
 import { MessagePort } from "node:worker_threads";
 import { beforeEach, describe, expect, vi } from "vitest";
 import { DummyJob } from "../test-jobs/dummy-job";
-import { AbortPortMessage } from "./runner";
 import { RunnerPool } from "./runner-pool";
 
 const piscinaMockInstance = {
@@ -71,8 +70,8 @@ describe("RunnerPool", () => {
         const controller = new AbortController();
         const runPromise = gracePool.run(jobData, controller.signal);
 
-        const message = new Promise<AbortPortMessage>((resolve) =>
-          capturedPort!.once("message", (m: AbortPortMessage) => resolve(m)),
+        const message = new Promise<AbortReasonMessage>((resolve) =>
+          capturedPort!.once("message", (m: AbortReasonMessage) => resolve(m)),
         );
 
         controller.abort(new JobTimeout(5000));
diff --git a/packages/engine/src/shared-runner/runner-pool.ts b/packages/engine/src/shared-runner/runner-pool.ts
@@ -1,10 +1,9 @@
-import { JobData, JobResult, JobTimeout, logger } from "@sidequest/core";
+import { JobData, JobResult, logger, serializeAbortReason } from "@sidequest/core";
 import { MessageChannel } from "node:worker_threads";
 import Piscina from "piscina";
 import { DEFAULT_RUNNER_PATH } from "../constants";
 import { NonNullableEngineConfig } from "../engine";
 import { JobRunner } from "./job-runner";
-import type { AbortPortMessage } from "./runner";
 
 /**
  * A pool of worker threads for running jobs in parallel using Piscina.
@@ -56,10 +55,7 @@ export class RunnerPool implements JobRunner {
     let graceTimer: ReturnType<typeof setTimeout> | undefined;
 
     const onAbort = () => {
-      const reason: unknown = signal.reason;
-      const message: AbortPortMessage =
-        reason instanceof JobTimeout ? { kind: "timeout", timeoutMs: reason.timeoutMs } : { kind: "canceled" };
-      channel.port1.postMessage(message);
+      channel.port1.postMessage(serializeAbortReason(signal.reason));
       graceTimer = setTimeout(() => hardKill.abort(), grace);
     };
 
diff --git a/packages/engine/src/shared-runner/runner.ts b/packages/engine/src/shared-runner/runner.ts
