Big News: Socket raises $60M Series C at a $1B valuation to secure software supply chains for AI-driven development.Announcement →
Sign In

p-queue

Package Overview
Dependencies
Maintainers
2
Versions
53
Alerts
File Explorer

Advanced tools

License
Socket logo

Install Socket

Detect and block malicious and high-risk dependencies

Install

p-queue - npm Package Compare versions

Comparing version
8.1.1
 to
9.0.0
+144
-11
dist/index.d.ts

@@ -6,3 +6,3 @@ import { EventEmitter } from 'eventemitter3';

type Task<TaskResultType> = ((options: TaskOptions) => PromiseLike<TaskResultType>) | ((options: TaskOptions) => TaskResultType);
type EventName = 'active' | 'idle' | 'empty' | 'add' | 'next' | 'completed' | 'error';
type EventName = 'active' | 'idle' | 'empty' | 'add' | 'next' | 'completed' | 'error' | 'pendingZero' | 'rateLimit' | 'rateLimitCleared';
/**

@@ -14,5 +14,15 @@ Promise queue with concurrency control.

/**
Per-operation timeout in milliseconds. Operations fulfill once `timeout` elapses if they haven't already.
Get or set the default timeout for all tasks. Can be changed at runtime.
Applies to each future operation.
Operations will throw a `TimeoutError` if they don't complete within the specified time.
The timeout begins when the operation is dequeued and starts execution, not while it's waiting in the queue.
@example
```
const queue = new PQueue({timeout: 5000});
// Change timeout for all future tasks
queue.timeout = 10000;
```
*/

@@ -63,6 +73,3 @@ timeout?: number;

*/
add<TaskResultType>(function_: Task<TaskResultType>, options: {
throwOnTimeout: true;
} & Exclude<EnqueueOptionsType, 'throwOnTimeout'>): Promise<TaskResultType>;
add<TaskResultType>(function_: Task<TaskResultType>, options?: Partial<EnqueueOptionsType>): Promise<TaskResultType | void>;
add<TaskResultType>(function_: Task<TaskResultType>, options?: Partial<EnqueueOptionsType>): Promise<TaskResultType>;
/**

@@ -73,6 +80,3 @@ Same as `.add()`, but accepts an array of sync or async functions.

*/
addAll<TaskResultsType>(functions: ReadonlyArray<Task<TaskResultsType>>, options?: {
throwOnTimeout: true;
} & Partial<Exclude<EnqueueOptionsType, 'throwOnTimeout'>>): Promise<TaskResultsType[]>;
addAll<TaskResultsType>(functions: ReadonlyArray<Task<TaskResultsType>>, options?: Partial<EnqueueOptionsType>): Promise<Array<TaskResultsType | void>>;
addAll<TaskResultsType>(functions: ReadonlyArray<Task<TaskResultsType>>, options?: Partial<EnqueueOptionsType>): Promise<TaskResultsType[]>;
/**

@@ -111,2 +115,46 @@ Start (or resume) executing enqueued tasks within concurrency limit. No need to call this if queue is not paused (via `options.autoStart = false` or by `.pause()` method.)

/**
The difference with `.onIdle` is that `.onPendingZero` only waits for currently running tasks to finish, ignoring queued tasks.
@returns A promise that settles when all currently running tasks have completed; `queue.pending === 0`.
*/
onPendingZero(): Promise<void>;
/**
@returns A promise that settles when the queue becomes rate-limited due to intervalCap.
*/
onRateLimit(): Promise<void>;
/**
@returns A promise that settles when the queue is no longer rate-limited.
*/
onRateLimitCleared(): Promise<void>;
/**
@returns A promise that rejects when any task in the queue errors.
Use with `Promise.race([queue.onError(), queue.onIdle()])` to fail fast on the first error while still resolving normally when the queue goes idle.
Important: The promise returned by `add()` still rejects. You must handle each `add()` promise (for example, `.catch(() => {})`) to avoid unhandled rejections.
@example
```
import PQueue from 'p-queue';
const queue = new PQueue({concurrency: 2});
queue.add(() => fetchData(1)).catch(() => {});
queue.add(() => fetchData(2)).catch(() => {});
queue.add(() => fetchData(3)).catch(() => {});
// Stop processing on first error
try {
await Promise.race([
queue.onError(),
queue.onIdle()
]);
} catch (error) {
queue.pause(); // Stop processing remaining tasks
console.error('Queue failed:', error);
}
```
*/
onError(): Promise<never>;
/**
Size of the queue, the number of queued items waiting to run.

@@ -129,4 +177,89 @@ */

