diff --git a/README.md b/README.md index e8aa7e64..b8764755 100644 --- a/README.md +++ b/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` | `` | 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`. | diff --git a/src/Bottleneck.ts b/src/Bottleneck.ts index dbc7143f..6dee1fc0 100644 --- a/src/Bottleneck.ts +++ b/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: "", + 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 { 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; } diff --git a/src/BottleneckError.ts b/src/BottleneckError.ts index 1dbda8f1..f3fe1e2d 100644 --- a/src/BottleneckError.ts +++ b/src/BottleneckError.ts @@ -6,6 +6,7 @@ export type BottleneckErrorCode = | "DROPPED" | "EXPIRED" + | "ABORTED" | "STOPPED" | "DUPLICATE_JOB_ID" | "OVERWEIGHT" diff --git a/src/DLList.ts b/src/DLList.ts index 545afb3f..f5735936 100644 --- a/src/DLList.ts +++ b/src/DLList.ts @@ -1,4 +1,4 @@ -type Node = { +export type Node = { value: T; prev: Node | null; next: Node | null; @@ -17,16 +17,37 @@ class DLList { this.incr = incr; this.decr = decr; } - push(value: T): void { + push(value: T): Node { this.length++; this.incr?.(); - const node = { value, prev: this._last, next: null }; + const node: Node = { 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): 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) { diff --git a/src/Job.ts b/src/Job.ts index c46d43d9..9f9f6986 100644 --- a/src/Job.ts +++ b/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 | 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 | 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,6 +154,7 @@ class Job { doRun(): Promise | undefined { if (this.retryCount === 0) { this._assertStatus("QUEUED"); + this._detachSignal(); this._states.next(this.options.id); } else { this._assertStatus("EXECUTING"); @@ -140,6 +162,38 @@ class Job { 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, ): Promise { - 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 | undefined { this._assertStatus("EXECUTING"); + this._detachSignal(); this._states.next(this.options.id); return this.Events.trigger("done", eventInfo); } diff --git a/src/Queues.ts b/src/Queues.ts index a8de56e9..4450e7ef 100644 --- a/src/Queues.ts +++ b/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 | undefined }; class Queues { Events: Events; @@ -46,7 +46,13 @@ class Queues { if (list == null) { throw new BottleneckError(`No queue for priority ${job.options.priority}`); } - list.push(job); + job._queueNode = list.push(job) as Node; + } + + /** Take `job` out of its priority list in O(1). */ + remove(job: J): boolean { + const node = job._queueNode as Node | undefined; + return node != null && (this._lists[job.options.priority]?.remove(node) ?? false); } queued(priority?: number): number { diff --git a/src/abort.ts b/src/abort.ts new file mode 100644 index 00000000..3afff3b7 --- /dev/null +++ b/src/abort.ts @@ -0,0 +1,51 @@ +import BottleneckError from "./BottleneckError"; + +const fallbackReasons = new WeakMap(); + +/** + * What an aborted job rejects with. `reason` is missing before Node 16.14; + * the fallback is one error per signal, so it compares like a real reason. + */ +export function abortReason(signal: AbortSignal): unknown { + if (signal.reason !== undefined) return signal.reason; + let fallback = fallbackReasons.get(signal); + if (fallback == null) { + fallback = new BottleneckError("This job was aborted", "ABORTED"); + fallbackReasons.set(signal, fallback); + } + return fallback; +} + +type Abortable = { _abort(): void }; +type Watchers = { jobs: Set; listener: () => void }; + +// One "abort" listener per signal, however many jobs share it: per-job +// listeners trip MaxListeners warnings on some runtimes, and removing one is +// O(listeners). +const watched = new WeakMap(); + +/** Call `job._abort()` when `signal` aborts, until unwatched. */ +export function watchAbort(signal: AbortSignal, job: Abortable): void { + let watchers = watched.get(signal); + if (watchers == null) { + const jobs = new Set(); + const listener = (): void => { + // Unregister first, so jobs unwatching themselves leave `jobs` intact. + watched.delete(signal); + for (const j of jobs) j._abort(); + }; + watchers = { jobs, listener }; + watched.set(signal, watchers); + signal.addEventListener("abort", listener, { once: true }); + } + watchers.jobs.add(job); +} + +export function unwatchAbort(signal: AbortSignal, job: Abortable): void { + const watchers = watched.get(signal); + if (watchers == null || !watchers.jobs.delete(job)) return; + if (watchers.jobs.size === 0) { + watched.delete(signal); + signal.removeEventListener("abort", watchers.listener); + } +} diff --git a/src/types.ts b/src/types.ts index 457cefcf..b86b2f81 100644 --- a/src/types.ts +++ b/src/types.ts @@ -89,6 +89,13 @@ export type JobOptions = { readonly expiration?: number | null; /** Optional identifier, helps with debug output. */ readonly id?: string | null; + /** + * Cancels the job if it hasn't started yet: it rejects with `signal.reason` + * (even when `rejectOnDrop` is false) and the task never runs. Aborting a + * job that has started does nothing; pass the same signal into the task to + * cancel in-flight work. + */ + readonly signal?: AbortSignal | null; }; /** Options accepted by `stop()`. */ @@ -256,6 +263,7 @@ export type ResolvedJobOptions = { weight: number; expiration: number | null; id: string; + signal?: AbortSignal | null; [key: string]: unknown; }; diff --git a/test/DLList.test.ts b/test/DLList.test.ts index a6bad4e4..4551a31a 100644 --- a/test/DLList.test.ts +++ b/test/DLList.test.ts @@ -152,4 +152,45 @@ describe("DLList", () => { expect(list.shift()).toStrictEqual(undefined); expect(list.first()).toStrictEqual(undefined); }); + + it("Should remove nodes from anywhere in O(1)", () => { + const queues = new FakeQueues(); + const list = new DLList(...queues.fns); + const [a, b, c, d] = [1, 2, 3, 4].map((v) => list.push(v)); + + expect(list.remove(b as never)).toBe(true); + expect(list.getArray()).toStrictEqual([1, 3, 4]); + expect(list.remove(a as never)).toBe(true); + expect(list.getArray()).toStrictEqual([3, 4]); + expect(list.first()).toStrictEqual(3); + expect(list.remove(d as never)).toBe(true); + expect(list.getArray()).toStrictEqual([3]); + expect(list.length).toStrictEqual(1); + expect(queues._length).toStrictEqual(1); + + expect(list.remove(c as never)).toBe(true); + expect(list.getArray()).toStrictEqual([]); + expect(list.first()).toStrictEqual(undefined); + expect(list.length).toStrictEqual(0); + expect(queues._length).toStrictEqual(0); + + list.push(5); + expect(list.getArray()).toStrictEqual([5]); + }); + + it("Should not remove a node that is no longer in the list", () => { + const queues = new FakeQueues(); + const list = new DLList(...queues.fns); + const a = list.push(1); + const b = list.push(2); + list.push(3); + + expect(list.shift()).toStrictEqual(1); + expect(list.remove(a)).toBe(false); + expect(list.remove(b)).toBe(true); + expect(list.remove(b)).toBe(false); + expect(list.getArray()).toStrictEqual([3]); + expect(list.length).toStrictEqual(1); + expect(queues._length).toStrictEqual(1); + }); }); diff --git a/test/abort.test.ts b/test/abort.test.ts new file mode 100644 index 00000000..7c6a3e2d --- /dev/null +++ b/test/abort.test.ts @@ -0,0 +1,283 @@ +import { describe, expect, vi } from "vitest"; +import { deferred, enqueued, test } from "./helpers/test-api"; + +/* + * `signal` cancels jobs that haven't started. Every case holds state with + * deferred() and barriers with enqueued() or a transition event, so it is + * deterministic under the local and redis projects alike. No fake clock: + * AbortSignal.timeout runs on Node's internal timers either way. + */ + +const zeroCounts = { RECEIVED: 0, QUEUED: 0, RUNNING: 0, EXECUTING: 0 }; + +describe("Abort signal", () => { + test("an already-aborted signal rejects without running the task", async ({ makeLimiter }) => { + const limiter = makeLimiter({ maxConcurrent: 1, rejectOnDrop: false }); + const controller = new AbortController(); + const reason = new Error("cancelled"); + controller.abort(reason); + const task = vi.fn<() => string>(() => "ran"); + + await expect(limiter.schedule({ signal: controller.signal }, task)).rejects.toBe(reason); + expect(task).not.toHaveBeenCalled(); + expect(limiter.counts()).toEqual(zeroCounts); + }); + + test("aborting mid-submit rejects and leaves the limiter usable", async ({ makeLimiter }) => { + const limiter = makeLimiter({ maxConcurrent: 1 }); + const controller = new AbortController(); + const task = vi.fn<() => string>(() => "ran"); + + // schedule() runs synchronously into `await __submit__`: still RECEIVED. + const victim = limiter.schedule({ signal: controller.signal }, task); + expect(limiter.counts().RECEIVED).toEqual(1); + controller.abort(); + + await expect(victim).rejects.toBe(controller.signal.reason); + await enqueued(limiter); + expect(task).not.toHaveBeenCalled(); + expect(limiter.counts()).toEqual(zeroCounts); + await expect(limiter.schedule(() => "next")).resolves.toEqual("next"); + }); + + test("aborting while waiting on the submit lock rejects", async ({ makeLimiter }) => { + const limiter = makeLimiter({ maxConcurrent: 1 }); + const controller = new AbortController(); + const task = vi.fn<() => string>(() => "ran"); + + const first = limiter.schedule(() => "first"); + const victim = limiter.schedule({ signal: controller.signal }, task); + controller.abort(); + + await expect(victim).rejects.toBe(controller.signal.reason); + await expect(first).resolves.toEqual("first"); + await enqueued(limiter); + expect(task).not.toHaveBeenCalled(); + expect(limiter.counts()).toEqual(zeroCounts); + }); + + test("aborting a queued job removes it and the next job still runs", async ({ makeLimiter }) => { + const limiter = makeLimiter({ maxConcurrent: 1 }); + const hold = deferred(); + const controller = new AbortController(); + const task = vi.fn<() => string>(() => "ran"); + + const blocker = limiter.schedule(() => hold.signal); + const victim = limiter.schedule({ id: "victim", signal: controller.signal }, task); + const next = limiter.schedule(() => "next"); + await enqueued(limiter); + expect(limiter.queued()).toEqual(2); + expect(limiter.jobStatus("victim")).toEqual("QUEUED"); + + let empties = 0; + limiter.on("empty", () => empties++); + controller.abort(); + + await expect(victim).rejects.toBe(controller.signal.reason); + expect(limiter.queued()).toEqual(1); + expect(limiter.jobStatus("victim")).toBeNull(); + expect(empties).toEqual(0); + + hold.release(); + await expect(blocker).resolves.toBeUndefined(); + await expect(next).resolves.toEqual("next"); + expect(task).not.toHaveBeenCalled(); + expect(empties).toEqual(1); + }); + + test("aborting the last queued job fires empty", async ({ makeLimiter }) => { + const limiter = makeLimiter({ maxConcurrent: 1 }); + const hold = deferred(); + const controller = new AbortController(); + + const blocker = limiter.schedule(() => hold.signal); + const victim = limiter.schedule({ signal: controller.signal }, () => "ran"); + await enqueued(limiter); + + let empties = 0; + limiter.on("empty", () => empties++); + controller.abort(); + expect(empties).toEqual(1); + expect(limiter.empty()).toBe(true); + + await expect(victim).rejects.toBe(controller.signal.reason); + hold.release(); + await expect(blocker).resolves.toBeUndefined(); + }); + + test("AbortSignal.timeout rejects with a TimeoutError", async ({ makeLimiter }) => { + const limiter = makeLimiter({ maxConcurrent: 1 }); + const hold = deferred(); + const task = vi.fn<() => string>(() => "ran"); + + const blocker = limiter.schedule(() => hold.signal); + const victim = limiter.schedule({ signal: AbortSignal.timeout(1) }, task); + + await expect(victim).rejects.toMatchObject({ name: "TimeoutError" }); + hold.release(); + await expect(blocker).resolves.toBeUndefined(); + expect(task).not.toHaveBeenCalled(); + }); + + test("aborting once the job has started does nothing", async ({ makeLimiter }) => { + const limiter = makeLimiter({ maxConcurrent: 1 }); + const running = new AbortController(); + const executing = new AbortController(); + + // RUNNING: abort inside the transition event, before the task's timer. + limiter.on("scheduled", (info) => { + if (info.options.id === "running") running.abort(); + }); + const p1 = limiter.schedule({ id: "running", signal: running.signal }, () => "ran 1"); + const p2 = limiter.schedule({ id: "executing", signal: executing.signal }, () => { + executing.abort(); + return "ran 2"; + }); + + await expect(p1).resolves.toEqual("ran 1"); + await expect(p2).resolves.toEqual("ran 2"); + expect(running.signal.aborted).toBe(true); + }); + + test("a shared signal holds one listener, removed once its jobs leave the queue", async ({ + makeLimiter, + }) => { + const limiter = makeLimiter({ maxConcurrent: 1 }); + const controller = new AbortController(); + const { signal } = controller; + const add = vi.spyOn(signal, "addEventListener"); + const remove = vi.spyOn(signal, "removeEventListener"); + const warnings: Error[] = []; + const onWarning = (warning: Error) => warnings.push(warning); + process.on("warning", onWarning); + + try { + const hold = deferred(); + const jobs = [ + limiter.schedule({ signal }, () => hold.signal), + ...Array.from({ length: 99 }, (_, i) => limiter.schedule({ signal }, () => i)), + ]; + await enqueued(limiter); + expect(limiter.queued()).toEqual(99); + expect(add).toHaveBeenCalledTimes(1); + expect(remove).not.toHaveBeenCalled(); + + hold.release(); + await Promise.all(jobs); + expect(add).toHaveBeenCalledTimes(1); + expect(remove).toHaveBeenCalledTimes(1); + expect(remove.mock.calls[0]?.[1]).toBe(add.mock.calls[0]?.[1]); + } finally { + process.off("warning", onWarning); + } + expect(warnings.filter((w) => w.name === "MaxListenersExceededWarning")).toEqual([]); + }); + + test("aborting during register frees the slot", async ({ makeLimiter }) => { + const limiter = makeLimiter({ maxConcurrent: 1 }); + const controller = new AbortController(); + const reason = new Error("cancelled"); + const task = vi.fn<() => string>(() => "ran"); + + // Debug listeners run synchronously, just before `await __register__`. + limiter.on("debug", (message) => { + if (message === "Draining victim") controller.abort(reason); + }); + const victim = limiter.schedule({ id: "victim", signal: controller.signal }, task); + + await expect(victim).rejects.toBe(reason); + // The victim's submit holds the lock through its drain, including the free. + await enqueued(limiter); + expect(await limiter.running()).toEqual(0); + expect(limiter.counts()).toEqual(zeroCounts); + await expect(limiter.schedule(() => "next")).resolves.toEqual("next"); + expect(task).not.toHaveBeenCalled(); + }); + + test("aborting in a chained limiter rejects the outer job as a cancellation", async ({ + makeLimiter, + }) => { + const outer = makeLimiter({ id: "abort-outer" }); + const inner = makeLimiter({ id: "abort-inner", maxConcurrent: 1 }); + outer.chain(inner); + const hold = deferred(); + const controller = new AbortController(); + const reason = new Error("cancelled"); + const stack = reason.stack; + const task = vi.fn<() => string>(() => "ran"); + const failed = vi.fn<() => number>(() => 0); + outer.on("failed", failed); + + const blocker = inner.schedule(() => hold.signal); + await enqueued(inner); + const queuedInInner = new Promise((resolve) => { + inner.on("queued", () => resolve()); + }); + const victim = outer.schedule({ signal: controller.signal }, task); + await queuedInInner; + controller.abort(reason); + + await expect(victim).rejects.toBe(reason); + expect(failed).not.toHaveBeenCalled(); + expect(reason.stack).toBe(stack); + expect(inner.queued()).toEqual(0); + + hold.release(); + await expect(blocker).resolves.toBeUndefined(); + expect(task).not.toHaveBeenCalled(); + }); + + test("a task that throws its own abort reason is not retried", async ({ makeLimiter }) => { + const limiter = makeLimiter({ maxConcurrent: 1 }); + const controller = new AbortController(); + const reason = new Error("cancelled"); + const stack = reason.stack; + const failed = vi.fn<() => number>(() => 0); + limiter.on("failed", failed); + + const task = vi.fn<() => never>(() => { + controller.abort(reason); + throw controller.signal.reason; + }); + + await expect(limiter.schedule({ signal: controller.signal }, task)).rejects.toBe(reason); + expect(task).toHaveBeenCalledTimes(1); + expect(failed).not.toHaveBeenCalled(); + expect(reason.stack).toBe(stack); + }); + + test("aborting during a retry delay skips the retry", async ({ makeLimiter }) => { + const limiter = makeLimiter({ maxConcurrent: 1 }); + const controller = new AbortController(); + const reason = new Error("cancelled"); + const failed = vi.fn<() => number>(() => 10); + limiter.on("failed", failed); + // "retry" fires just before the delay starts: the job is between attempts. + limiter.on("retry", () => controller.abort(reason)); + const task = vi.fn<() => never>(() => { + throw new Error("ignores the signal"); + }); + + await expect(limiter.schedule({ signal: controller.signal }, task)).rejects.toBe(reason); + expect(task).toHaveBeenCalledTimes(1); + expect(failed).toHaveBeenCalledTimes(1); + expect(await limiter.running()).toEqual(0); + await expect(limiter.schedule(() => "next")).resolves.toEqual("next"); + }); + + test("a failure after the signal aborted is not retried", async ({ makeLimiter }) => { + const limiter = makeLimiter({ maxConcurrent: 1 }); + const controller = new AbortController(); + const reason = new Error("cancelled"); + const failed = vi.fn<() => number>(() => 0); + limiter.on("failed", failed); + const task = vi.fn<() => never>(() => { + controller.abort(reason); + throw new Error("unrelated failure"); + }); + + await expect(limiter.schedule({ signal: controller.signal }, task)).rejects.toBe(reason); + expect(task).toHaveBeenCalledTimes(1); + expect(failed).not.toHaveBeenCalled(); + }); +}); diff --git a/test/error-codes.test.ts b/test/error-codes.test.ts index 24a9b262..fbf6b577 100644 --- a/test/error-codes.test.ts +++ b/test/error-codes.test.ts @@ -93,6 +93,24 @@ describe("BottleneckError codes", () => { expect(codeOf(err)).toEqual("STOPPED"); }); + test("aborts without a signal reason carry code ABORTED", async ({ makeLimiter }) => { + const limiter = makeLimiter({ maxConcurrent: 1 }); + // Node < 16.14 aborts without setting `signal.reason`. + const signal = { + aborted: true, + reason: undefined, + addEventListener() {}, + removeEventListener() {}, + } as unknown as AbortSignal; + const err = await capture( + limiter.schedule({ signal }, () => "never"), + "aborted job should have been rejected", + ); + expect(err).toBeInstanceOf(BottleneckError); + expect(codeOf(err)).toEqual("ABORTED"); + expect((err as Error).message).toEqual("This job was aborted"); + }); + test("constructor misuse carries codes", async () => { const invalidArgs = await capture( Promise.resolve().then(() => new Bottleneck("nope" as never)), diff --git a/test/smoke/light.test.ts b/test/smoke/light.test.ts index ba72d8af..85549035 100644 --- a/test/smoke/light.test.ts +++ b/test/smoke/light.test.ts @@ -14,6 +14,25 @@ describe("dist/light smoke", () => { await limiter.disconnect(false); }); + it("aborts a queued job", async () => { + const limiter = new Bottleneck({ maxConcurrent: 1 }); + let release: (() => void) | undefined; + const gate = new Promise((resolve) => { + release = resolve; + }); + const blocker = limiter.schedule(() => gate); + const controller = new AbortController(); + const victim = limiter.schedule({ signal: controller.signal }, () => "ran"); + await limiter._submitLock.schedule(() => Promise.resolve()); + expect(limiter.queued()).toBe(1); + controller.abort(); + await expect(victim).rejects.toBe(controller.signal.reason); + expect(limiter.queued()).toBe(0); + release?.(); + await blocker; + await limiter.disconnect(false); + }); + it("throws when clustering datastore is requested", () => { expect(() => { void new Bottleneck({ datastore: "redis" });