feat: cancel queued jobs with an AbortSignal

README.md

@@ -360,6 +360,22 @@ wrapped.withOptions({/* options */}, arg1, arg2);
 | `weight`     | `1`                | Must be an integer equal to or higher than `0`. The `weight` is what increases the number of running jobs (up to `maxConcurrent`) and decreases the `reservoir` value.                                                                                                                |
 | `expiration` | `null` (unlimited) | The number of milliseconds a job is given to complete. Jobs that execute for longer than `expiration` ms will be failed with a `BottleneckError`.                                                                                                                                     |
 | `id`         | `<no-id>`          | You should give an ID to your jobs, it helps with [debugging](#debugging-your-application).                                                                                                                                                                                           |
+| `signal`     | `null`             | An `AbortSignal` that cancels the job if it hasn't started yet. See [Cancelling jobs](#cancelling-jobs).                                                                                                                                                                              |
+
+#### Cancelling jobs
+
+Pass an `AbortSignal` as `signal` to cancel a job that is still waiting. If the signal aborts before the job starts, it is removed from the queue, its task never runs, and its promise rejects with `signal.reason`. This applies even when `rejectOnDrop` is `false`. A signal that is already aborted rejects right away.
+
+```js
+const controller = new AbortController();
+const result = limiter.schedule({ signal: controller.signal }, fetchUser, id);
+controller.abort(); // rejects with an AbortError if the job hasn't started
+
+// Give up on jobs that wait too long: rejects with a TimeoutError
+limiter.schedule({ signal: AbortSignal.timeout(5000) }, fetchUser, id);
+```
+
+Aborting a job that has already started doesn't interrupt it; to cancel in-flight work, pass the same signal into the task (e.g. to `fetch`). Once the signal has aborted, the job won't start another attempt: if the running attempt fails, or the abort lands during a retry delay, the job rejects with `signal.reason` without firing the `"failed"` event or retrying. This also covers chained limiters, where an abort in the chained limiter's queue rejects the outer job.
 
 ### Strategies
 
@@ -1061,6 +1077,7 @@ try {
 | ------------------- | ------------------------------------------------------------------- |
 | `DROPPED`           | The job was shed by the queue strategy, or dropped by `stop()`.     |
 | `EXPIRED`           | The job ran longer than its `expiration`.                           |
+| `ABORTED`           | The job's `signal` aborted without a `reason` (Node < 16.14).       |
 | `STOPPED`           | The job was submitted after `stop()`, or `stop()` was called twice. |
 | `DUPLICATE_JOB_ID`  | A job with the same `id` already exists.                            |
 | `OVERWEIGHT`        | The job's `weight` exceeds the limiter's `maxConcurrent`.           |

src/Bottleneck.ts

@@ -13,6 +13,7 @@ import type {
   StrategyConstants,
 } from "./types";
 import pkg from "../package.json" with { type: "json" };
+import { abortReason } from "./abort";
 import Batcher from "./Batcher";
 import BottleneckError from "./BottleneckError";
 import IORedisConnection from "./cluster/IORedisConnection";
@@ -75,6 +76,7 @@ class Bottleneck {
     weight: 1,
     expiration: null,
     id: "<no-id>",
+    signal: null,
   };
   /** @internal */
   storeDefaults: StoreOptions = {
@@ -174,6 +176,7 @@ class Bottleneck {
 
   constructor(options?: BottleneckOptions, ...invalid: unknown[]) {
     this._addToQueue = this._addToQueueImpl.bind(this);
+    this._abortJob = this._abortJob.bind(this);
     options ??= {};
     this._validateOptions(options, invalid);
     load(options, this.instanceDefaults, this);
@@ -361,7 +364,12 @@ class Bottleneck {
 
       this.Events.trigger("debug", `Drained ${options.id}`, { success, args, options });
 
-      if (success) {
+      if (success && next.options.signal?.aborted) {
+        // Aborted during the register: it already left the queue, but the
+        // slot (and any reservoir/minTime it consumed) was taken for it.
+        await this._store.__free__(index, options.weight);
+        return 0;
+      } else if (success) {
         queue.shift();
         const empty = this.empty();
         if (empty) {
@@ -466,6 +474,10 @@ class Bottleneck {
   async _addToQueueImpl(job: Job): Promise<boolean> {
     let blocked: boolean, reachedHWM: boolean, strategy: unknown;
     const { args, options } = job;
+    // Aborted while RECEIVED: it already rejected and left States.
+    if (options.signal?.aborted) {
+      return false;
+    }
     try {
       ({ reachedHWM, blocked, strategy } = await this._store.__submit__(
         this.queued(),
@@ -476,6 +488,9 @@ class Bottleneck {
       job.doDrop({ error });
       return false;
     }
+    if (options.signal?.aborted) {
+      return false;
+    }
 
     if (blocked) {
       job.doDrop();
@@ -506,6 +521,20 @@ class Bottleneck {
     return reachedHWM;
   }
 
+  /** @internal Only reached while the job is RECEIVED or QUEUED. */
+  _abortJob(job: Job): void {
+    const status = this._states.jobStatus(job.options.id);
+    if (status === "QUEUED") {
+      this._queues.remove(job);
+      job.doAbort();
+      if (this.empty()) {
+        this.Events.trigger("empty");
+      }
+    } else if (status === "RECEIVED") {
+      job.doAbort();
+    }
+  }
+
   /** @internal */
   _receive: (job: Job) => unknown = (job: Job): unknown => {
     if (this._states.jobStatus(job.options.id) != null) {
@@ -555,7 +584,13 @@ class Bottleneck {
         Bottleneck.prototype.schedule,
         this.schedule,
       ),
+      this._abortJob,
     );
+    const { signal } = job.options;
+    if (signal?.aborted) {
+      job._reject(abortReason(signal));
+      return job.promise;
+    }
     this._receive(job);
     return job.promise;
   }

src/BottleneckError.ts

@@ -6,6 +6,7 @@
 export type BottleneckErrorCode =
   | "DROPPED"
   | "EXPIRED"
+  | "ABORTED"
   | "STOPPED"
   | "DUPLICATE_JOB_ID"
   | "OVERWEIGHT"

src/DLList.ts

@@ -1,4 +1,4 @@
-type Node<T> = {
+export type Node<T> = {
   value: T;
   prev: Node<T> | null;
   next: Node<T> | null;
@@ -17,16 +17,37 @@ class DLList<T = unknown> {
     this.incr = incr;
     this.decr = decr;
   }
-  push(value: T): void {
+  push(value: T): Node<T> {
     this.length++;
     this.incr?.();
-    const node = { value, prev: this._last, next: null };
+    const node: Node<T> = { value, prev: this._last, next: null };
     if (this._last != null) {
       this._last.next = node;
       this._last = node;
     } else {
       this._first = this._last = node;
     }
+    return node;
+  }
+  /** Unlink `node` in O(1). False if it isn't in this list (already shifted or removed). */
+  remove(node: Node<T>): boolean {
+    if (node.prev != null ? node.prev.next !== node : this._first !== node) {
+      return false;
+    }
+    if (node.prev != null) {
+      node.prev.next = node.next;
+    } else {
+      this._first = node.next;
+    }
+    if (node.next != null) {
+      node.next.prev = node.prev;
+    } else {
+      this._last = node.prev;
+    }
+    node.prev = node.next = null;
+    this.length--;
+    this.decr?.();
+    return true;
   }
   shift(): T | undefined {
     if (this._first == null) {

src/Job.ts

@@ -1,8 +1,10 @@
 import type { AsyncResource } from "node:async_hooks";
 import type Bottleneck from "./Bottleneck";
+import type { Node } from "./DLList";
 import type Events from "./Events";
 import type States from "./States";
 import type { EventInfo, EventInfoRetryable, JobDefaults, ResolvedJobOptions } from "./types";
+import { abortReason, unwatchAbort, watchAbort } from "./abort";
 import { captureAsyncResource, runWithAsyncResource } from "./async-context";
 import BottleneckError from "./BottleneckError";
 import { load } from "./parser";
@@ -44,6 +46,15 @@ class Job {
   _resolve: (value: unknown) => void = null as never;
   /** @internal */
   _reject: (reason?: unknown) => void = null as never;
+  /**
+   * Called when `options.signal` aborts while the job is RECEIVED or QUEUED.
+   * @internal
+   */
+  _onAbort: ((job: Job) => void) | undefined;
+  /** @internal */
+  _watchingSignal = false;
+  /** This job's node in its priority queue, for O(1) removal. @internal */
+  _queueNode: Node<unknown> | undefined;
 
   constructor(
     task: (...args: never[]) => unknown,
@@ -54,6 +65,7 @@ class Job {
     Events: Events,
     _states: States,
     scheduleStackCapture?: ScheduleStackCapture,
+    onAbort?: (job: Job) => void,
   ) {
     this.task = task;
     this.args = args;
@@ -62,6 +74,7 @@ class Job {
     this._states = _states;
     this.asyncResource = captureAsyncResource();
     this.scheduleStackCapture = scheduleStackCapture;
+    this._onAbort = onAbort;
     this.options = load(options ?? {}, jobDefaults) as ResolvedJobOptions;
     this.options.priority = this._sanitizePriority(this.options.priority);
     if (this.options.id === jobDefaults.id) {
@@ -88,6 +101,7 @@ class Job {
 
   doDrop(params?: { error?: unknown; message?: string }): boolean {
     const { error, message = "This job has been dropped by Bottleneck" } = params || {};
+    this._detachSignal();
     if (this._states.remove(this.options.id)) {
       if (this.rejectOnDrop) {
         this._reject(error ?? new BottleneckError(message, "DROPPED"));
@@ -114,8 +128,15 @@ class Job {
     }
   }
 
+  // Watch only while the job holds its id in States, so an abort can never
+  // land on another job that reuses the id.
   doReceive(): Promise<unknown> | undefined {
     this._states.start(this.options.id);
+    const { signal } = this.options;
+    if (signal != null && this._onAbort != null) {
+      watchAbort(signal, this);
+      this._watchingSignal = true;
+    }
     return this.Events.trigger("received", { args: this.args, options: this.options });
   }
 
@@ -133,13 +154,46 @@ class Job {
   doRun(): Promise<unknown> | undefined {
     if (this.retryCount === 0) {
       this._assertStatus("QUEUED");
+      this._detachSignal();
       this._states.next(this.options.id);
     } else {
       this._assertStatus("EXECUTING");
     }
     return this.Events.trigger("scheduled", { args: this.args, options: this.options });
   }
 
+  /** @internal */
+  _detachSignal(): void {
+    if (this._watchingSignal) {
+      this._watchingSignal = false;
+      unwatchAbort(this.options.signal as AbortSignal, this);
+    }
+  }
+
+  /** The signal aborted while this job was watching it. @internal */
+  _abort(): void {
+    this._detachSignal();
+    this._onAbort?.(this);
+  }
+
+  /**
+   * Reject with the signal's reason. No "dropped" event, and no schedule
+   * stack: the reason belongs to the caller and may be shared by many jobs.
+   */
+  doAbort(): boolean {
+    this._detachSignal();
+    if (this._states.remove(this.options.id)) {
+      this._reject(abortReason(this.options.signal as AbortSignal));
+      this.Events.trigger("debug", `Aborted ${this.options.id}`, {
+        args: this.args,
+        options: this.options,
+      });
+      return true;
+    } else {
+      return false;
+    }
+  }
+
   async doExecute(
     chained: Bottleneck | null,
     clearGlobalState: () => boolean,
@@ -157,6 +211,11 @@ class Job {
       options: this.options as unknown as EventInfo["options"],
       retryCount: this.retryCount,
     };
+    // A retry is an attempt that hasn't started: skip it once the signal aborted.
+    const { signal } = this.options;
+    if (this.retryCount > 0 && signal?.aborted) {
+      return this._onFailure(abortReason(signal), eventInfo, clearGlobalState, run, free);
+    }
     this.Events.trigger("executing", eventInfo);
 
     try {
@@ -210,9 +269,14 @@ class Job {
     run: (retryAfter: number) => unknown,
     free: (options: ResolvedJobOptions, eventInfo: EventInfo) => Promise<unknown>,
   ): Promise<unknown> {
-    error = attachScheduleStack(error, this.scheduleStackCapture);
+    // Once the signal aborts, a failure (whatever its cause: the reason from a
+    // chained limiter or a task that honored the signal, or an unrelated error)
+    // ends the job as a cancellation: no annotation, no "failed", no retry.
+    const { signal } = this.options;
+    const aborted = signal?.aborted === true;
+    error = aborted ? abortReason(signal) : attachScheduleStack(error, this.scheduleStackCapture);
     if (clearGlobalState()) {
-      const retry = await this.Events.trigger("failed", error, eventInfo);
+      const retry = aborted ? null : await this.Events.trigger("failed", error, eventInfo);
       if (retry != null) {
         const retryAfter = ~~(retry as number);
         this.Events.trigger(
@@ -233,6 +297,7 @@ class Job {
 
   doDone(eventInfo: EventInfo): Promise<unknown> | undefined {
     this._assertStatus("EXECUTING");
+    this._detachSignal();
     this._states.next(this.options.id);
     return this.Events.trigger("done", eventInfo);
   }

src/Queues.ts

@@ -1,9 +1,9 @@
 import type Job from "./Job";
 import BottleneckError from "./BottleneckError";
-import DLList from "./DLList";
+import DLList, { type Node } from "./DLList";
 import Events from "./Events";
 
-type JobLike = { options: { priority: number } };
+type JobLike = { options: { priority: number }; _queueNode?: Node<unknown> | undefined };
 
 class Queues<J extends JobLike = Job> {
   Events: Events;
@@ -46,7 +46,13 @@ class Queues<J extends JobLike = Job> {
     if (list == null) {
       throw new BottleneckError(`No queue for priority ${job.options.priority}`);
     }
-    list.push(job);
+    job._queueNode = list.push(job) as Node<unknown>;
+  }
+
+  /** Take `job` out of its priority list in O(1). */
+  remove(job: J): boolean {
+    const node = job._queueNode as Node<J> | undefined;
+    return node != null && (this._lists[job.options.priority]?.remove(node) ?? false);
   }
 
   queued(priority?: number): number {