get isPaused(): boolean;
/**
Whether the queue is currently rate-limited due to intervalCap.
*/
get isRateLimited(): boolean;
/**
Whether the queue is saturated. Returns `true` when:
- All concurrency slots are occupied and tasks are waiting, OR
- The queue is rate-limited and tasks are waiting
Useful for detecting backpressure and potential hanging tasks.
```js
import PQueue from 'p-queue';
const queue = new PQueue({concurrency: 2});
// Backpressure handling
if (queue.isSaturated) {
console.log('Queue is saturated, waiting for capacity...');
await queue.onSizeLessThan(queue.concurrency);
}
// Monitoring for stuck tasks
setInterval(() => {
if (queue.isSaturated) {
console.warn(`Queue saturated: ${queue.pending} running, ${queue.size} waiting`);
}
}, 60000);
```
*/
get isSaturated(): boolean;
/**
The tasks currently being executed. Each task includes its `id`, `priority`, `startTime`, and `timeout` (if set).
Returns an array of task info objects.
```js
import PQueue from 'p-queue';
const queue = new PQueue({concurrency: 2});
// Add tasks with IDs for better debugging
queue.add(() => fetchUser(123), {id: 'user-123'});
queue.add(() => fetchPosts(456), {id: 'posts-456', priority: 1});
// Check what's running
console.log(queue.runningTasks);
// => [{
// id: 'user-123',
// priority: 0,
// startTime: 1759253001716,
// timeout: undefined
// }, {
// id: 'posts-456',
// priority: 1,
// startTime: 1759253001916,
// timeout: undefined
// }]
```
*/
get runningTasks(): ReadonlyArray<{
readonly id?: string;
readonly priority: number;
readonly startTime: number;
readonly timeout?: number;
}>;
}
export type { Queue } from './queue.js';
export { type QueueAddOptions, type Options } from './options.js';
/**
Error thrown when a task times out.
@example
```
import PQueue, {TimeoutError} from 'p-queue';
const queue = new PQueue({timeout: 1000});
try {
await queue.add(() => someTask());
} catch (error) {
if (error instanceof TimeoutError) {
console.log('Task timed out');
}
}
```
*/
export { TimeoutError } from 'p-timeout';
+312
-38
dist/index.js
import { EventEmitter } from 'eventemitter3';
import pTimeout, { TimeoutError } from 'p-timeout';
import pTimeout from 'p-timeout';
import PriorityQueue from './priority-queue.js';

@@ -8,8 +8,11 @@ /**

export default class PQueue extends EventEmitter {
#carryoverConcurrencyCount;
#carryoverIntervalCount;
#isIntervalIgnored;
#intervalCount = 0;
#intervalCap;
#rateLimitedInInterval = false;
#rateLimitFlushScheduled = false;
#interval;
#intervalEnd = 0;
#lastExecutionTime = 0;
#intervalId;

@@ -23,12 +26,22 @@ #timeoutId;

#isPaused;
#throwOnTimeout;
// Use to assign a unique identifier to a promise function, if not explicitly specified
#idAssigner = 1n;
// Track currently running tasks for debugging
#runningTasks = new Map();
/**
Per-operation timeout in milliseconds. Operations fulfill once `timeout` elapses if they haven't already.
Get or set the default timeout for all tasks. Can be changed at runtime.
Applies to each future operation.
Operations will throw a `TimeoutError` if they don't complete within the specified time.
The timeout begins when the operation is dequeued and starts execution, not while it's waiting in the queue.
@example
```
const queue = new PQueue({timeout: 5000});
// Change timeout for all future tasks
queue.timeout = 10000;
```
*/
timeout;
// TODO: The `throwOnTimeout` option should affect the return types of `add()` and `addAll()`
constructor(options) {

@@ -38,3 +51,3 @@ super();

options = {
carryoverConcurrencyCount: false,
carryoverIntervalCount: false,
intervalCap: Number.POSITIVE_INFINITY,

@@ -53,3 +66,5 @@ interval: 0,

}
this.#carryoverConcurrencyCount = options.carryoverConcurrencyCount;
// TODO: Remove this fallback in the next major version
// eslint-disable-next-line @typescript-eslint/no-deprecated
this.#carryoverIntervalCount = options.carryoverIntervalCount ?? options.carryoverConcurrencyCount ?? false;
this.#isIntervalIgnored = options.intervalCap === Number.POSITIVE_INFINITY || options.interval === 0;

@@ -61,5 +76,8 @@ this.#intervalCap = options.intervalCap;

this.concurrency = options.concurrency;
if (options.timeout !== undefined && !(Number.isFinite(options.timeout) && options.timeout > 0)) {
throw new TypeError(`Expected \`timeout\` to be a positive finite number, got \`${options.timeout}\` (${typeof options.timeout})`);
}
this.timeout = options.timeout;
this.#throwOnTimeout = options.throwOnTimeout === true;
this.#isPaused = options.autoStart === false;
this.#setupRateLimitTracking();
}

@@ -74,2 +92,5 @@ get #doesIntervalAllowAnother() {

this.#pending--;
if (this.#pending === 0) {
this.emit('pendingZero');
}
this.#tryToStartAnother();

@@ -79,3 +100,3 @@ this.emit('next');

#onResumeInterval() {
this.#onInterval();
this.#onInterval(); // Already schedules update
this.#initializeIntervalIfNeeded();

@@ -89,13 +110,19 @@ this.#timeoutId = undefined;

