diff --git a/README.md b/README.md index f41833c..f76ce65 100644 --- a/README.md +++ b/README.md @@ -208,6 +208,49 @@ try { } ``` +### AsyncLock + +A promise-based mutex keyed by string, derived from the unmaintained +[async-lock](https://github.com/rogierschouten/async-lock) (MIT). Tasks with the +same key run one at a time in FIFO order; different keys run concurrently. + +Differences from `async-lock`: promises only (no callbacks, custom `Promise` +or domain support), and each task runs in the `AsyncLocalStorage` context of +its own `acquire` caller instead of the previous lock holder's one, so no +`AsyncResource.bind` is needed. A task that outlives `maxExecutionTime` can no +longer release the lock a second time, and queued tasks that time out are +removed from the queue. + +```js +import {AsyncLock} from 'asyncbox'; + +const lock = new AsyncLock({maxPending: 10}); + +const result = await lock.acquire('key', async () => { + // runs exclusively for 'key' + return 42; +}); + +// several keys are held together; they are acquired in sorted order, so overlapping calls cannot deadlock +await lock.acquire(['key1', 'key2'], () => {}); + +// per-call options: timeout, maxOccupationTime, maxExecutionTime, maxPending, skipQueue +await lock.acquire('key', () => {}, {timeout: 5000, skipQueue: true}); + +lock.getQueueLength('key'); // tasks waiting for the key (the running one is not counted) +lock.isBusy('key'); // running or pending tasks for the key +lock.isBusy(); // any key +``` + +Options can be passed to the constructor and overridden per `acquire` call. +Times are in ms, up to 2147483647, and `0` means no limit. + +- `timeout` (default `0`) - max wait in the queue (for all the keys of a multi-key call together) before the task is rejected and dropped +- `maxOccupationTime` (default `0`) - max time from `acquire` to completion, queue wait included; one budget is shared by all keys of a multi-key call +- `maxExecutionTime` (default `0`) - max time the task may run while holding the lock; the lock is released when exceeded (for a multi-key call, it counts from when all keys are held and releases all of them) +- `maxPending` (default `1000`) - max tasks waiting per key; further ones are rejected right away, so `0` rejects every task that would have to wait +- `skipQueue` (default `false`, per call only) - put the task at the front of the queue + ### Run the tests ``` diff --git a/lib/asyncbox.ts b/lib/asyncbox.ts deleted file mode 100644 index 61fb840..0000000 --- a/lib/asyncbox.ts +++ /dev/null @@ -1,382 +0,0 @@ -import type {limitFunction} from 'p-limit'; - -import type { - CancellablePromise, - LongSleepOptions, - MapFilterOptions, - SleepArg, - SleepOptions, - WaitForConditionOptions, -} from './types.js'; - -const LONG_SLEEP_THRESHOLD = 5000; // anything over 5000ms will turn into a spin - -type PLimitModule = {limitFunction: typeof limitFunction}; - -// p-limit is ESM-only. tsc downlevels a plain `import()` in CommonJS output into a -// require()-in-a-microtask, which still hits Node's require(esm) interop and can race with a -// concurrent dynamic import() of the same module elsewhere in the process -// (ERR_REQUIRE_ESM_RACE_CONDITION). Route through `new Function` so tsc can't see or rewrite the -// import() call, keeping it a genuine native dynamic import in both the ESM and CJS builds. -const dynamicImport = new Function('specifier', 'return import(specifier)') as ( - specifier: string, -) => Promise; - -/** Error thrown by {@link withTimeout} when the deadline is exceeded. */ -export class TimeoutError extends Error { - constructor(message?: string) { - super(message ?? 'Operation timed out'); - this.name = 'TimeoutError'; - } -} - -/** Thrown when a promise is cancelled via `cancel`. */ -export class PromiseCancellationError extends Error { - constructor(message: string = 'Promise cancelled') { - super(message); - this.name = 'PromiseCancellationError'; - } -} - -/** - * An async/await version of `setTimeout`. The returned promise has a `cancel()` method that clears - * the timer and usually rejects with {@link PromiseCancellationError} unless you use the object form with - * `cancelError: null` to resolve on cancel. - * - * @param ms - Milliseconds to wait - * @returns A thenable you can `await`; call `.cancel()` to abort early. - */ -export function sleep(ms: number): CancellablePromise; -/** - * Object form: `ms` plus optional `cancelError` (non-empty string or `Error`, `null` to resolve on - * cancel, or omitted / `undefined` / `''` for default {@link PromiseCancellationError}). Any non-array - * object with a finite `ms` is accepted at runtime (including class instances), matching structural typing. - * - * @param options - Duration and optional cancellation behavior - * @returns A thenable you can `await`; call `.cancel()` to abort early. - */ -export function sleep(options: SleepOptions): CancellablePromise; -export function sleep(arg: SleepArg): CancellablePromise { - const {ms, cancelError} = parseSleepArg(arg); - let timeoutId: NodeJS.Timeout | undefined; - let resolveFn: ((value: void) => void) | undefined; - let rejectFn: ((error: Error) => void) | undefined; - - const promise = new Promise((resolve, reject) => { - resolveFn = resolve; - rejectFn = reject; - timeoutId = setTimeout(resolve, ms); - }) as CancellablePromise; - - promise.cancel = () => { - if (timeoutId) { - clearTimeout(timeoutId); - timeoutId = undefined; - } - if (cancelError === null) { - resolveFn?.(undefined); - } else { - let err: Error; - if (typeof cancelError === 'string' && cancelError) { - err = new PromiseCancellationError(cancelError); - } else if (cancelError instanceof Error) { - err = cancelError; - } else { - err = new PromiseCancellationError(); - } - rejectFn?.(err); - } - resolveFn = undefined; - rejectFn = undefined; - }; - - return promise; -} - -/** - * Resolves with `promise` or rejects if it does not settle before `timeoutMs`. - * - * If the deadline passes first: with no third argument (or a falsy non-error value), rejects - * with {@link TimeoutError} whose message includes `timeoutMs`; with a non-empty string, - * rejects with {@link TimeoutError} using that message; with an `Error` instance, rejects - * with that same instance. - * - * @param promise - Promise to race against the timeout. - * @param timeoutMs - Maximum time in milliseconds before rejecting if `promise` has not settled. - * @param messageOrError - Optional override for the timeout rejection: custom {@link TimeoutError} - * message when the string is non-empty, or an existing `Error` to reject with verbatim. - * Omitted, `undefined`, or other falsy non-error values use the default timeout message (includes `timeoutMs`). - */ -export async function withTimeout( - promise: Promise, - timeoutMs: number, - messageOrError?: string | Error, -): Promise { - let timer: NodeJS.Timeout | null = null; - try { - return await Promise.race([ - promise, - new Promise((_resolve, reject) => { - timer = setTimeout(() => { - if (typeof messageOrError === 'string' && messageOrError) { - reject(new TimeoutError(messageOrError)); - } else if (!messageOrError) { - reject(new TimeoutError(`Operation timed out after ${timeoutMs}ms`)); - } else { - reject(messageOrError); - } - }, timeoutMs); - }), - ]); - } finally { - if (timer) { - clearTimeout(timer); - } - } -} - -/** - * Sometimes `Promise.delay` or `setTimeout` are inaccurate for large wait - * times. To safely wait for these long times (e.g. in the 5+ minute range), you - * can use `longSleep`. - * - * You can also pass a `progressCb` option which is a callback function that - * receives an object with the properties `elapsedMs`, `timeLeft`, and - * `progress`. This will be called on every wait interval so you can do your - * wait logging or whatever. - * @param ms - The number of milliseconds to wait - * @param options - Options for controlling the long sleep behavior - */ -export async function longSleep( - ms: number, - {thresholdMs = LONG_SLEEP_THRESHOLD, intervalMs = 1000, progressCb = null}: LongSleepOptions = {}, -): Promise { - if (ms < thresholdMs) { - return await sleep(ms); - } - const endAt = Date.now() + ms; - let timeLeft: number; - let elapsedMs = 0; - do { - const pre = Date.now(); - await sleep(intervalMs); - const post = Date.now(); - timeLeft = endAt - post; - elapsedMs = elapsedMs + (post - pre); - if (typeof progressCb === 'function') { - progressCb({elapsedMs, timeLeft, progress: elapsedMs / ms}); - } - } while (timeLeft > 0); -} - -/** - * An async/await way of running a method until it doesn't throw an error - * @param times - The maximum number of times to retry the function - * @param fn - The async function to retry - * @param args - Arguments to pass to the function - */ -export async function retry( - times: number, - fn: (...args: any[]) => Promise, - ...args: any[] -): Promise { - let tries = 0; - let done = false; - let res: T | null = null; - while (!done && tries < times) { - tries++; - try { - res = await fn(...args); - done = true; - } catch (err) { - if (tries >= times) { - throw err; - } - } - } - return res; -} - -/** - * You can also use `retryInterval` to add a sleep in between retries. This can - * be useful if you want to throttle how fast we retry. - * @param times - The maximum number of times to retry the function - * @param sleepMs - The number of milliseconds to wait between retries - * @param fn - The async function to retry - * @param args - Arguments to pass to the function - */ -export async function retryInterval( - times: number, - sleepMs: number, - fn: (...args: any[]) => Promise, - ...args: any[] -): Promise { - let count = 0; - const wrapped = async (): Promise => { - count++; - let res: T; - try { - res = await fn(...args); - } catch (e) { - // do not pause when finished the last retry - if (count !== times) { - await sleep(sleepMs); - } - throw e; - } - return res; - }; - return await retry(times, wrapped); -} - -/** - * Similar to `Array.prototype.map`; runs in parallel, serial, or with custom concurrency pool - * @param coll - The collection to map over - * @param mapper - The function to apply to each element - * @param options - Options for controlling parallelism (default: true - fully parallel) - */ -export async function asyncmap( - coll: T[], - mapper: (value: T) => R | Promise, - options: MapFilterOptions = true, -): Promise { - if (options === null) { - throw new Error('Options cannot be null'); - } - // limitFunction requires the mapper to always return a promise - const mapperAsync = async (value: T): Promise => mapper(value); - if (options === false) { - return coll.reduce>( - async (acc, item) => [...(await acc), await mapperAsync(item)], - Promise.resolve([]), - ); - } - const adjustedMapper = - options === true ? mapperAsync : (await getLimitFunction())(mapperAsync, {concurrency: options.concurrency}); - return Promise.all(coll.map(adjustedMapper)); -} - -/** - * Similar to `Array.prototype.filter`; runs in parallel, serial, or with custom concurrency pool - * @param coll - The collection to filter - * @param filter - The function to test each element - * @param options - Options for controlling parallelism (default: true - fully parallel) - */ -export async function asyncfilter( - coll: T[], - filter: (value: T) => boolean | Promise, - options: MapFilterOptions = true, -): Promise { - if (options === null) { - throw new Error('Options cannot be null'); - } - // limitFunction requires the filter to always return a promise - const filterAsync = async (value: T): Promise => filter(value); - if (options === false) { - return coll.reduce>(async (accP, item) => { - const acc = await accP; - if (await filterAsync(item)) { - acc.push(item); - } - return acc; - }, Promise.resolve([])); - } - const adjustedFilter = - options === true ? filterAsync : (await getLimitFunction())(filterAsync, {concurrency: options.concurrency}); - const bools = await Promise.all(coll.map(adjustedFilter)); - return coll.reduce((acc, item, i) => { - if (bools[i]) { - acc.push(item); - } - return acc; - }, []); -} - -/** - * Takes a condition (a function returning a boolean or boolean promise), and - * waits until the condition is true. - * - * Throws a `/Condition unmet/` error if the condition has not been satisfied - * within the allocated time, unless an error is provided in the options, as the - * `error` property, which is either thrown itself, or used as the message. - * - * The condition result is returned if it is not falsy. If the condition throws an - * error then this exception will be immediately passed through. - * - * The default options are: `{ waitMs: 5000, intervalMs: 500 }` - * @param condFn - The condition function to evaluate - * @param options - Options for controlling the wait behavior - */ -export async function waitForCondition( - condFn: () => Promise | T, - options: WaitForConditionOptions = {}, -): Promise { - const opts: WaitForConditionOptions & {waitMs: number; intervalMs: number} = { - ...options, - waitMs: typeof options.waitMs === 'number' ? options.waitMs : 5000, - intervalMs: typeof options.intervalMs === 'number' ? options.intervalMs : 500, - }; - const debug = opts.logger ? opts.logger.debug.bind(opts.logger) : () => undefined; - const error = opts.error; - const begunAt = Date.now(); - const endAt = begunAt + opts.waitMs; - const spin = async function spin(): Promise { - const result = await condFn(); - if (result) { - return result; - } - const now = Date.now(); - const waited = now - begunAt; - const remainingTime = endAt - now; - if (now < endAt) { - debug(`Waited for ${waited} ms so far`); - await sleep(Math.min(opts.intervalMs, remainingTime)); - return await spin(); - } - // if there is an error option, it is either a string message or an error itself - if (error) { - throw typeof error === 'string' ? new Error(error) : error; - } - throw new Error(`Condition unmet after ${waited} ms. Timing out.`); - }; - return await spin(); -} - -// Re-export types -export type { - CancellablePromise, - Progress, - ProgressCallback, - LongSleepOptions, - SleepArg, - SleepOptions, - WaitForConditionOptions, -} from './types.js'; - -/** Non-array object values (including class instances); excludes `null` and arrays. */ -function isSleepArgObject(value: unknown): value is Record { - return value !== null && typeof value === 'object' && !Array.isArray(value); -} - -function parseSleepArg(arg: SleepArg): {ms: number; cancelError?: string | Error | null} { - if (typeof arg === 'number') { - if (!Number.isFinite(arg)) { - throw new TypeError('sleep: expected a finite number or an object with ms'); - } - return {ms: arg}; - } - if (isSleepArgObject(arg)) { - const ms = arg.ms; - if (typeof ms !== 'number' || !Number.isFinite(ms)) { - throw new TypeError('sleep: options.ms must be a finite number'); - } - return {ms, cancelError: arg.cancelError as string | Error | null | undefined}; - } - throw new TypeError('sleep: expected a finite number or an object with ms'); -} - -let limitFunctionPromise: Promise | undefined; -async function getLimitFunction() { - limitFunctionPromise ??= dynamicImport('p-limit'); - return (await limitFunctionPromise).limitFunction; -} diff --git a/lib/asyncfilter.ts b/lib/asyncfilter.ts new file mode 100644 index 0000000..efdf4c2 --- /dev/null +++ b/lib/asyncfilter.ts @@ -0,0 +1,38 @@ +import {getLimitFunction} from './limit-function.js'; +import type {MapFilterOptions} from './types.js'; + +/** + * Similar to `Array.prototype.filter`; runs in parallel, serial, or with custom concurrency pool + * @param coll - The collection to filter + * @param filter - The function to test each element + * @param options - Options for controlling parallelism (default: true - fully parallel) + */ +export async function asyncfilter( + coll: T[], + filter: (value: T) => boolean | Promise, + options: MapFilterOptions = true, +): Promise { + if (options === null) { + throw new Error('Options cannot be null'); + } + // limitFunction requires the filter to always return a promise + const filterAsync = async (value: T): Promise => filter(value); + if (options === false) { + return coll.reduce>(async (accP, item) => { + const acc = await accP; + if (await filterAsync(item)) { + acc.push(item); + } + return acc; + }, Promise.resolve([])); + } + const adjustedFilter = + options === true ? filterAsync : (await getLimitFunction())(filterAsync, {concurrency: options.concurrency}); + const bools = await Promise.all(coll.map(adjustedFilter)); + return coll.reduce((acc, item, i) => { + if (bools[i]) { + acc.push(item); + } + return acc; + }, []); +} diff --git a/lib/asyncmap.ts b/lib/asyncmap.ts new file mode 100644 index 0000000..f0d66fb --- /dev/null +++ b/lib/asyncmap.ts @@ -0,0 +1,29 @@ +import {getLimitFunction} from './limit-function.js'; +import type {MapFilterOptions} from './types.js'; + +/** + * Similar to `Array.prototype.map`; runs in parallel, serial, or with custom concurrency pool + * @param coll - The collection to map over + * @param mapper - The function to apply to each element + * @param options - Options for controlling parallelism (default: true - fully parallel) + */ +export async function asyncmap( + coll: T[], + mapper: (value: T) => R | Promise, + options: MapFilterOptions = true, +): Promise { + if (options === null) { + throw new Error('Options cannot be null'); + } + // limitFunction requires the mapper to always return a promise + const mapperAsync = async (value: T): Promise => mapper(value); + if (options === false) { + return coll.reduce>( + async (acc, item) => [...(await acc), await mapperAsync(item)], + Promise.resolve([]), + ); + } + const adjustedMapper = + options === true ? mapperAsync : (await getLimitFunction())(mapperAsync, {concurrency: options.concurrency}); + return Promise.all(coll.map(adjustedMapper)); +} diff --git a/lib/errors.ts b/lib/errors.ts new file mode 100644 index 0000000..fd3f864 --- /dev/null +++ b/lib/errors.ts @@ -0,0 +1,15 @@ +/** Error thrown by {@link withTimeout} when the deadline is exceeded. */ +export class TimeoutError extends Error { + constructor(message?: string) { + super(message ?? 'Operation timed out'); + this.name = 'TimeoutError'; + } +} + +/** Thrown when a promise is cancelled via `cancel`. */ +export class PromiseCancellationError extends Error { + constructor(message: string = 'Promise cancelled') { + super(message); + this.name = 'PromiseCancellationError'; + } +} diff --git a/lib/index.ts b/lib/index.ts new file mode 100644 index 0000000..d24c886 --- /dev/null +++ b/lib/index.ts @@ -0,0 +1,21 @@ +export {TimeoutError, PromiseCancellationError} from './errors.js'; +export {sleep} from './sleep.js'; +export {withTimeout} from './with-timeout.js'; +export {longSleep} from './long-sleep.js'; +export {retry, retryInterval} from './retry.js'; +export {asyncmap} from './asyncmap.js'; +export {asyncfilter} from './asyncfilter.js'; +export {waitForCondition} from './wait-for-condition.js'; +export {AsyncLock} from './lock.js'; + +export type {AsyncLockOptions, AcquireOptions} from './lock.js'; +export type { + CancellablePromise, + Progress, + ProgressCallback, + LongSleepOptions, + MapFilterOptions, + SleepArg, + SleepOptions, + WaitForConditionOptions, +} from './types.js'; diff --git a/lib/limit-function.ts b/lib/limit-function.ts new file mode 100644 index 0000000..7a0940c --- /dev/null +++ b/lib/limit-function.ts @@ -0,0 +1,18 @@ +import type {limitFunction} from 'p-limit'; + +type PLimitModule = {limitFunction: typeof limitFunction}; + +// p-limit is ESM-only. tsc downlevels a plain `import()` in CommonJS output into a +// require()-in-a-microtask, which still hits Node's require(esm) interop and can race with a +// concurrent dynamic import() of the same module elsewhere in the process +// (ERR_REQUIRE_ESM_RACE_CONDITION). Route through `new Function` so tsc can't see or rewrite the +// import() call, keeping it a genuine native dynamic import in both the ESM and CJS builds. +const dynamicImport = new Function('specifier', 'return import(specifier)') as ( + specifier: string, +) => Promise; + +let limitFunctionPromise: Promise | undefined; +export async function getLimitFunction() { + limitFunctionPromise ??= dynamicImport('p-limit'); + return (await limitFunctionPromise).limitFunction; +} diff --git a/lib/lock.ts b/lib/lock.ts new file mode 100644 index 0000000..b0da7a6 --- /dev/null +++ b/lib/lock.ts @@ -0,0 +1,360 @@ +import {AsyncResource} from 'node:async_hooks'; + +const DEFAULT_MAX_PENDING = 1000; +// setTimeout fires after 1 ms for anything above this +const MAX_TIMER_MS = 2 ** 31 - 1; + +/** Options shared by the {@link AsyncLock} constructor and {@link AsyncLock.acquire}. */ +export interface AsyncLockOptions { + /** + * Max ms a task may wait in the queue before failing. `0` means never. + * @default 0 + */ + timeout?: number; + /** + * Max ms between calling `acquire` and the task finishing (queue wait included). `0` means never. + * @default 0 + */ + maxOccupationTime?: number; + /** + * Max ms the task may run once it holds the lock; the lock is then released. `0` means never. + * @default 0 + */ + maxExecutionTime?: number; + /** + * Max number of tasks waiting per key; further ones are rejected right away. `0` rejects every + * task that would have to wait. + * @default 1000 + */ + maxPending?: number; +} + +/** Per-call options of {@link AsyncLock.acquire}; override the constructor ones. */ +export interface AcquireOptions extends AsyncLockOptions { + /** + * Put the task at the front of the queue instead of the back. + * @default false + */ + skipQueue?: boolean; +} + +type Task = () => T | PromiseLike; +type Outcome = {value: T} | {error: unknown}; +type Queues = Map void>>; +type ResolvedOptions = Required & {skipQueue?: boolean}; +type ClaimOptions = Pick; + +/** + * Promise-based mutex keyed by string. Tasks with the same key run one at a time, in FIFO order. + * + * Each task runs in the async context (`AsyncLocalStorage`) of the code that called + * {@link AsyncLock.acquire}, not in the context of the task that held the lock before it. + */ +export class AsyncLock { + private readonly queues: Queues = new Map(); + private readonly defaults: Required; + + constructor(opts: AsyncLockOptions = {}) { + validateOptions(opts); + this.defaults = { + timeout: 0, + maxOccupationTime: 0, + maxExecutionTime: 0, + maxPending: DEFAULT_MAX_PENDING, + ...stripUndefined(opts), + }; + } + + /** + * Run `fn` once the lock for `key` is free and release the lock when it settles. + * + * Multiple keys are held together. They are acquired one by one in sorted order, whatever order + * they are given in, so that overlapping calls cannot deadlock; duplicates are ignored. Keys that + * are already held stay held while a later one is awaited. `timeout` and `maxOccupationTime` are + * budgets for the whole call, while `maxExecutionTime` counts from the moment all the keys are + * held and releases all of them. + * + * @param key - Resource key, or a non-empty array of keys + * @param fn - Task to run while holding the lock; may be sync or async + * @param opts - Per-call options + * @returns The result of `fn` + * @throws {TypeError} Synchronously, if an argument is invalid + */ + acquire(key: string | readonly string[], fn: Task, opts: AcquireOptions = {}): Promise { + if (typeof fn !== 'function') { + throw new TypeError('You must pass a function to execute'); + } + if ( + typeof key !== 'string' && + !(Array.isArray(key) && key.length > 0 && [...key].every((k) => typeof k === 'string')) + ) { + throw new TypeError('The key must be a string or a non-empty array of strings'); + } + validateOptions(opts); + // A canonical order keeps overlapping multi-key calls, e.g. ['a', 'b'] and ['b', 'a'], from deadlocking + const keys = typeof key === 'string' ? [key] : [...new Set(key)].sort(); + const acquisition = new Acquisition(this.queues, keys, fn, {...this.defaults, ...stripUndefined(opts)}); + acquisition.enqueue(); + return acquisition.promise; + } + + /** + * Number of tasks waiting for `key`, not counting the running one. + * + * @param key - Resource key + */ + getQueueLength(key: string): number { + return this.queues.get(key)?.length ?? 0; + } + + /** + * Whether `key` has a running or pending task. Without a key, whether any key does. + * + * @param key - Resource key + */ + isBusy(key?: string): boolean { + return key === undefined ? this.queues.size > 0 : this.queues.has(key); + } +} + +/** The place of one acquisition in the queue of one key; once granted, it holds the key. */ +class KeyClaim { + private holding = false; + private dropped = false; + private queueTimer?: NodeJS.Timeout; + + constructor( + private readonly queues: Queues, + private readonly key: string, + private readonly opts: ClaimOptions, + private readonly onGranted: () => void, + private readonly onDenied: (error: Error) => void, + ) {} + + /** + * Take the key right away if it is free, otherwise wait in its queue. + * + * @returns Whether the key is held. If not, `onGranted` is called once it is, unless the claim + * is denied. + */ + join(): boolean { + const {timeout, maxPending, skipQueue} = this.opts; + const queue = this.queues.get(this.key); + if (!queue) { + this.queues.set(this.key, []); + this.holding = true; + return true; + } + if (queue.length >= maxPending) { + this.onDenied(new Error(`Too many pending tasks in queue ${this.key}`)); + } else { + if (skipQueue) { + queue.unshift(this.grant); + } else { + queue.push(this.grant); + } + if (timeout) { + this.queueTimer = setTimeout( + () => this.onDenied(new Error(`async-lock timed out in queue ${this.key}`)), + timeout, + ); + } + } + return false; + } + + /** + * Leave the queue, or release the key if it is held. Idempotent. + * + * @returns The handover of the key to its next waiter. The caller must run it only after it + * has released all of its keys, so that no waiter starts while a batch is partially released. + */ + drop(): (() => void) | undefined { + if (this.dropped) { + return undefined; + } + this.dropped = true; + clearTimeout(this.queueTimer); + if (this.holding) { + return this.releaseKey(); + } + const queue = this.queues.get(this.key); + const idx = queue?.indexOf(this.grant) ?? -1; + if (idx >= 0) { + queue?.splice(idx, 1); + } + return undefined; + } + + private releaseKey(): (() => void) | undefined { + const next = this.queues.get(this.key)?.shift(); + if (!next) { + this.queues.delete(this.key); + } + return next; + } + + // An arrow function, because this exact reference sits in the queue and is removed by drop() + private readonly grant = (): void => { + clearTimeout(this.queueTimer); + this.holding = true; + if (this.dropped) { + // Dropped while the key was being handed over to it: pass the key on instead of using it + this.releaseKey()?.(); + return; + } + this.onGranted(); + }; +} + +/** + * One `acquire` call: its promise, its timers and its claims on all of its keys. The keys are + * claimed one by one, the task runs once all of them are held and they are released together. + */ +class Acquisition { + readonly promise: Promise; + private resolve!: (value: T) => void; + private reject!: (error: unknown) => void; + private readonly task: Task; + private readonly claims: KeyClaim[] = []; + private running = false; + private settled = false; + private queueDeadline = 0; + private occupationTimer?: NodeJS.Timeout; + private executionTimer?: NodeJS.Timeout; + + constructor( + private readonly queues: Queues, + private readonly keys: string[], + task: Task, + private readonly opts: ResolvedOptions, + ) { + // Queued tasks are started from the previous task's promise chain; bind to keep our context + this.task = AsyncResource.bind(task); + this.promise = new Promise((resolve, reject) => { + this.resolve = resolve; + this.reject = reject; + }); + } + + /** Start the task right away if all the keys are free, otherwise wait for them. */ + enqueue(): void { + const {timeout, maxOccupationTime} = this.opts; + if (timeout) { + this.queueDeadline = performance.now() + timeout; + } + if (maxOccupationTime) { + this.occupationTimer = setTimeout(() => this.expire(), maxOccupationTime); + } + this.claimKeys(0); + } + + // Free keys are claimed in a loop, not recursively, so that a large batch cannot overflow the stack + private claimKeys(from: number): void { + const {maxPending, skipQueue} = this.opts; + try { + for (let index = from; index < this.keys.length; index++) { + // The queue timeout is a budget for waiting for all the keys, not for each one + const timeout = this.queueDeadline ? Math.max(this.queueDeadline - performance.now(), 1) : 0; + const claim = new KeyClaim( + this.queues, + this.keys[index], + {timeout, maxPending, skipQueue}, + () => this.claimKeys(index + 1), + (error) => this.abort(error), + ); + this.claims.push(claim); + if (!claim.join()) { + return; + } + } + this.execute(); + } catch (error) { + this.abort(error); + } + } + + private execute(): void { + const {maxExecutionTime} = this.opts; + this.running = true; + if (maxExecutionTime) { + this.executionTimer = setTimeout( + () => this.abort(new Error(`Maximum execution time is exceeded ${this.keys.join(', ')}`)), + maxExecutionTime, + ); + } + new Promise((resolve) => resolve(this.task())).then( + (value) => this.finish({value}), + (error) => this.finish({error}), + ); + } + + private finish(outcome: Outcome): void { + this.settle(outcome); + this.release(); + } + + // Reject and release all the keys; a running task is left to finish on its own + private abort(error: unknown): void { + this.settle({error}); + this.release(); + } + + // A running task keeps its keys until it finishes, an unstarted one gives them up + private expire(): void { + const keys = this.keys.join(', '); + if (this.running) { + this.settle({error: new Error(`Maximum occupation time is exceeded while executing ${keys}`)}); + } else { + this.abort(new Error(`Maximum occupation time is exceeded in queue ${keys}`)); + } + } + + private settle(outcome: Outcome): void { + if (this.settled) { + return; + } + this.settled = true; + clearTimeout(this.occupationTimer); + if ('error' in outcome) { + this.reject(outcome.error); + } else { + this.resolve(outcome.value); + } + } + + // Idempotent, so a task that outlives its execution timeout cannot release the keys twice + private release(): void { + clearTimeout(this.executionTimer); + const handovers = this.claims + .splice(0) + .reverse() + .map((claim) => claim.drop()); + for (const handover of handovers) { + handover?.(); + } + } +} + +function validateOptions(opts: AcquireOptions): void { + if (opts === null || typeof opts !== 'object') { + throw new TypeError('The options must be an object'); + } + for (const name of ['timeout', 'maxOccupationTime', 'maxExecutionTime'] as const) { + const value = opts[name]; + if (value !== undefined && !(Number.isFinite(value) && value >= 0 && value <= MAX_TIMER_MS)) { + throw new TypeError(`${name} must be a number of milliseconds between 0 and ${MAX_TIMER_MS}`); + } + } + if (opts.skipQueue !== undefined && typeof opts.skipQueue !== 'boolean') { + throw new TypeError('skipQueue must be a boolean'); + } + const {maxPending} = opts; + if (maxPending !== undefined && maxPending !== Infinity && !(Number.isInteger(maxPending) && maxPending >= 0)) { + throw new TypeError('maxPending must be a non-negative integer or Infinity'); + } +} + +function stripUndefined(obj: T): Partial { + return Object.fromEntries(Object.entries(obj).filter(([, v]) => v !== undefined)) as Partial; +} diff --git a/lib/long-sleep.ts b/lib/long-sleep.ts new file mode 100644 index 0000000..ce3bc51 --- /dev/null +++ b/lib/long-sleep.ts @@ -0,0 +1,38 @@ +import {sleep} from './sleep.js'; +import type {LongSleepOptions} from './types.js'; + +const LONG_SLEEP_THRESHOLD = 5000; // anything over 5000ms will turn into a spin + +/** + * Sometimes `Promise.delay` or `setTimeout` are inaccurate for large wait + * times. To safely wait for these long times (e.g. in the 5+ minute range), you + * can use `longSleep`. + * + * You can also pass a `progressCb` option which is a callback function that + * receives an object with the properties `elapsedMs`, `timeLeft`, and + * `progress`. This will be called on every wait interval so you can do your + * wait logging or whatever. + * @param ms - The number of milliseconds to wait + * @param options - Options for controlling the long sleep behavior + */ +export async function longSleep( + ms: number, + {thresholdMs = LONG_SLEEP_THRESHOLD, intervalMs = 1000, progressCb = null}: LongSleepOptions = {}, +): Promise { + if (ms < thresholdMs) { + return await sleep(ms); + } + const endAt = performance.now() + ms; + let timeLeft: number; + let elapsedMs = 0; + do { + const pre = performance.now(); + await sleep(intervalMs); + const post = performance.now(); + timeLeft = endAt - post; + elapsedMs = elapsedMs + (post - pre); + if (typeof progressCb === 'function') { + progressCb({elapsedMs: Math.round(elapsedMs), timeLeft: Math.round(timeLeft), progress: elapsedMs / ms}); + } + } while (timeLeft > 0); +} diff --git a/lib/retry.ts b/lib/retry.ts new file mode 100644 index 0000000..023edb7 --- /dev/null +++ b/lib/retry.ts @@ -0,0 +1,61 @@ +import {sleep} from './sleep.js'; + +/** + * An async/await way of running a method until it doesn't throw an error + * @param times - The maximum number of times to retry the function + * @param fn - The async function to retry + * @param args - Arguments to pass to the function + */ +export async function retry( + times: number, + fn: (...args: any[]) => Promise, + ...args: any[] +): Promise { + let tries = 0; + let done = false; + let res: T | null = null; + while (!done && tries < times) { + tries++; + try { + res = await fn(...args); + done = true; + } catch (err) { + if (tries >= times) { + throw err; + } + } + } + return res; +} + +/** + * You can also use `retryInterval` to add a sleep in between retries. This can + * be useful if you want to throttle how fast we retry. + * @param times - The maximum number of times to retry the function + * @param sleepMs - The number of milliseconds to wait between retries + * @param fn - The async function to retry + * @param args - Arguments to pass to the function + */ +export async function retryInterval( + times: number, + sleepMs: number, + fn: (...args: any[]) => Promise, + ...args: any[] +): Promise { + let count = 0; + const wrapped = async (): Promise => { + count++; + let res: T; + try { + res = await fn(...args); + } catch (e) { + // do not pause when finished the last retry + if (count !== times) { + await sleep(sleepMs); + } + throw e; + } + return res; + }; + return await retry(times, wrapped); +} diff --git a/lib/sleep.ts b/lib/sleep.ts new file mode 100644 index 0000000..41ab9fd --- /dev/null +++ b/lib/sleep.ts @@ -0,0 +1,79 @@ +import {PromiseCancellationError} from './errors.js'; +import type {CancellablePromise, SleepArg, SleepOptions} from './types.js'; + +/** + * An async/await version of `setTimeout`. The returned promise has a `cancel()` method that clears + * the timer and usually rejects with {@link PromiseCancellationError} unless you use the object form with + * `cancelError: null` to resolve on cancel. + * + * @param ms - Milliseconds to wait + * @returns A thenable you can `await`; call `.cancel()` to abort early. + */ +export function sleep(ms: number): CancellablePromise; +/** + * Object form: `ms` plus optional `cancelError` (non-empty string or `Error`, `null` to resolve on + * cancel, or omitted / `undefined` / `''` for default {@link PromiseCancellationError}). Any non-array + * object with a finite `ms` is accepted at runtime (including class instances), matching structural typing. + * + * @param options - Duration and optional cancellation behavior + * @returns A thenable you can `await`; call `.cancel()` to abort early. + */ +export function sleep(options: SleepOptions): CancellablePromise; +export function sleep(arg: SleepArg): CancellablePromise { + const {ms, cancelError} = parseSleepArg(arg); + let timeoutId: NodeJS.Timeout | undefined; + let resolveFn: ((value: void) => void) | undefined; + let rejectFn: ((error: Error) => void) | undefined; + + const promise = new Promise((resolve, reject) => { + resolveFn = resolve; + rejectFn = reject; + timeoutId = setTimeout(resolve, ms); + }) as CancellablePromise; + + promise.cancel = () => { + if (timeoutId) { + clearTimeout(timeoutId); + timeoutId = undefined; + } + if (cancelError === null) { + resolveFn?.(undefined); + } else { + let err: Error; + if (typeof cancelError === 'string' && cancelError) { + err = new PromiseCancellationError(cancelError); + } else if (cancelError instanceof Error) { + err = cancelError; + } else { + err = new PromiseCancellationError(); + } + rejectFn?.(err); + } + resolveFn = undefined; + rejectFn = undefined; + }; + + return promise; +} + +/** Non-array object values (including class instances); excludes `null` and arrays. */ +function isSleepArgObject(value: unknown): value is Record { + return value !== null && typeof value === 'object' && !Array.isArray(value); +} + +function parseSleepArg(arg: SleepArg): {ms: number; cancelError?: string | Error | null} { + if (typeof arg === 'number') { + if (!Number.isFinite(arg)) { + throw new TypeError('sleep: expected a finite number or an object with ms'); + } + return {ms: arg}; + } + if (isSleepArgObject(arg)) { + const ms = arg.ms; + if (typeof ms !== 'number' || !Number.isFinite(ms)) { + throw new TypeError('sleep: options.ms must be a finite number'); + } + return {ms, cancelError: arg.cancelError as string | Error | null | undefined}; + } + throw new TypeError('sleep: expected a finite number or an object with ms'); +} diff --git a/lib/wait-for-condition.ts b/lib/wait-for-condition.ts new file mode 100644 index 0000000..37f6e4d --- /dev/null +++ b/lib/wait-for-condition.ts @@ -0,0 +1,52 @@ +import {sleep} from './sleep.js'; +import type {WaitForConditionOptions} from './types.js'; + +/** + * Takes a condition (a function returning a boolean or boolean promise), and + * waits until the condition is true. + * + * Throws a `/Condition unmet/` error if the condition has not been satisfied + * within the allocated time, unless an error is provided in the options, as the + * `error` property, which is either thrown itself, or used as the message. + * + * The condition result is returned if it is not falsy. If the condition throws an + * error then this exception will be immediately passed through. + * + * The default options are: `{ waitMs: 5000, intervalMs: 500 }` + * @param condFn - The condition function to evaluate + * @param options - Options for controlling the wait behavior + */ +export async function waitForCondition( + condFn: () => Promise | T, + options: WaitForConditionOptions = {}, +): Promise { + const opts: WaitForConditionOptions & {waitMs: number; intervalMs: number} = { + ...options, + waitMs: typeof options.waitMs === 'number' ? options.waitMs : 5000, + intervalMs: typeof options.intervalMs === 'number' ? options.intervalMs : 500, + }; + const debug = opts.logger ? opts.logger.debug.bind(opts.logger) : () => undefined; + const error = opts.error; + const begunAt = performance.now(); + const endAt = begunAt + opts.waitMs; + const spin = async function spin(): Promise { + const result = await condFn(); + if (result) { + return result; + } + const now = performance.now(); + const waited = Math.round(now - begunAt); + const remainingTime = endAt - now; + if (now < endAt) { + debug(`Waited for ${waited} ms so far`); + await sleep(Math.min(opts.intervalMs, remainingTime)); + return await spin(); + } + // if there is an error option, it is either a string message or an error itself + if (error) { + throw typeof error === 'string' ? new Error(error) : error; + } + throw new Error(`Condition unmet after ${waited} ms. Timing out.`); + }; + return await spin(); +} diff --git a/lib/with-timeout.ts b/lib/with-timeout.ts new file mode 100644 index 0000000..7fbcde4 --- /dev/null +++ b/lib/with-timeout.ts @@ -0,0 +1,43 @@ +import {TimeoutError} from './errors.js'; + +/** + * Resolves with `promise` or rejects if it does not settle before `timeoutMs`. + * + * If the deadline passes first: with no third argument (or a falsy non-error value), rejects + * with {@link TimeoutError} whose message includes `timeoutMs`; with a non-empty string, + * rejects with {@link TimeoutError} using that message; with an `Error` instance, rejects + * with that same instance. + * + * @param promise - Promise to race against the timeout. + * @param timeoutMs - Maximum time in milliseconds before rejecting if `promise` has not settled. + * @param messageOrError - Optional override for the timeout rejection: custom {@link TimeoutError} + * message when the string is non-empty, or an existing `Error` to reject with verbatim. + * Omitted, `undefined`, or other falsy non-error values use the default timeout message (includes `timeoutMs`). + */ +export async function withTimeout( + promise: Promise, + timeoutMs: number, + messageOrError?: string | Error, +): Promise { + let timer: NodeJS.Timeout | null = null; + try { + return await Promise.race([ + promise, + new Promise((_resolve, reject) => { + timer = setTimeout(() => { + if (typeof messageOrError === 'string' && messageOrError) { + reject(new TimeoutError(messageOrError)); + } else if (!messageOrError) { + reject(new TimeoutError(`Operation timed out after ${timeoutMs}ms`)); + } else { + reject(messageOrError); + } + }, timeoutMs); + }), + ]); + } finally { + if (timer) { + clearTimeout(timer); + } + } +} diff --git a/package.json b/package.json index 3b13a16..a15639c 100644 --- a/package.json +++ b/package.json @@ -28,18 +28,18 @@ "build/esm/package.json" ], "type": "module", - "main": "./build/cjs/lib/asyncbox.js", - "module": "./build/esm/lib/asyncbox.js", - "types": "./build/cjs/lib/asyncbox.d.ts", + "main": "./build/cjs/lib/index.js", + "module": "./build/esm/lib/index.js", + "types": "./build/cjs/lib/index.d.ts", "exports": { ".": { "import": { - "types": "./build/esm/lib/asyncbox.d.ts", - "default": "./build/esm/lib/asyncbox.js" + "types": "./build/esm/lib/index.d.ts", + "default": "./build/esm/lib/index.js" }, "require": { - "types": "./build/cjs/lib/asyncbox.d.ts", - "default": "./build/cjs/lib/asyncbox.js" + "types": "./build/cjs/lib/index.d.ts", + "default": "./build/cjs/lib/index.js" } }, "./package.json": "./package.json" diff --git a/test/asyncbox.spec.ts b/test/asyncbox.spec.ts deleted file mode 100644 index 312afcc..0000000 --- a/test/asyncbox.spec.ts +++ /dev/null @@ -1,442 +0,0 @@ -import assert from 'node:assert/strict'; -import {it, describe, beforeEach} from 'node:test'; - -import { - sleep, - longSleep, - retry, - retryInterval, - asyncmap, - asyncfilter, - waitForCondition, - withTimeout, - TimeoutError, - PromiseCancellationError, - type SleepOptions, -} from '../lib/asyncbox.js'; - -describe('sleep', function () { - it('should work like setTimeout', async function () { - const now = Date.now(); - await sleep(20); - assert.ok(Date.now() - now >= 19); - }); - it('should expose cancel on the promise', function () { - const d = sleep(100); - assert.ok(typeof d.cancel === 'function'); - }); - it('should reject with PromiseCancellationError when cancelled', async function () { - const d = sleep(10_000); - d.cancel(); - await assert.rejects(d, PromiseCancellationError); - }); - it('should reject immediately on cancel without waiting for ms', async function () { - const d = sleep(10_000); - const start = Date.now(); - d.cancel(); - await assert.rejects(d, PromiseCancellationError); - assert.ok(Date.now() - start < 100); - }); - it('should use default PromiseCancellationError when cancelError is empty string', async function () { - const d = sleep({ms: 10_000, cancelError: ''}); - d.cancel(); - try { - await d; - assert.fail('expected rejection'); - } catch (err: unknown) { - assert.ok(err instanceof PromiseCancellationError); - assert.strictEqual((err as PromiseCancellationError).message, 'Promise cancelled'); - } - }); - it('should reject with PromiseCancellationError using cancelError string when cancelled', async function () { - const d = sleep({ms: 10_000, cancelError: 'aborted'}); - d.cancel(); - try { - await d; - assert.fail('expected rejection'); - } catch (err: unknown) { - assert.ok(err instanceof PromiseCancellationError); - assert.strictEqual((err as PromiseCancellationError).message, 'aborted'); - } - }); - it('should reject with a provided Error instance on cancel', async function () { - class CustomCancel extends Error { - constructor(message?: string) { - super(message ?? 'custom default'); - this.name = 'CustomCancel'; - } - } - const customErr = new CustomCancel('nope'); - const d = sleep({ms: 10_000, cancelError: customErr}); - d.cancel(); - try { - await d; - assert.fail('expected rejection'); - } catch (err: unknown) { - assert.strictEqual(err, customErr); - assert.ok(err instanceof CustomCancel); - assert.strictEqual((err as CustomCancel).message, 'nope'); - } - }); - it('should resolve when cancelled if cancelError is null', async function () { - const d = sleep({ms: 10_000, cancelError: null}); - const start = Date.now(); - d.cancel(); - await d; - assert.ok(Date.now() - start < 100); - }); - it('should throw TypeError when ms is not finite', function () { - assert.throws(() => sleep(Number.NaN), /finite number or an object with ms/); - assert.throws(() => sleep(Number.POSITIVE_INFINITY), /finite number or an object with ms/); - }); - it('should throw TypeError when arg is not a number or object', function () { - assert.throws(() => sleep(null as unknown as number), /finite number or an object with ms/); - assert.throws(() => sleep([] as unknown as number), /finite number or an object with ms/); - }); - it('should throw TypeError when object has invalid ms', function () { - assert.throws(() => sleep({} as unknown as SleepOptions), /options\.ms must be a finite number/); - assert.throws(() => sleep({ms: Number.NaN}), /options\.ms must be a finite number/); - }); - it('should accept a null-prototype object with ms', async function () { - const o = Object.create(null) as {ms: number}; - o.ms = 20; - const now = Date.now(); - await sleep(o); - assert.ok(Date.now() - now >= 19); - }); - it('should accept a class instance that satisfies SleepOptions', async function () { - class Opts implements SleepOptions { - ms = 25; - } - const now = Date.now(); - await sleep(new Opts()); - assert.ok(Date.now() - now >= 24); - }); -}); - -describe('withTimeout', function () { - function neverSettles(): Promise { - return new Promise(() => {}); - } - - it('should resolve when the promise settles before the deadline', async function () { - const result = await withTimeout(Promise.resolve(42), 1000); - assert.strictEqual(result, 42); - }); - it('should reject with TimeoutError when the deadline is exceeded', async function () { - await assert.rejects(withTimeout(neverSettles(), 30), TimeoutError); - }); - it('should use the default TimeoutError message when none is provided', async function () { - const timeoutMs = 20; - try { - await withTimeout(neverSettles(), timeoutMs); - assert.fail('expected rejection'); - } catch (err: unknown) { - assert.ok(err instanceof TimeoutError); - assert.strictEqual((err as TimeoutError).message, `Operation timed out after ${timeoutMs}ms`); - } - }); - it('should use a custom message on TimeoutError when provided', async function () { - try { - await withTimeout(neverSettles(), 20, 'custom timeout'); - assert.fail('expected rejection'); - } catch (err: unknown) { - assert.ok(err instanceof TimeoutError); - assert.strictEqual((err as TimeoutError).message, 'custom timeout'); - } - }); - it('should reject with a provided Error instance on timeout', async function () { - class CustomTimeout extends Error { - constructor(message?: string) { - super(message ?? 'custom default'); - this.name = 'CustomTimeout'; - } - } - const customErr = new CustomTimeout('overridden'); - try { - await withTimeout(neverSettles(), 20, customErr); - assert.fail('expected rejection'); - } catch (err: unknown) { - assert.strictEqual(err, customErr); - assert.ok(err instanceof CustomTimeout); - assert.strictEqual((err as CustomTimeout).message, 'overridden'); - } - }); - it('should propagate rejection from the underlying promise before the deadline', async function () { - const failing = (async () => { - await sleep(10); - throw new Error('boom'); - })(); - await assert.rejects(withTimeout(failing, 1000), /boom/); - }); -}); - -describe('longSleep', function () { - it('should work like sleep in general', async function () { - const now = Date.now(); - await longSleep(20); - assert.ok(Date.now() - now >= 19); - }); - it('should work like sleep with values less than threshold', async function () { - const now = Date.now(); - await longSleep(20, {thresholdMs: 100}); - assert.ok(Date.now() - now >= 19); - }); - it('should work like sleep with values above threshold, but quantized', async function () { - const now = Date.now(); - await longSleep(50, {thresholdMs: 20, intervalMs: 40}); - assert.ok(Date.now() - now >= 79); - }); - it('should trigger a progress callback if specified', async function () { - let callCount = 0; - let curElapsed = 0; - let curTimeLeft = 10000; - let curProgress = 0; - const progressCb = function ({ - elapsedMs, - timeLeft, - progress, - }: { - elapsedMs: number; - timeLeft: number; - progress: number; - }) { - assert.ok(elapsedMs > curElapsed); - assert.ok(timeLeft < curTimeLeft); - assert.ok(progress > curProgress); - curElapsed = elapsedMs; - curTimeLeft = timeLeft; - curProgress = progress; - callCount += 1; - }; - const now = Date.now(); - await longSleep(500, {thresholdMs: 1, intervalMs: 100, progressCb}); - assert.ok(Date.now() - now > 49); - assert.ok(callCount > 3); - assert.ok(curProgress >= 1); - assert.ok(curTimeLeft <= 0); - assert.ok(curElapsed >= 50); - }); -}); - -describe('retry', function () { - let okFnCalls = 0; - const okFn = async function (val1: number, val2: number): Promise { - await sleep(15); - okFnCalls++; - return val1 * val2; - }; - let badFnCalls = 0; - const badFn = async function (): Promise { - await sleep(15); - badFnCalls++; - throw new Error('bad'); - }; - let eventuallyOkFnCalls = 0; - const eventuallyOkFn = async function (times: number): Promise { - await sleep(15); - eventuallyOkFnCalls++; - if (eventuallyOkFnCalls < times) { - throw new Error('not ok yet'); - } - return times * times; - }; - const eventuallyOkNoSleepFn = async function (times: number): Promise { - eventuallyOkFnCalls++; - if (eventuallyOkFnCalls < times) { - throw new Error('not ok yet'); - } - return times * times; - }; - beforeEach(function () { - okFnCalls = 0; - badFnCalls = 0; - eventuallyOkFnCalls = 0; - }); - it('should return the result of a passing function', async function () { - const start = Date.now(); - const res = await retry(3, okFn, 5, 4); - assert.strictEqual(res, 20); - assert.ok(Date.now() - start >= 14); - assert.strictEqual(okFnCalls, 1); - }); - it('should retry a failing function and eventually throw the same err', async function () { - let err: Error | null = null; - const start = Date.now(); - try { - await retry(3, badFn); - } catch (e) { - err = e as Error; - } - assert.ok(err); - assert.strictEqual(err!.message, 'bad'); - assert.strictEqual(badFnCalls, 3); - assert.ok(Date.now() - start >= 44); - }); - it('should return the correct value with a function that eventually passes', async function () { - let err: Error | null = null; - let start = Date.now(); - try { - await retry(3, eventuallyOkFn, 4); - } catch (e) { - err = e as Error; - } - assert.ok(err); - assert.strictEqual(err!.message, 'not ok yet'); - assert.strictEqual(eventuallyOkFnCalls, 3); - assert.ok(Date.now() - start > 35); - - // rerun with ok number of calls - start = Date.now(); - eventuallyOkFnCalls = 0; - const res = await retry(3, eventuallyOkFn, 3); - assert.strictEqual(eventuallyOkFnCalls, 3); - assert.strictEqual(res, 9); - assert.ok(Date.now() - start > 35); - }); - describe('retryInterval', function () { - it('should return the correct value with a function that eventually passes', async function () { - eventuallyOkFnCalls = 0; - let err: Error | null = null; - let start = Date.now(); - try { - await retryInterval(3, 15, eventuallyOkNoSleepFn, 4); - } catch (e) { - err = e as Error; - } - assert.ok(err); - assert.strictEqual(err!.message, 'not ok yet'); - assert.strictEqual(eventuallyOkFnCalls, 3); - assert.ok(Date.now() - start >= 30); - - // rerun with ok number of calls - start = Date.now(); - eventuallyOkFnCalls = 0; - const res = await retryInterval(3, 15, eventuallyOkNoSleepFn, 3); - assert.strictEqual(eventuallyOkFnCalls, 3); - assert.strictEqual(res, 9); - assert.ok(Date.now() - start >= 30); - }); - it('should not wait on the final error', async function () { - const start = Date.now(); - try { - await retryInterval(3, 2000, badFn); - } catch { - assert.ok(Date.now() - start < 4100); - } - }); - }); -}); - -describe('waitForCondition', function () { - it('should wait and succeed', async function () { - const ref = Date.now(); - function condFn(): boolean { - return Date.now() - ref > 200; - } - const result = await waitForCondition(condFn, {waitMs: 1000, intervalMs: 10}); - const duration = Date.now() - ref; - assert.ok(duration > 200); - assert.ok(duration < 250); - assert.ok(result); - }); - it('should wait and fail', async function () { - const ref = Date.now(); - function condFn(): boolean { - return Date.now() - ref > 200; - } - try { - await waitForCondition(condFn, {waitMs: 100, intervalMs: 10}); - assert.fail('Should have thrown an error'); - } catch (err: any) { - assert.ok(err.message.match(/Condition unmet/)); - } - }); - it('should reduce interval to not exceed timeout', async function () { - const ref = Date.now(); - function condFn(): boolean { - return Date.now() - ref > 25; - } - await waitForCondition(condFn, {waitMs: 30, intervalMs: 20}); - const duration = Date.now() - ref; - assert.ok(duration < 35); - }); -}); - -describe('asyncmap', function () { - const mapper = async function (el: number): Promise { - await sleep(10); - return el * 2; - }; - const coll = [1, 2, 3, 4, 5]; - const newColl = [2, 4, 6, 8, 10]; - it('should map elements one at a time', async function () { - const start = Date.now(); - assert.deepStrictEqual(await asyncmap(coll, mapper, false), newColl); - assert.ok(Date.now() - start >= 50); - }); - it('should map elements in parallel', async function () { - const start = Date.now(); - assert.deepStrictEqual(await asyncmap(coll, mapper), newColl); - assert.ok(Date.now() - start <= 20); - }); - it('should map elements with concurrency', async function () { - const start = Date.now(); - assert.deepStrictEqual(await asyncmap(coll, mapper, {concurrency: 2}), newColl); - assert.ok(Date.now() - start >= 29); - assert.ok(Date.now() - start <= 40); - }); - it('should handle an empty array', async function () { - assert.deepStrictEqual(await asyncmap([], mapper, false), []); - }); - it('should handle an empty array in parallel', async function () { - assert.deepStrictEqual(await asyncmap([], mapper), []); - }); - it('should work for a sync mapper function', async function () { - const syncmapper = (el: number): number => el * 2; - assert.deepStrictEqual(await asyncmap(coll, syncmapper, false), newColl); - assert.deepStrictEqual(await asyncmap(coll, syncmapper), newColl); - }); - it('should raise an error if options is null', async function () { - await assert.rejects(() => asyncmap(coll, mapper, null as any), /Options cannot be null/); - }); -}); - -describe('asyncfilter', function () { - const filter = async function (el: number): Promise { - await sleep(10); - return el % 2 === 0; - }; - const coll = [1, 2, 3, 4, 5]; - const newColl = [2, 4]; - it('should filter elements one at a time', async function () { - const start = Date.now(); - assert.deepStrictEqual(await asyncfilter(coll, filter, false), newColl); - assert.ok(Date.now() - start >= 50); - }); - it('should filter elements in parallel', async function () { - const start = Date.now(); - assert.deepStrictEqual(await asyncfilter(coll, filter), newColl); - assert.ok(Date.now() - start <= 20); - }); - it('should filter elements with concurrency', async function () { - const start = Date.now(); - assert.deepStrictEqual(await asyncfilter(coll, filter, {concurrency: 2}), newColl); - assert.ok(Date.now() - start >= 29); - assert.ok(Date.now() - start <= 40); - }); - it('should handle an empty array', async function () { - assert.deepStrictEqual(await asyncfilter([], filter, false), []); - }); - it('should handle an empty array in parallel', async function () { - assert.deepStrictEqual(await asyncfilter([], filter), []); - }); - it('should work for a sync filter function', async function () { - const syncfilter = (el: number): boolean => el % 2 === 0; - assert.deepStrictEqual(await asyncfilter(coll, syncfilter, false), newColl); - assert.deepStrictEqual(await asyncfilter(coll, syncfilter), newColl); - }); - it('should raise an error if options is null', async function () { - await assert.rejects(() => asyncfilter(coll, filter, null as any), /Options cannot be null/); - }); -}); diff --git a/test/asyncfilter.spec.ts b/test/asyncfilter.spec.ts new file mode 100644 index 0000000..8795d01 --- /dev/null +++ b/test/asyncfilter.spec.ts @@ -0,0 +1,48 @@ +import assert from 'node:assert/strict'; +import {describe, it} from 'node:test'; + +import {asyncfilter} from '../lib/asyncfilter.js'; +import {sleep} from '../lib/sleep.js'; +import {createConcurrencyTracker} from './helpers.js'; + +describe('asyncfilter', function () { + const coll = [1, 2, 3, 4, 5]; + const expected = [2, 4]; + const makeFn = () => { + const tracker = createConcurrencyTracker(); + const fn = tracker.wrap(async (el: number): Promise => { + await sleep(10); + return el % 2 === 0; + }); + return {tracker, fn}; + }; + it('should filter elements one at a time', async function () { + const {tracker, fn} = makeFn(); + assert.deepStrictEqual(await asyncfilter(coll, fn, false), expected); + assert.strictEqual(tracker.max, 1); + }); + it('should filter elements in parallel', async function () { + const {tracker, fn} = makeFn(); + assert.deepStrictEqual(await asyncfilter(coll, fn), expected); + assert.strictEqual(tracker.max, coll.length); + }); + it('should filter elements with concurrency', async function () { + const {tracker, fn} = makeFn(); + assert.deepStrictEqual(await asyncfilter(coll, fn, {concurrency: 2}), expected); + assert.strictEqual(tracker.max, 2); + }); + it('should handle an empty array', async function () { + assert.deepStrictEqual(await asyncfilter([], makeFn().fn, false), []); + }); + it('should handle an empty array in parallel', async function () { + assert.deepStrictEqual(await asyncfilter([], makeFn().fn), []); + }); + it('should work for a sync filter function', async function () { + const syncFn = (el: number): boolean => el % 2 === 0; + assert.deepStrictEqual(await asyncfilter(coll, syncFn, false), expected); + assert.deepStrictEqual(await asyncfilter(coll, syncFn), expected); + }); + it('should raise an error if options is null', async function () { + await assert.rejects(() => asyncfilter(coll, makeFn().fn, null as any), /Options cannot be null/); + }); +}); diff --git a/test/asyncmap.spec.ts b/test/asyncmap.spec.ts new file mode 100644 index 0000000..0ac33a4 --- /dev/null +++ b/test/asyncmap.spec.ts @@ -0,0 +1,48 @@ +import assert from 'node:assert/strict'; +import {describe, it} from 'node:test'; + +import {asyncmap} from '../lib/asyncmap.js'; +import {sleep} from '../lib/sleep.js'; +import {createConcurrencyTracker} from './helpers.js'; + +describe('asyncmap', function () { + const coll = [1, 2, 3, 4, 5]; + const expected = [2, 4, 6, 8, 10]; + const makeFn = () => { + const tracker = createConcurrencyTracker(); + const fn = tracker.wrap(async (el: number): Promise => { + await sleep(10); + return el * 2; + }); + return {tracker, fn}; + }; + it('should map elements one at a time', async function () { + const {tracker, fn} = makeFn(); + assert.deepStrictEqual(await asyncmap(coll, fn, false), expected); + assert.strictEqual(tracker.max, 1); + }); + it('should map elements in parallel', async function () { + const {tracker, fn} = makeFn(); + assert.deepStrictEqual(await asyncmap(coll, fn), expected); + assert.strictEqual(tracker.max, coll.length); + }); + it('should map elements with concurrency', async function () { + const {tracker, fn} = makeFn(); + assert.deepStrictEqual(await asyncmap(coll, fn, {concurrency: 2}), expected); + assert.strictEqual(tracker.max, 2); + }); + it('should handle an empty array', async function () { + assert.deepStrictEqual(await asyncmap([], makeFn().fn, false), []); + }); + it('should handle an empty array in parallel', async function () { + assert.deepStrictEqual(await asyncmap([], makeFn().fn), []); + }); + it('should work for a sync map function', async function () { + const syncFn = (el: number): number => el * 2; + assert.deepStrictEqual(await asyncmap(coll, syncFn, false), expected); + assert.deepStrictEqual(await asyncmap(coll, syncFn), expected); + }); + it('should raise an error if options is null', async function () { + await assert.rejects(() => asyncmap(coll, makeFn().fn, null as any), /Options cannot be null/); + }); +}); diff --git a/test/cjs-interop.spec.cjs b/test/cjs-interop.spec.cjs index 3519ce2..68132b8 100644 --- a/test/cjs-interop.spec.cjs +++ b/test/cjs-interop.spec.cjs @@ -7,5 +7,6 @@ const asyncbox = require('asyncbox'); test('require() interop with the CJS build', async () => { assert.equal(typeof asyncbox.sleep, 'function'); assert.equal(typeof asyncbox.retry, 'function'); + assert.equal(typeof asyncbox.AsyncLock, 'function'); await asyncbox.sleep(1); }); diff --git a/test/helpers.ts b/test/helpers.ts new file mode 100644 index 0000000..647e703 --- /dev/null +++ b/test/helpers.ts @@ -0,0 +1,33 @@ +import type {TestContext} from 'node:test'; + +/** Records the delay of every `setTimeout` call, still scheduling it as usual. */ +export function recordTimeoutDelays(t: TestContext): number[] { + const delays: number[] = []; + const original = globalThis.setTimeout; + t.mock.method(globalThis, 'setTimeout', ((fn: () => void, ms?: number, ...args: unknown[]) => { + delays.push(ms as number); + return (original as any)(fn, ms, ...args); + }) as any); + return delays; +} + +/** Wraps functions to measure how many of them run at the same time. */ +export function createConcurrencyTracker() { + let active = 0; + let max = 0; + return { + get max(): number { + return max; + }, + wrap(fn: (...args: A) => Promise): (...args: A) => Promise { + return async (...args: A) => { + max = Math.max(max, ++active); + try { + return await fn(...args); + } finally { + active--; + } + }; + }, + }; +} diff --git a/test/index.spec.ts b/test/index.spec.ts new file mode 100644 index 0000000..55be8da --- /dev/null +++ b/test/index.spec.ts @@ -0,0 +1,26 @@ +import assert from 'node:assert/strict'; +import {describe, it} from 'node:test'; + +import * as asyncbox from '../lib/index.js'; + +describe('public API', function () { + it('should export exactly the documented runtime members', function () { + assert.deepStrictEqual(Object.keys(asyncbox).sort(), [ + 'AsyncLock', + 'PromiseCancellationError', + 'TimeoutError', + 'asyncfilter', + 'asyncmap', + 'longSleep', + 'retry', + 'retryInterval', + 'sleep', + 'waitForCondition', + 'withTimeout', + ]); + }); + it('should export error classes that extend Error', function () { + assert.ok(new asyncbox.TimeoutError() instanceof Error); + assert.ok(new asyncbox.PromiseCancellationError() instanceof Error); + }); +}); diff --git a/test/lock.spec.ts b/test/lock.spec.ts new file mode 100644 index 0000000..70f66b3 --- /dev/null +++ b/test/lock.spec.ts @@ -0,0 +1,569 @@ +import assert from 'node:assert/strict'; +import {AsyncLocalStorage} from 'node:async_hooks'; +import {describe, it} from 'node:test'; + +import {AsyncLock} from '../lib/lock.js'; +import {sleep} from '../lib/sleep.js'; + +function deferred() { + let resolve!: (v: T) => void; + let reject!: (e: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return {promise, resolve, reject}; +} + +describe('AsyncLock', function () { + it('should return the task result', async function () { + const lock = new AsyncLock(); + assert.equal(await lock.acquire('k', () => 1), 1); + assert.equal(await lock.acquire('k', async () => 'a'), 'a'); + }); + + it('should start a task synchronously if the lock is free', function () { + const lock = new AsyncLock(); + let ran = false; + const p = lock.acquire('k', () => { + ran = true; + }); + assert.ok(ran); + assert.ok(lock.isBusy('k')); + return p; + }); + + it('should run tasks with the same key one at a time in FIFO order', async function () { + const lock = new AsyncLock(); + const log: string[] = []; + const task = (name: string, ms: number) => async () => { + log.push(`${name}:start`); + await sleep(ms); + log.push(`${name}:end`); + }; + await Promise.all([ + lock.acquire('k', task('a', 20)), + lock.acquire('k', task('b', 1)), + lock.acquire('k', task('c', 1)), + ]); + assert.deepEqual(log, ['a:start', 'a:end', 'b:start', 'b:end', 'c:start', 'c:end']); + }); + + it('should run tasks with different keys concurrently', async function () { + const lock = new AsyncLock(); + const gate = deferred(); + const started: string[] = []; + const a = lock.acquire('a', async () => { + started.push('a'); + await gate.promise; + }); + const b = lock.acquire('b', async () => { + started.push('b'); + await gate.promise; + }); + assert.deepEqual(started, ['a', 'b']); + gate.resolve(); + await Promise.all([a, b]); + }); + + it('should propagate sync throws and rejections and keep working', async function () { + const lock = new AsyncLock(); + await assert.rejects( + lock.acquire('k', () => { + throw new Error('sync'); + }), + /sync/, + ); + await assert.rejects( + lock.acquire('k', async () => Promise.reject(new Error('async'))), + /async/, + ); + assert.equal(await lock.acquire('k', () => 'ok'), 'ok'); + assert.ok(!lock.isBusy()); + }); + + it('should throw if the task is not a function', function () { + const lock = new AsyncLock(); + assert.throws(() => lock.acquire('k', 'nope' as any), TypeError); + assert.ok(!lock.isBusy()); + }); + + it('should throw if the key is neither a string nor a non-empty array of strings', function () { + const lock = new AsyncLock(); + const sparse: string[] = []; + sparse[0] = 'a'; + sparse[2] = 'b'; // leaves a hole at index 1 + for (const key of [undefined, null, 1, {}, ['valid', 123], [undefined], sparse, new Array(2)]) { + assert.throws(() => lock.acquire(key as any, () => 1), TypeError); + } + assert.ok(!lock.isBusy()); + }); + + it('should support keys that clash with Object.prototype members', async function () { + const lock = new AsyncLock(); + for (const key of ['__proto__', 'constructor', 'toString', 'hasOwnProperty']) { + assert.equal(await lock.acquire(key, () => key), key); + } + assert.ok(!lock.isBusy()); + }); + + describe('async context', function () { + it('should run queued tasks in the context of their own caller', async function () { + const als = new AsyncLocalStorage(); + const lock = new AsyncLock(); + const gate = deferred(); + const first = als.run('first', () => + lock.acquire('k', async () => { + await gate.promise; + return als.getStore(); + }), + ); + const second = als.run('second', () => lock.acquire('k', () => als.getStore())); + const third = als.run('third', () => lock.acquire('k', async () => als.getStore())); + gate.resolve(); + assert.deepEqual(await Promise.all([first, second, third]), ['first', 'second', 'third']); + }); + + it('should keep the context for every key of a batch acquire', async function () { + const als = new AsyncLocalStorage(); + const lock = new AsyncLock(); + const gate = deferred(); + const holder = als.run('holder', () => + lock.acquire('b', async () => { + await gate.promise; + }), + ); + const waiter = als.run('waiter', () => lock.acquire(['a', 'b'], () => als.getStore())); + gate.resolve(); + await holder; + assert.equal(await waiter, 'waiter'); + }); + }); + + describe('multiple keys', function () { + it('should hold all the keys while the task runs', async function () { + const lock = new AsyncLock(); + const gate = deferred(); + const p = lock.acquire(['a', 'b'], () => gate.promise); + assert.ok(lock.isBusy('a')); + assert.ok(lock.isBusy('b')); + gate.resolve(); + await p; + assert.ok(!lock.isBusy()); + }); + + it('should wait for any of the keys', async function () { + const lock = new AsyncLock(); + const gate = deferred(); + const log: string[] = []; + const holder = lock.acquire('b', async () => { + await gate.promise; + log.push('holder'); + }); + const batch = lock.acquire(['a', 'b'], () => { + log.push('batch'); + }); + await sleep(5); + assert.deepEqual(log, []); + gate.resolve(); + await Promise.all([holder, batch]); + assert.deepEqual(log, ['holder', 'batch']); + }); + + it('should not deadlock overlapping calls that list the keys in opposite orders', async function () { + const lock = new AsyncLock(); + const log: string[] = []; + const task = (name: string) => async () => { + log.push(name); + await sleep(5); + }; + // a deadlock would make `maxOccupationTime` reject the calls + const opts = {maxOccupationTime: 2000}; + await Promise.all([ + lock.acquire(['a', 'b'], task('ab'), opts), + lock.acquire(['b', 'a'], task('ba'), opts), + lock.acquire(['a', 'b', 'c'], task('abc'), opts), + lock.acquire(['c', 'b'], task('cb'), opts), + ]); + assert.deepEqual([...log].sort(), ['ab', 'abc', 'ba', 'cb']); + assert.ok(!lock.isBusy()); + }); + + it('should ignore duplicate keys instead of deadlocking', async function () { + const lock = new AsyncLock(); + assert.equal(await lock.acquire(['a', 'a'], () => 1, {timeout: 50}), 1); + }); + + it('should throw for an empty key list instead of running the task without a lock', function () { + const lock = new AsyncLock(); + assert.throws(() => lock.acquire([], () => 1), TypeError); + }); + + it('should claim a large batch of free keys without overflowing the stack', async function () { + const lock = new AsyncLock(); + const keys = Array.from({length: 20_000}, (_, i) => `key${i}`); + assert.equal(await lock.acquire(keys, () => 1), 1); + assert.ok(!lock.isBusy()); + }); + + it('should claim a large batch resumed after waiting for its first key', async function () { + const lock = new AsyncLock(); + const keys = Array.from({length: 20_000}, (_, i) => `key${i}`); + const gate = deferred(); + const holder = lock.acquire(keys[0], () => gate.promise); + const batch = lock.acquire(keys, () => 1); + gate.resolve(); + await holder; + assert.equal(await batch, 1); + assert.ok(!lock.isBusy()); + }); + + it('should not mutate the passed array', async function () { + const lock = new AsyncLock(); + const keys = ['a', 'b']; + await lock.acquire(keys, () => {}); + assert.deepEqual(keys, ['a', 'b']); + }); + }); + + describe('options', function () { + it('should run a skipQueue task before the already queued ones', async function () { + const lock = new AsyncLock(); + const gate = deferred(); + const log: string[] = []; + const ps = [ + lock.acquire('k', () => gate.promise), + lock.acquire('k', () => void log.push('queued')), + lock.acquire('k', () => void log.push('priority'), {skipQueue: true}), + ]; + gate.resolve(); + await Promise.all(ps); + assert.deepEqual(log, ['priority', 'queued']); + }); + + it('should reject when too many tasks are pending', async function () { + const lock = new AsyncLock({maxPending: 1}); + const gate = deferred(); + const running = lock.acquire('k', () => gate.promise); + const pending = lock.acquire('k', () => 'pending'); + await assert.rejects( + lock.acquire('k', () => 'extra'), + /Too many pending tasks in queue k/, + ); + gate.resolve(); + await running; + assert.equal(await pending, 'pending'); + }); + + it('should allow no pending tasks with maxPending 0 (per call too)', async function () { + const lock = new AsyncLock(); + const gate = deferred(); + const running = lock.acquire('k', () => gate.promise); + await assert.rejects( + lock.acquire('k', () => 1, {maxPending: 0}), + /Too many pending/, + ); + gate.resolve(); + await running; + }); + + it('should use 1000 as the default maxPending', async function () { + const lock = new AsyncLock(); + const gate = deferred(); + const ps = [lock.acquire('k', () => gate.promise)]; + for (let i = 0; i < 1000; i++) { + ps.push(lock.acquire('k', () => {})); + } + await assert.rejects( + lock.acquire('k', () => {}), + /Too many pending/, + ); + gate.resolve(); + await Promise.all(ps); + }); + + it('should time out a task waiting in the queue and drop it from the queue', async function () { + const lock = new AsyncLock({maxPending: 1}); + const gate = deferred(); + let ran = false; + const running = lock.acquire('k', () => gate.promise); + await assert.rejects( + lock.acquire('k', () => void (ran = true), {timeout: 10}), + /timed out in queue k/, + ); + // the timed out task no longer occupies the single pending slot + const next = lock.acquire('k', () => 'next'); + gate.resolve(); + await running; + assert.equal(await next, 'next'); + assert.ok(!ran); + }); + + it('should not time out a task that got the lock in time', async function () { + const lock = new AsyncLock({timeout: 50}); + const first = lock.acquire('k', () => {}); + const second = lock.acquire('k', async () => { + await sleep(120); // longer than the queue timeout + return 'done'; + }); + await first; + assert.equal(await second, 'done'); + }); + + it('should fail on maxOccupationTime, also while still queued', async function () { + const lock = new AsyncLock({maxOccupationTime: 20}); + const gate = deferred(); + let ran = false; + const running = lock.acquire('k', () => gate.promise, {maxOccupationTime: 0}); + await assert.rejects( + lock.acquire('k', () => void (ran = true)), + /Maximum occupation time is exceeded in queue k/, + ); + gate.resolve(); + await running; + assert.ok(!ran); + assert.ok(!lock.isBusy()); + }); + + it('should share one maxOccupationTime budget across all keys of a batch', async function () { + const lock = new AsyncLock(); + const gateA = deferred(); + const gateB = deferred(); + let ran = false; + const holderA = lock.acquire('a', () => gateA.promise); + const holderB = lock.acquire('b', () => gateB.promise); + const batch = lock.acquire(['a', 'b'], () => void (ran = true), {maxOccupationTime: 200}); + await sleep(100); + gateA.resolve(); // the batch gets 'a' with about half of its budget left and queues on 'b' + await holderA; + await assert.rejects(batch, /Maximum occupation time is exceeded/); + // the nested 'b' waiter must expire with the batch instead of getting a fresh budget + await sleep(50); + assert.equal(lock.getQueueLength('b'), 0); + gateB.resolve(); + await holderB; + assert.ok(!ran); + assert.ok(!lock.isBusy()); + }); + + it('should apply maxExecutionTime of a batch to the task only and release all keys', async function () { + const lock = new AsyncLock(); + const holderGate = deferred(); + const taskGate = deferred(); + let ran = false; + let settled = false; + const holder = lock.acquire('b', () => holderGate.promise); + const batch = lock.acquire( + ['a', 'b'], + () => { + ran = true; + return taskGate.promise; + }, + {maxExecutionTime: 20}, + ); + batch.then( + () => (settled = true), + () => (settled = true), + ); + await sleep(60); // waiting for 'b' is not execution, so 'a' stays held and nothing times out + assert.ok(!ran); + assert.ok(!settled); + assert.ok(lock.isBusy('a')); + holderGate.resolve(); + await holder; + await assert.rejects(batch, /Maximum execution time is exceeded/); + assert.ok(ran); + assert.ok(!lock.isBusy()); + taskGate.resolve(); + }); + + it('should release all keys of a batch before starting a waiter', async function () { + for (const [waitedKey, otherKey] of [ + ['a', 'b'], + ['b', 'a'], + ]) { + const lock = new AsyncLock(); + const taskGate = deferred(); + const batch = lock.acquire(['a', 'b'], () => taskGate.promise, {maxExecutionTime: 10}); + const otherKeyBusy = lock.acquire(waitedKey, () => lock.isBusy(otherKey)); + await assert.rejects(batch, /Maximum execution time/); + assert.equal(await otherKeyBusy, false, `'${otherKey}' was still held when a waiter of '${waitedKey}' started`); + taskGate.resolve(); + } + }); + + it('should keep the lock after maxOccupationTime until the task finishes', async function () { + const lock = new AsyncLock(); + const gate = deferred(); + const log: string[] = []; + const slow = lock.acquire( + 'k', + async () => { + await gate.promise; + log.push('slow'); + }, + {maxOccupationTime: 10}, + ); + const next = lock.acquire('k', () => void log.push('next')); + await assert.rejects(slow, /Maximum occupation time is exceeded while executing k/); + assert.deepEqual(log, []); + gate.resolve(); + await next; + assert.deepEqual(log, ['slow', 'next']); + }); + + it('should release the lock exactly once on maxExecutionTime', async function () { + const lock = new AsyncLock({maxExecutionTime: 10}); + const gate = deferred(); + const secondDone = deferred(); + let active = 0; + let maxActive = 0; + const track = (fn: () => Promise) => async () => { + maxActive = Math.max(maxActive, ++active); + try { + await fn(); + } finally { + active--; + } + }; + const opts = {maxExecutionTime: 0}; + const stuck = lock.acquire('k', () => gate.promise); + const second = lock.acquire( + 'k', + track(() => secondDone.promise), + opts, + ); + const third = lock.acquire( + 'k', + track(async () => {}), + opts, + ); + await assert.rejects(stuck, /Maximum execution time is exceeded k/); + // the stuck task finishing late must not release the lock held by `second` + gate.resolve(); + await sleep(30); + assert.equal(active, 1, 'only `second` is running'); + secondDone.resolve(); + await Promise.all([second, third]); + assert.equal(maxActive, 1); + assert.ok(!lock.isBusy()); + }); + + it('should apply timeout to the whole wait for all the keys of a batch', async function () { + const lock = new AsyncLock(); + const gateA = deferred(); + const gateB = deferred(); + const holderA = lock.acquire('a', () => gateA.promise); + const holderB = lock.acquire('b', () => gateB.promise); + const batch = lock.acquire(['a', 'b'], () => 'done', {timeout: 200}); + const outcome = batch.then( + () => 'acquired', + (e: Error) => e.message, + ); + await sleep(100); + gateA.resolve(); // the batch holds 'a' now and waits for 'b' with about half of its budget left + await holderA; + await sleep(180); // 'b' is released after the whole budget (200) but before a fresh one (100 + 200) + gateB.resolve(); + await holderB; + assert.match(await outcome, /timed out in queue b/); + assert.ok(!lock.isBusy()); + }); + + it('should let per-call options override the constructor ones', async function () { + const lock = new AsyncLock({timeout: 5}); + const gate = deferred(); + const running = lock.acquire('k', () => gate.promise); + const waiting = lock.acquire('k', () => 'ok', {timeout: 0}); + await sleep(20); + gate.resolve(); + await running; + assert.equal(await waiting, 'ok'); + }); + + it('should validate options', function () { + assert.throws(() => new AsyncLock(null as any), TypeError); + assert.throws(() => new AsyncLock({timeout: -1}), TypeError); + assert.throws(() => new AsyncLock({maxOccupationTime: 3e9}), TypeError); + assert.throws(() => new AsyncLock({maxPending: 1.5}), TypeError); + assert.throws(() => new AsyncLock({maxExecutionTime: NaN}), TypeError); + assert.doesNotThrow(() => new AsyncLock({maxPending: Infinity})); + }); + + it('should throw on invalid per-call options', function () { + const lock = new AsyncLock(); + assert.throws(() => lock.acquire('k', () => 1, {timeout: -1}), TypeError); + assert.throws(() => lock.acquire('k', () => 1, {timeout: 2 ** 31}), TypeError); // would fire at once + assert.throws(() => lock.acquire('k', () => 1, null as any), TypeError); + assert.throws(() => lock.acquire('k', () => 1, {skipQueue: 'yes' as any}), TypeError); + assert.ok(!lock.isBusy()); + }); + }); + + describe('handover', function () { + it('should pass the key on if the waiter is aborted while the key is being handed over', async function (t) { + t.mock.timers.enable({apis: ['setTimeout']}); + const lock = new AsyncLock(); + const gate = deferred(); + let victimRan = false; + // releases 'b' first, then 'a'; the waiter on 'b' runs first and expires the waiter on 'a' + const holder = lock.acquire(['a', 'b'], () => gate.promise); + const trigger = lock.acquire('b', () => t.mock.timers.tick(100)); + const victim = lock.acquire('a', () => void (victimRan = true), {timeout: 50}); + const victimOutcome = assert.rejects(victim, /timed out in queue a/); + gate.resolve(); + await Promise.all([holder, trigger, victimOutcome]); + await Promise.resolve(); + assert.ok(!victimRan, 'an expired waiter must not run its task'); + assert.ok(!lock.isBusy(), 'the key must not stay held'); + assert.equal(await lock.acquire('a', () => 'free'), 'free'); + }); + }); + + describe('getQueueLength', function () { + it('should count waiting tasks, not the running one', async function () { + const lock = new AsyncLock(); + assert.equal(lock.getQueueLength('k'), 0); + const gate = deferred(); + const ps = [lock.acquire('k', () => gate.promise)]; + assert.equal(lock.getQueueLength('k'), 0); + ps.push( + lock.acquire('k', () => {}), + lock.acquire('k', () => {}), + ); + assert.equal(lock.getQueueLength('k'), 2); + assert.equal(lock.getQueueLength('other'), 0); + gate.resolve(); + await Promise.all(ps); + assert.equal(lock.getQueueLength('k'), 0); + }); + + it('should not count tasks that timed out in the queue', async function () { + const lock = new AsyncLock(); + const gate = deferred(); + const running = lock.acquire('k', () => gate.promise); + const waiting = lock.acquire('k', () => {}, {timeout: 10}); + assert.equal(lock.getQueueLength('k'), 1); + await assert.rejects(waiting, /timed out/); + assert.equal(lock.getQueueLength('k'), 0); + gate.resolve(); + await running; + }); + }); + + describe('isBusy', function () { + it('should report running and pending keys', async function () { + const lock = new AsyncLock(); + assert.ok(!lock.isBusy()); + assert.ok(!lock.isBusy('k')); + const gate = deferred(); + const p = lock.acquire('k', () => gate.promise); + assert.ok(lock.isBusy()); + assert.ok(lock.isBusy('k')); + assert.ok(!lock.isBusy('other')); + gate.resolve(); + await p; + assert.ok(!lock.isBusy()); + }); + }); +}); diff --git a/test/long-sleep.spec.ts b/test/long-sleep.spec.ts new file mode 100644 index 0000000..78238b4 --- /dev/null +++ b/test/long-sleep.spec.ts @@ -0,0 +1,51 @@ +import assert from 'node:assert/strict'; +import {describe, it} from 'node:test'; + +import {longSleep} from '../lib/long-sleep.js'; +import {recordTimeoutDelays} from './helpers.js'; + +describe('longSleep', function () { + it('should work like sleep in general', async function (t) { + const delays = recordTimeoutDelays(t); + await longSleep(20); + assert.deepStrictEqual(delays, [20]); + }); + it('should work like sleep with values less than threshold', async function (t) { + const delays = recordTimeoutDelays(t); + await longSleep(20, {thresholdMs: 100}); + assert.deepStrictEqual(delays, [20]); + }); + it('should work like sleep with values above threshold, but quantized', async function (t) { + const delays = recordTimeoutDelays(t); + await longSleep(50, {thresholdMs: 20, intervalMs: 40}); + assert.deepStrictEqual(delays, [40, 40]); + }); + it('should trigger a progress callback if specified', async function () { + let callCount = 0; + let curElapsed = 0; + let curTimeLeft = 10000; + let curProgress = 0; + const progressCb = function ({ + elapsedMs, + timeLeft, + progress, + }: { + elapsedMs: number; + timeLeft: number; + progress: number; + }) { + assert.ok(elapsedMs > curElapsed); + assert.ok(timeLeft < curTimeLeft); + assert.ok(progress > curProgress); + curElapsed = elapsedMs; + curTimeLeft = timeLeft; + curProgress = progress; + callCount += 1; + }; + await longSleep(500, {thresholdMs: 1, intervalMs: 100, progressCb}); + assert.ok(callCount > 3); + assert.ok(curProgress >= 1); + assert.ok(curTimeLeft <= 0); + assert.ok(curElapsed >= 50); + }); +}); diff --git a/test/retry.spec.ts b/test/retry.spec.ts new file mode 100644 index 0000000..5305519 --- /dev/null +++ b/test/retry.spec.ts @@ -0,0 +1,84 @@ +import assert from 'node:assert/strict'; +import {beforeEach, describe, it} from 'node:test'; + +import {retry, retryInterval} from '../lib/retry.js'; +import {sleep} from '../lib/sleep.js'; +import {recordTimeoutDelays} from './helpers.js'; + +describe('retry', function () { + let okFnCalls = 0; + const okFn = async function (val1: number, val2: number): Promise { + await sleep(15); + okFnCalls++; + return val1 * val2; + }; + let badFnCalls = 0; + const badFn = async function (): Promise { + await sleep(15); + badFnCalls++; + throw new Error('bad'); + }; + let eventuallyOkFnCalls = 0; + const eventuallyOkFn = async function (times: number): Promise { + await sleep(15); + eventuallyOkFnCalls++; + if (eventuallyOkFnCalls < times) { + throw new Error('not ok yet'); + } + return times * times; + }; + const eventuallyOkNoSleepFn = async function (times: number): Promise { + eventuallyOkFnCalls++; + if (eventuallyOkFnCalls < times) { + throw new Error('not ok yet'); + } + return times * times; + }; + beforeEach(function () { + okFnCalls = 0; + badFnCalls = 0; + eventuallyOkFnCalls = 0; + }); + it('should return the result of a passing function', async function () { + const res = await retry(3, okFn, 5, 4); + assert.strictEqual(res, 20); + assert.strictEqual(okFnCalls, 1); + }); + it('should retry a failing function and eventually throw the same err', async function () { + await assert.rejects(retry(3, badFn), {message: 'bad'}); + assert.strictEqual(badFnCalls, 3); + }); + it('should return the correct value with a function that eventually passes', async function () { + await assert.rejects(retry(3, eventuallyOkFn, 4), {message: 'not ok yet'}); + assert.strictEqual(eventuallyOkFnCalls, 3); + + // rerun with ok number of calls + eventuallyOkFnCalls = 0; + const res = await retry(3, eventuallyOkFn, 3); + assert.strictEqual(eventuallyOkFnCalls, 3); + assert.strictEqual(res, 9); + }); + describe('retryInterval', function () { + it('should return the correct value with a function that eventually passes', async function (t) { + const delays = recordTimeoutDelays(t); + await assert.rejects(retryInterval(3, 15, eventuallyOkNoSleepFn, 4), {message: 'not ok yet'}); + assert.strictEqual(eventuallyOkFnCalls, 3); + + // rerun with ok number of calls + eventuallyOkFnCalls = 0; + const res = await retryInterval(3, 15, eventuallyOkNoSleepFn, 3); + assert.strictEqual(eventuallyOkFnCalls, 3); + assert.strictEqual(res, 9); + // two pauses per run: after each failed attempt but the last + assert.deepStrictEqual( + delays.filter((ms) => ms === 15), + [15, 15, 15, 15], + ); + }); + it('should not wait on the final error', async function (t) { + const delays = recordTimeoutDelays(t); + await assert.rejects(retryInterval(3, 23, badFn), {message: 'bad'}); + assert.strictEqual(delays.filter((ms) => ms === 23).length, 2); + }); + }); +}); diff --git a/test/sleep.spec.ts b/test/sleep.spec.ts new file mode 100644 index 0000000..b26edf2 --- /dev/null +++ b/test/sleep.spec.ts @@ -0,0 +1,106 @@ +import assert from 'node:assert/strict'; +import {describe, it} from 'node:test'; + +import {PromiseCancellationError} from '../lib/errors.js'; +import {sleep} from '../lib/sleep.js'; +import type {SleepOptions} from '../lib/types.js'; +import {recordTimeoutDelays} from './helpers.js'; + +describe('sleep', function () { + it('should work like setTimeout', async function (t) { + const delays = recordTimeoutDelays(t); + await sleep(20); + assert.deepStrictEqual(delays, [20]); + }); + it('should expose cancel on the promise', function () { + const d = sleep(100); + assert.ok(typeof d.cancel === 'function'); + }); + it('should reject with PromiseCancellationError when cancelled', async function () { + const d = sleep(10_000); + d.cancel(); + await assert.rejects(d, PromiseCancellationError); + }); + it('should reject immediately on cancel without waiting for ms', async function () { + const d = sleep(10_000); + const start = performance.now(); + d.cancel(); + await assert.rejects(d, PromiseCancellationError); + assert.ok(performance.now() - start < 5000); + }); + it('should use default PromiseCancellationError when cancelError is empty string', async function () { + const d = sleep({ms: 10_000, cancelError: ''}); + d.cancel(); + try { + await d; + assert.fail('expected rejection'); + } catch (err: unknown) { + assert.ok(err instanceof PromiseCancellationError); + assert.strictEqual((err as PromiseCancellationError).message, 'Promise cancelled'); + } + }); + it('should reject with PromiseCancellationError using cancelError string when cancelled', async function () { + const d = sleep({ms: 10_000, cancelError: 'aborted'}); + d.cancel(); + try { + await d; + assert.fail('expected rejection'); + } catch (err: unknown) { + assert.ok(err instanceof PromiseCancellationError); + assert.strictEqual((err as PromiseCancellationError).message, 'aborted'); + } + }); + it('should reject with a provided Error instance on cancel', async function () { + class CustomCancel extends Error { + constructor(message?: string) { + super(message ?? 'custom default'); + this.name = 'CustomCancel'; + } + } + const customErr = new CustomCancel('nope'); + const d = sleep({ms: 10_000, cancelError: customErr}); + d.cancel(); + try { + await d; + assert.fail('expected rejection'); + } catch (err: unknown) { + assert.strictEqual(err, customErr); + assert.ok(err instanceof CustomCancel); + assert.strictEqual((err as CustomCancel).message, 'nope'); + } + }); + it('should resolve when cancelled if cancelError is null', async function () { + const d = sleep({ms: 10_000, cancelError: null}); + const start = performance.now(); + d.cancel(); + await d; + assert.ok(performance.now() - start < 5000); + }); + it('should throw TypeError when ms is not finite', function () { + assert.throws(() => sleep(Number.NaN), /finite number or an object with ms/); + assert.throws(() => sleep(Number.POSITIVE_INFINITY), /finite number or an object with ms/); + }); + it('should throw TypeError when arg is not a number or object', function () { + assert.throws(() => sleep(null as unknown as number), /finite number or an object with ms/); + assert.throws(() => sleep([] as unknown as number), /finite number or an object with ms/); + }); + it('should throw TypeError when object has invalid ms', function () { + assert.throws(() => sleep({} as unknown as SleepOptions), /options\.ms must be a finite number/); + assert.throws(() => sleep({ms: Number.NaN}), /options\.ms must be a finite number/); + }); + it('should accept a null-prototype object with ms', async function (t) { + const delays = recordTimeoutDelays(t); + const o = Object.create(null) as {ms: number}; + o.ms = 20; + await sleep(o); + assert.deepStrictEqual(delays, [20]); + }); + it('should accept a class instance that satisfies SleepOptions', async function (t) { + class Opts implements SleepOptions { + ms = 25; + } + const delays = recordTimeoutDelays(t); + await sleep(new Opts()); + assert.deepStrictEqual(delays, [25]); + }); +}); diff --git a/test/wait-for-condition.spec.ts b/test/wait-for-condition.spec.ts new file mode 100644 index 0000000..9fef148 --- /dev/null +++ b/test/wait-for-condition.spec.ts @@ -0,0 +1,49 @@ +import assert from 'node:assert/strict'; +import {describe, it} from 'node:test'; + +import {waitForCondition} from '../lib/wait-for-condition.js'; +import {recordTimeoutDelays} from './helpers.js'; + +describe('waitForCondition', function () { + it('should wait and succeed', async function (t) { + const delays = recordTimeoutDelays(t); + let calls = 0; + const result = await waitForCondition(() => ++calls >= 5 && 'done', {waitMs: 5000, intervalMs: 10}); + assert.strictEqual(result, 'done'); + assert.strictEqual(calls, 5); + assert.deepStrictEqual( + delays.filter((ms) => ms <= 10), + [10, 10, 10, 10], + ); + }); + it('should wait and fail', async function () { + await assert.rejects( + waitForCondition(() => false, {waitMs: 50, intervalMs: 10}), + /Condition unmet/, + ); + }); + it('should throw the provided error message or instance', async function () { + await assert.rejects( + waitForCondition(() => false, {waitMs: 20, intervalMs: 10, error: 'custom'}), + { + message: 'custom', + }, + ); + const error = new RangeError('instance'); + await assert.rejects( + waitForCondition(() => false, {waitMs: 20, intervalMs: 10, error}), + error, + ); + }); + it('should reduce interval to not exceed timeout', async function (t) { + const delays = recordTimeoutDelays(t); + await assert.rejects( + waitForCondition(() => false, {waitMs: 60, intervalMs: 50}), + /Condition unmet/, + ); + const sleeps = delays.filter((ms) => ms <= 50); + assert.strictEqual(sleeps[0], 50); + // only the time left is slept on the next spin + assert.ok(sleeps[1] < 50); + }); +}); diff --git a/test/with-timeout.spec.ts b/test/with-timeout.spec.ts new file mode 100644 index 0000000..a56a385 --- /dev/null +++ b/test/with-timeout.spec.ts @@ -0,0 +1,63 @@ +import assert from 'node:assert/strict'; +import {describe, it} from 'node:test'; + +import {TimeoutError} from '../lib/errors.js'; +import {sleep} from '../lib/sleep.js'; +import {withTimeout} from '../lib/with-timeout.js'; + +describe('withTimeout', function () { + function neverSettles(): Promise { + return new Promise(() => {}); + } + + it('should resolve when the promise settles before the deadline', async function () { + const result = await withTimeout(Promise.resolve(42), 1000); + assert.strictEqual(result, 42); + }); + it('should reject with TimeoutError when the deadline is exceeded', async function () { + await assert.rejects(withTimeout(neverSettles(), 30), TimeoutError); + }); + it('should use the default TimeoutError message when none is provided', async function () { + const timeoutMs = 20; + try { + await withTimeout(neverSettles(), timeoutMs); + assert.fail('expected rejection'); + } catch (err: unknown) { + assert.ok(err instanceof TimeoutError); + assert.strictEqual((err as TimeoutError).message, `Operation timed out after ${timeoutMs}ms`); + } + }); + it('should use a custom message on TimeoutError when provided', async function () { + try { + await withTimeout(neverSettles(), 20, 'custom timeout'); + assert.fail('expected rejection'); + } catch (err: unknown) { + assert.ok(err instanceof TimeoutError); + assert.strictEqual((err as TimeoutError).message, 'custom timeout'); + } + }); + it('should reject with a provided Error instance on timeout', async function () { + class CustomTimeout extends Error { + constructor(message?: string) { + super(message ?? 'custom default'); + this.name = 'CustomTimeout'; + } + } + const customErr = new CustomTimeout('overridden'); + try { + await withTimeout(neverSettles(), 20, customErr); + assert.fail('expected rejection'); + } catch (err: unknown) { + assert.strictEqual(err, customErr); + assert.ok(err instanceof CustomTimeout); + assert.strictEqual((err as CustomTimeout).message, 'overridden'); + } + }); + it('should propagate rejection from the underlying promise before the deadline', async function () { + const failing = (async () => { + await sleep(10); + throw new Error('boom'); + })(); + await assert.rejects(withTimeout(failing, 1000), /boom/); + }); +});