if (delay < 0) {
// Act as the interval was done
// We don't need to resume it here because it will be resumed on line 160
this.#intervalCount = (this.#carryoverConcurrencyCount) ? this.#pending : 0;
// If the interval has expired while idle, check if we should enforce the interval
// from the last task execution. This ensures proper spacing between tasks even
// when the queue becomes empty and then new tasks are added.
if (this.#lastExecutionTime > 0) {
const timeSinceLastExecution = now - this.#lastExecutionTime;
if (timeSinceLastExecution < this.#interval) {
// Not enough time has passed since the last task execution
this.#createIntervalTimeout(this.#interval - timeSinceLastExecution);
return true;
}
}
// Enough time has passed or no previous execution, allow execution
this.#intervalCount = (this.#carryoverIntervalCount) ? this.#pending : 0;
}
else {
// Act as the interval is pending
if (this.#timeoutId === undefined) {
this.#timeoutId = setTimeout(() => {
this.#onResumeInterval();
}, delay);
}
this.#createIntervalTimeout(delay);
return true;

@@ -106,2 +133,22 @@ }

}
#createIntervalTimeout(delay) {
if (this.#timeoutId !== undefined) {
return;
}
this.#timeoutId = setTimeout(() => {
this.#onResumeInterval();
}, delay);
}
#clearIntervalTimer() {
if (this.#intervalId) {
clearInterval(this.#intervalId);
this.#intervalId = undefined;
}
}
#clearTimeoutTimer() {
if (this.#timeoutId) {
clearTimeout(this.#timeoutId);
this.#timeoutId = undefined;
}
}
#tryToStartAnother() {

@@ -111,8 +158,7 @@ if (this.#queue.size === 0) {

// Because we can redo it later ("resume")
if (this.#intervalId) {
clearInterval(this.#intervalId);
}
this.#intervalId = undefined;
this.#clearIntervalTimer();
this.emit('empty');
if (this.#pending === 0) {
// Clear timeout as well when completely idle
this.#clearTimeoutTimer();
this.emit('idle');

@@ -122,2 +168,3 @@ }

}
let taskStarted = false;
if (!this.#isPaused) {

@@ -127,6 +174,9 @@ const canInitializeInterval = !this.#isIntervalPaused;

const job = this.#queue.dequeue();
if (!job) {
return false;
// Increment interval count immediately to prevent race conditions
if (!this.#isIntervalIgnored) {
this.#intervalCount++;
this.#scheduleRateLimitUpdate();
}
this.emit('active');
this.#lastExecutionTime = Date.now();
job();

@@ -136,6 +186,6 @@ if (canInitializeInterval) {

}
return true;
taskStarted = true;
}
}
return false;
return taskStarted;
}

@@ -153,7 +203,7 @@ #initializeIntervalIfNeeded() {

if (this.#intervalCount === 0 && this.#pending === 0 && this.#intervalId) {
clearInterval(this.#intervalId);
this.#intervalId = undefined;
this.#clearIntervalTimer();
}
this.#intervalCount = this.#carryoverConcurrencyCount ? this.#pending : 0;
this.#intervalCount = this.#carryoverIntervalCount ? this.#pending : 0;
this.#processQueue();
this.#scheduleRateLimitUpdate();
}

@@ -221,2 +271,5 @@ /**

setPriority(id, priority) {
if (typeof priority !== 'number' || !Number.isFinite(priority)) {
throw new TypeError(`Expected \`priority\` to be a finite number, got \`${priority}\` (${typeof priority})`);
}
this.#queue.setPriority(id, priority);

@@ -229,14 +282,37 @@ }

timeout: this.timeout,
throwOnTimeout: this.#throwOnTimeout,
...options,
};
return new Promise((resolve, reject) => {
// Create a unique symbol for tracking this task
const taskSymbol = Symbol(`task-${options.id}`);
this.#queue.enqueue(async () => {
this.#pending++;
// Track this running task
this.#runningTasks.set(taskSymbol, {
id: options.id,
priority: options.priority ?? 0, // Match priority-queue default
startTime: Date.now(),
timeout: options.timeout,
});
try {
options.signal?.throwIfAborted();
this.#intervalCount++;
// Check abort signal - if aborted, need to decrement the counter
// that was incremented in tryToStartAnother
try {
options.signal?.throwIfAborted();
}
catch (error) {
// Decrement the counter that was already incremented
if (!this.#isIntervalIgnored) {
this.#intervalCount--;
}
// Clean up tracking before throwing
this.#runningTasks.delete(taskSymbol);
throw error;
}
let operation = function_({ signal: options.signal });
if (options.timeout) {
operation = pTimeout(Promise.resolve(operation), { milliseconds: options.timeout });
operation = pTimeout(Promise.resolve(operation), {
milliseconds: options.timeout,
message: `Task timed out after ${options.timeout}ms (queue has ${this.#pending} running, ${this.#queue.size} waiting)`,
});
}

@@ -251,6 +327,2 @@ if (options.signal) {

catch (error) {
if (error instanceof TimeoutError && !options.throwOnTimeout) {
resolve();
return;
}
reject(error);

@@ -260,3 +332,8 @@ this.emit('error', error);

finally {
this.#next();
// Remove from running tasks
this.#runningTasks.delete(taskSymbol);
// Use queueMicrotask to prevent deep recursion while maintaining timing
queueMicrotask(() => {
this.#next();
});
}

@@ -293,2 +370,6 @@ }, options);

this.#queue = new this.#queueClass();
// Note: We don't clear #runningTasks as those tasks are still running
// They will be removed when they complete in the finally block
// Force synchronous update since clear() should have immediate effect
this.#updateRateLimitState();
}

@@ -333,2 +414,70 @@ /**

}
/**
The difference with `.onIdle` is that `.onPendingZero` only waits for currently running tasks to finish, ignoring queued tasks.
@returns A promise that settles when all currently running tasks have completed; `queue.pending === 0`.
*/
async onPendingZero() {
if (this.#pending === 0) {
return;
}
await this.#onEvent('pendingZero');
}
/**
@returns A promise that settles when the queue becomes rate-limited due to intervalCap.
*/
async onRateLimit() {
if (this.isRateLimited) {
return;
}
await this.#onEvent('rateLimit');
}
/**
@returns A promise that settles when the queue is no longer rate-limited.
*/
async onRateLimitCleared() {
if (!this.isRateLimited) {
return;
}
await this.#onEvent('rateLimitCleared');
}
/**
@returns A promise that rejects when any task in the queue errors.
Use with `Promise.race([queue.onError(), queue.onIdle()])` to fail fast on the first error while still resolving normally when the queue goes idle.
Important: The promise returned by `add()` still rejects. You must handle each `add()` promise (for example, `.catch(() => {})`) to avoid unhandled rejections.
@example
```
import PQueue from 'p-queue';
const queue = new PQueue({concurrency: 2});
queue.add(() => fetchData(1)).catch(() => {});
queue.add(() => fetchData(2)).catch(() => {});
queue.add(() => fetchData(3)).catch(() => {});
// Stop processing on first error
try {
await Promise.race([
queue.onError(),
queue.onIdle()
]);
} catch (error) {
queue.pause(); // Stop processing remaining tasks
console.error('Queue failed:', error);
}
```
*/
// eslint-disable-next-line @typescript-eslint/promise-function-async
async onError() {
return new Promise((_resolve, reject) => {
const handleError = (error) => {
this.off('error', handleError);
reject(error);
};
this.on('error', handleError);
});
}
async #onEvent(event, filter) {

@@ -373,2 +522,127 @@ return new Promise(resolve => {

}
#setupRateLimitTracking() {
// Only schedule updates when rate limiting is enabled
if (this.#isIntervalIgnored) {
return;
}
// Wire up to lifecycle events that affect rate limit state
// Only 'add' and 'next' can actually change rate limit state
this.on('add', () => {
if (this.#queue.size > 0) {
this.#scheduleRateLimitUpdate();
}
});
this.on('next', () => {
this.#scheduleRateLimitUpdate();
});
}
#scheduleRateLimitUpdate() {
// Skip if rate limiting is not enabled or already scheduled
if (this.#isIntervalIgnored || this.#rateLimitFlushScheduled) {
return;
}
this.#rateLimitFlushScheduled = true;
queueMicrotask(() => {
this.#rateLimitFlushScheduled = false;
this.#updateRateLimitState();
});
}
#updateRateLimitState() {
const previous = this.#rateLimitedInInterval;
const shouldBeRateLimited = !this.#isIntervalIgnored
&& this.#intervalCount >= this.#intervalCap
&& this.#queue.size > 0;
if (shouldBeRateLimited !== previous) {
this.#rateLimitedInInterval = shouldBeRateLimited;
this.emit(shouldBeRateLimited ? 'rateLimit' : 'rateLimitCleared');
}
}
/**
Whether the queue is currently rate-limited due to intervalCap.
*/
get isRateLimited() {
return this.#rateLimitedInInterval;
}
/**
Whether the queue is saturated. Returns `true` when:
- All concurrency slots are occupied and tasks are waiting, OR
- The queue is rate-limited and tasks are waiting
Useful for detecting backpressure and potential hanging tasks.
```js
import PQueue from 'p-queue';
const queue = new PQueue({concurrency: 2});
// Backpressure handling
if (queue.isSaturated) {
console.log('Queue is saturated, waiting for capacity...');
await queue.onSizeLessThan(queue.concurrency);
}
// Monitoring for stuck tasks
setInterval(() => {
if (queue.isSaturated) {
console.warn(`Queue saturated: ${queue.pending} running, ${queue.size} waiting`);
}
}, 60000);
```
*/
get isSaturated() {
return (this.#pending === this.#concurrency && this.#queue.size > 0)
|| (this.isRateLimited && this.#queue.size > 0);
}
/**
The tasks currently being executed. Each task includes its `id`, `priority`, `startTime`, and `timeout` (if set).
Returns an array of task info objects.
```js
import PQueue from 'p-queue';
const queue = new PQueue({concurrency: 2});
// Add tasks with IDs for better debugging
queue.add(() => fetchUser(123), {id: 'user-123'});
queue.add(() => fetchPosts(456), {id: 'posts-456', priority: 1});
// Check what's running
console.log(queue.runningTasks);
// => [{
// id: 'user-123',
// priority: 0,
// startTime: 1759253001716,
// timeout: undefined
// }, {
// id: 'posts-456',
// priority: 1,
// startTime: 1759253001916,
// timeout: undefined
// }]
```
*/
get runningTasks() {
// Return fresh array with fresh objects to prevent mutations
return [...this.#runningTasks.values()].map(task => ({ ...task }));
}
}
/**
Error thrown when a task times out.
@example
```
import PQueue, {TimeoutError} from 'p-queue';
const queue = new PQueue({timeout: 1000});
try {
await queue.add(() => someTask());
} catch (error) {
if (error instanceof TimeoutError) {
console.log('Task timed out');
}
}
```
*/
export { TimeoutError } from 'p-timeout';
+22
-7
dist/options.d.ts
import { type Queue, type RunFunction } from './queue.js';
type TimeoutOptions = {
/**
Per-operation timeout in milliseconds. Operations fulfill once `timeout` elapses if they haven't already.
Per-operation timeout in milliseconds. Operations will throw a `TimeoutError` if they don't complete within the specified time.
The timeout begins when the operation is dequeued and starts execution, not while it's waiting in the queue.
@default undefined
Can be overridden per task using the `timeout` option in `.add()`:
@example
```
const queue = new PQueue({timeout: 5000});
// This task uses the global 5s timeout
await queue.add(() => fetchData());
// This task has a 10s timeout
await queue.add(() => slowTask(), {timeout: 10000});
```
*/
timeout?: number;
/**
Whether or not a timeout is considered an exception.
@default false
*/
throwOnTimeout?: boolean;
};

@@ -54,2 +65,6 @@ export type Options<QueueType extends Queue<RunFunction, QueueOptions>, QueueOptions extends QueueAddOptions> = {

*/
readonly carryoverIntervalCount?: boolean;
/**
@deprecated Renamed to `carryoverIntervalCount`.
*/
readonly carryoverConcurrencyCount?: boolean;

@@ -56,0 +71,0 @@ } & TimeoutOptions;

+4
-7
dist/priority-queue.js

@@ -5,12 +5,9 @@ import lowerBound from './lower-bound.js';

enqueue(run, options) {
options = {
priority: 0,
...options,
};
const { priority = 0, id, } = options ?? {};
const element = {
priority: options.priority,
id: options.id,
priority,
id,
run,
};
if (this.size === 0 || this.#queue[this.size - 1].priority >= options.priority) {
if (this.size === 0 || this.#queue[this.size - 1].priority >= priority) {
this.#queue.push(element);

@@ -17,0 +14,0 @@ return;

+19
-27
package.json
{
"name": "p-queue",
"version": "8.1.1",
"version": "9.0.0",
"description": "Promise queue with concurrency control",

@@ -15,7 +15,7 @@ "license": "MIT",

"engines": {
"node": ">=18"
"node": ">=20"
},
"scripts": {
"build": "del-cli dist && tsc",
"test": "xo && ava && del-cli dist && tsc && tsd",
"test": "xo && node --import=tsx/esm --test test/*.ts && del-cli dist && tsc && tsd",
"bench": "node --import=tsx/esm bench.ts",

@@ -52,33 +52,20 @@ "prepublishOnly": "del-cli dist && tsc"

"eventemitter3": "^5.0.1",
"p-timeout": "^6.1.2"
"p-timeout": "^7.0.0"
},
"devDependencies": {
"@sindresorhus/tsconfig": "^5.0.0",
"@sindresorhus/tsconfig": "^8.0.1",
"@types/benchmark": "^2.1.5",
"@types/node": "^20.10.4",
"ava": "^5.3.1",
"@types/node": "^24.5.1",
"benchmark": "^2.1.4",
"del-cli": "^5.1.0",
"del-cli": "^6.0.0",
"delay": "^6.0.0",
"in-range": "^3.0.0",
"p-defer": "^4.0.0",
"random-int": "^3.0.0",
"p-defer": "^4.0.1",
"random-int": "^3.1.0",
"time-span": "^5.1.0",
"tsd": "^0.29.0",
"tsx": "^4.6.2",
"typescript": "^5.3.3",
"xo": "^0.56.0"
"tsd": "^0.33.0",
"tsx": "^4.20.5",
"typescript": "^5.9.2",
"xo": "^1.2.2"
},
"ava": {
"workerThreads": false,
"files": [
"test/**"
],
"extensions": {
"ts": "module"
},
"nodeArguments": [
"--import=tsx/esm"
]
},
"xo": {

@@ -88,5 +75,10 @@ "rules": {

"@typescript-eslint/no-floating-promises": "off",
"@typescript-eslint/no-invalid-void-type": "off"
"@typescript-eslint/no-unsafe-call": "off",
"@typescript-eslint/no-unsafe-assignment": "off",
"@typescript-eslint/no-unsafe-return": "off",
"@typescript-eslint/no-unsafe-argument": "off",
"@typescript-eslint/prefer-promise-reject-errors": "off",
"ava/no-import-test-files": "off"
}
}
}
+467
-13
readme.md

@@ -49,3 +49,3 @@ # p-queue

Here we run only one promise at the time. For example, set `concurrency` to 4 to run four promises at the same time.
Here we run only one promise at a time. For example, set `concurrency` to 4 to run four promises at the same time.

@@ -89,13 +89,21 @@ ```js

Type: `number`
Type: `number`\
Default: `undefined`
Per-operation timeout in milliseconds. Operations fulfill once `timeout` elapses if they haven't already.
Per-operation timeout in milliseconds. Operations will throw a `TimeoutError` if they don't complete within the specified time.
##### throwOnTimeout
The timeout begins when the operation is dequeued and starts execution, not while it's waiting in the queue.
Type: `boolean`\
Default: `false`
Can be overridden per task using the `timeout` option in `.add()`:
Whether or not a timeout is considered an exception.
```js
const queue = new PQueue({timeout: 5000});
// This task uses the global 5s timeout
await queue.add(() => fetchData());
// This task has a 10s timeout
await queue.add(() => slowTask(), {timeout: 10000});
```
##### autoStart

@@ -130,3 +138,3 @@

##### carryoverConcurrencyCount
##### carryoverIntervalCount

@@ -146,6 +154,10 @@ Type: `boolean`\

Returns a promise with the return value of `fn`.
Returns a promise that settles when the task completes, not when it's added to the queue. The promise resolves with the return value of `fn`.
Note: If your items can potentially throw an exception, you must handle those errors from the returned Promise or they may be reported as an unhandled Promise rejection and potentially cause your process to exit immediately.
> [!IMPORTANT]
> If you `await` this promise, you will wait for the task to finish running, which may defeat the purpose of using a queue for concurrency. See the [Usage](#usage) section for examples.
> [!NOTE]
> If your items can potentially throw an exception, you must handle those errors from the returned Promise or they may be reported as an unhandled Promise rejection and potentially cause your process to exit immediately.
##### fn

@@ -229,2 +241,5 @@

> [!NOTE]
> The promise returned by `.onEmpty()` resolves **once** when the queue becomes empty. If you want to be notified every time the queue becomes empty, use the `empty` event instead: `queue.on('empty', () => {})`.
#### .onIdle()

@@ -236,2 +251,77 @@

> [!NOTE]
> The promise returned by `.onIdle()` resolves **once** when the queue becomes idle. If you want to be notified every time the queue becomes idle, use the `idle` event instead: `queue.on('idle', () => {})`.
#### .onPendingZero()
Returns a promise that settles when all currently running tasks have completed; `queue.pending === 0`.
The difference with `.onIdle` is that `.onPendingZero` only waits for currently running tasks to finish, ignoring queued tasks. This is useful when you want to drain in-flight tasks before mutating shared state.
```js
queue.pause();
await queue.onPendingZero();
// All running tasks have finished, though the queue may still have items
```
#### .onRateLimit()
Returns a promise that settles when the queue becomes rate-limited due to `intervalCap`. If the queue is already rate-limited, the promise resolves immediately.
Useful for implementing backpressure to prevent memory issues when producers are faster than consumers.
```js
const queue = new PQueue({intervalCap: 5, interval: 1000});
// Add many tasks
for (let index = 0; index < 10; index++) {
queue.add(() => someTask());
}
await queue.onRateLimit();
console.log('Queue is now rate-limited - time for maintenance tasks');
```
#### .onRateLimitCleared()
Returns a promise that settles when the queue is no longer rate-limited. If the queue is not currently rate-limited, the promise resolves immediately.
```js
const queue = new PQueue({intervalCap: 5, interval: 1000});
// Wait for rate limiting to be cleared
await queue.onRateLimitCleared();
console.log('Rate limit cleared - can add more tasks');
```
#### .onError()
Returns a promise that rejects when any task in the queue errors.
Use with `Promise.race([queue.onError(), queue.onIdle()])` to fail fast on the first error while still resolving normally when the queue goes idle.
> [!IMPORTANT]
> The promise returned by `add()` still rejects. You must handle each `add()` promise (for example, `.catch(() => {})`) to avoid unhandled rejections.
```js
import PQueue from 'p-queue';
const queue = new PQueue({concurrency: 2});
queue.add(() => fetchData(1)).catch(() => {});
queue.add(() => fetchData(2)).catch(() => {});
queue.add(() => fetchData(3)).catch(() => {});
// Stop processing on first error
try {
await Promise.race([
queue.onError(),
queue.onIdle()
]);
} catch (error) {
queue.pause(); // Stop processing remaining tasks
console.error('Queue failed:', error);
}
```
#### .onSizeLessThan(limit)

@@ -319,2 +409,17 @@

Type: `number | undefined`
Get or set the default timeout for all tasks. Can be changed at runtime.
Operations will throw a `TimeoutError` if they don't complete within the specified time.
The timeout begins when the operation is dequeued and starts execution, not while it's waiting in the queue.
```js
const queue = new PQueue({timeout: 5000});
// Change timeout for all future tasks
queue.timeout = 10000;
```
#### [.concurrency](#concurrency)

@@ -326,2 +431,68 @@

#### .isRateLimited
Whether the queue is currently rate-limited due to `intervalCap`. Returns `true` when the number of tasks executed in the current interval has reached the `intervalCap` and there are still tasks waiting to be processed.
#### .isSaturated
Whether the queue is saturated. Returns `true` when:
- All concurrency slots are occupied and tasks are waiting, OR
- The queue is rate-limited and tasks are waiting
Useful for detecting backpressure and potential hanging tasks.
```js
import PQueue from 'p-queue';
const queue = new PQueue({concurrency: 2});
// Backpressure handling
if (queue.isSaturated) {
console.log('Queue is saturated, waiting for capacity...');
await queue.onSizeLessThan(queue.concurrency);
}
// Monitoring for stuck tasks
setInterval(() => {
if (queue.isSaturated) {
console.warn(`Queue saturated: ${queue.pending} running, ${queue.size} waiting`);
}
}, 60000);
```
#### .runningTasks
The tasks currently being executed. Each task includes its `id`, `priority`, `startTime`, and `timeout` (if set).
Returns an array of task info objects.
```js
import PQueue from 'p-queue';
const queue = new PQueue({concurrency: 2});
// Add tasks with IDs for better debugging
queue.add(() => fetchUser(123), {id: 'user-123'});
queue.add(() => fetchPosts(456), {id: 'posts-456', priority: 1});
// Check what's running
console.log(queue.runningTasks);
/*
[
{
id: 'user-123',
priority: 0,
startTime: 1759253001716,
timeout: undefined
},
{
id: 'posts-456',
priority: 1,
startTime: 1759253001916,
timeout: undefined
}
]
*/
```
## Events

@@ -370,6 +541,5 @@

Emitted if an item throws an error.
Emitted if an item throws an error. The promise returned by `add()` is still rejected, so you must handle both.
```js
import delay from 'delay';
import PQueue from 'p-queue';

@@ -383,3 +553,6 @@

queue.add(() => Promise.reject(new Error('error')));
// Handle the promise to prevent unhandled rejection
queue.add(() => Promise.reject(new Error('error'))).catch(() => {
// Error already handled by event listener
});
```

@@ -399,2 +572,8 @@

#### pendingZero
Emitted every time the number of running tasks becomes zero; `queue.pending === 0`.
The difference with `idle` is that `pendingZero` is emitted even when the queue still has items waiting to run, whereas `idle` requires both an empty queue and no pending tasks.
```js

@@ -458,2 +637,52 @@ import delay from 'delay';

#### rateLimit
Emitted when the queue becomes rate-limited due to `intervalCap`. This happens when the maximum number of tasks allowed per interval has been reached.
Useful for implementing backpressure to prevent memory issues when producers are faster than consumers.
```js
import delay from 'delay';
import PQueue from 'p-queue';
const queue = new PQueue({
intervalCap: 2,
interval: 1000
});
queue.on('rateLimit', () => {
console.log('Queue is rate-limited - processing backlog or maintenance tasks');
});
// Add 3 tasks - third one triggers rate limiting
queue.add(() => delay(100));
queue.add(() => delay(100));
queue.add(() => delay(100));
```
#### rateLimitCleared
Emitted when the queue is no longer rate-limited—either because the interval reset and new tasks can start, or because the backlog was drained.
```js
import delay from 'delay';
import PQueue from 'p-queue';
const queue = new PQueue({
intervalCap: 1,
interval: 1000
});
queue.on('rateLimit', () => {
console.log('Rate limited - waiting for interval to reset');
});
queue.on('rateLimitCleared', () => {
console.log('Rate limit cleared - can process more tasks');
});
queue.add(() => delay(100));
queue.add(() => delay(100)); // This triggers rate limiting
```
## Advanced example

@@ -529,2 +758,47 @@

## Handling timeouts
You can set a timeout for all tasks or override it per task. When a task times out, a `TimeoutError` is thrown.
```js
import PQueue, {TimeoutError} from 'p-queue';
// Set a global timeout for all tasks
const queue = new PQueue({
concurrency: 2,
timeout: 5000, // 5 seconds
});
(async () => {
// This task will use the global timeout
try {
await queue.add(() => fetchData());
} catch (error) {
if (error instanceof TimeoutError) {
console.log('Task timed out after 5 seconds');
}
}
})();
(async () => {
// Override timeout for a specific task
try {
await queue.add(() => slowTask(), {
timeout: 10000, // 10 seconds for this task only
});
} catch (error) {
if (error instanceof TimeoutError) {
console.log('Slow task timed out after 10 seconds');
}
}
})();
(async () => {
// No timeout for this task
await queue.add(() => verySlowTask(), {
timeout: undefined,
});
})();
```
## Custom QueueClass

@@ -570,2 +844,182 @@

#### How do I implement backpressure?
Use `.onSizeLessThan()` to prevent the queue from growing unbounded and causing memory issues when producers are faster than consumers:
```js
const queue = new PQueue();
// Wait for queue to have space before adding more
await queue.onSizeLessThan(100);
queue.add(() => someTask());
```
Note: `.size` counts queued items, while `.pending` counts running items. The total is `queue.size + queue.pending`.
You can also use `.onRateLimit()` for backpressure during rate limiting. See the [`.onRateLimit()`](#onratelimit) docs.
#### How do I cancel or remove a queued task?
Use `AbortSignal` for targeted cancellation. Aborting removes a waiting task and rejects the `.add()` promise. For bulk operations, use `queue.clear()` or share one `AbortController` across tasks.
```js
import PQueue from 'p-queue';
const queue = new PQueue();
const controller = new AbortController();
const promise = queue.add(({signal}) => doWork({signal}), {signal: controller.signal});
controller.abort(); // Cancels if still queued; running tasks must handle `signal` themselves
```
Direct removal methods are not provided as they would leak internals and risk dangling promises.
#### How do I get results in the order they were added?
This package executes tasks in priority order, but doesn't guarantee completion order. If you need results in the order they were added, use `Promise.all()`, which maintains the order of the input array:
```js
import PQueue from 'p-queue';
const queue = new PQueue({concurrency: 4});
const tasks = [
() => fetchData(1), // May finish third
() => fetchData(2), // May finish first
() => fetchData(3), // May finish second
];
const results = await Promise.all(
tasks.map(task => queue.add(task))
);
// results = [result1, result2, result3] ✅ Always in input order
// Or more concisely:
const urls = ['url1', 'url2', 'url3'];
const results = await Promise.all(
urls.map(url => queue.add(() => fetch(url)))
);
```
If you don't need `p-queue`'s advanced features, consider using [`p-map`](https://github.com/sindresorhus/p-map), which is specifically designed for this use case.
#### How do I stream results as they complete in order?
For progressive results that maintain input order, use [`pMapIterable`](https://github.com/sindresorhus/p-map#pmapiterable) from `p-map`:
```js
import {pMapIterable} from 'p-map';
// Stream results in order as they complete
for await (const result of pMapIterable(items, fetchItem, {concurrency: 4})) {
console.log(result); // Results arrive in input order
}
```
You can combine it with `p-queue` when you need priorities or a shared concurrency cap:
```js
import PQueue from 'p-queue';
import {pMapIterable} from 'p-map';
// Let p-queue handle concurrency
const queue = new PQueue({concurrency: 4});
for await (const result of pMapIterable(
items,
item => queue.add(() => fetchItem(item), {priority: item.priority})
)) {
console.log(result); // Still in input order
}
```
#### How do I debug a queue that stops processing tasks?
If your queue stops processing tasks after extended use, it's likely that some tasks are hanging indefinitely, exhausting the concurrency limit. Use the `.runningTasks` property to identify which specific tasks are stuck.
Common causes:
- Network requests without timeouts
- Database queries that hang
- File operations on unresponsive network drives
- Unhandled promise rejections
Debugging steps:
```js
// 1. Add timeouts to prevent hanging
const queue = new PQueue({
concurrency: 2,
timeout: 30000 // 30 seconds
});
// 2. Always add IDs to tasks for debugging
queue.add(() => processItem(item), {id: `item-${item.id}`});
// 3. Monitor for stuck tasks using runningTasks
setInterval(() => {
const now = Date.now();
const stuckTasks = queue.runningTasks.filter(task =>
now - task.startTime > 30000 // Running for over 30 seconds
);
if (stuckTasks.length > 0) {
console.error('Stuck tasks:', stuckTasks);
// Consider aborting or logging more details
}
// Detect saturation (potential hanging if persistent)
if (queue.isSaturated) {
console.warn(`Queue saturated: ${queue.pending} running, ${queue.size} waiting`);
}
}, 60000);
// 4. Track task lifecycle
queue.on('completed', result => {
console.log('Task completed');
});
queue.on('error', error => {
console.error('Task failed:', error);
});
// 5. Wrap tasks with debugging
const debugTask = async (fn, name) => {
const start = Date.now();
console.log(`Starting: ${name}`);
try {
const result = await fn();
console.log(`Completed: ${name} (${Date.now() - start}ms)`);
return result;
} catch (error) {
console.error(`Failed: ${name} (${Date.now() - start}ms)`, error);
throw error;
}
};
queue.add(() => debugTask(() => fetchData(), 'fetchData'), {id: 'fetchData'});
```
Prevention:
- Always use timeouts for I/O operations
- Ensure all async functions properly resolve or reject
- Use the `timeout` option to enforce task time limits
- Monitor `queue.size` and `queue.pending` in production
#### How do I test code that uses `p-queue` with Jest fake timers?
Jest fake timers don't work well with `p-queue` because it uses `queueMicrotask` internally.
Workaround:
```js
const flushPromises = () => new Promise(resolve => setImmediate(resolve));
jest.useFakeTimers();
// ... your test code ...
await jest.runAllTimersAsync();
await flushPromises();
```
## Maintainers

@@ -572,0 +1026,0 @@