Add onError() method

Fixes #29

Author: sindresorhus

readme.md

@@ -288,6 +288,36 @@ 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)
 
 Returns a promise that settles when the queue size is less than the given limit: `queue.size < limit`.

source/index.ts

@@ -547,6 +547,47 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 		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(): Promise<never> {
+		return new Promise<never>((_resolve, reject) => {
+			const handleError = (error: unknown) => {
+				this.off('error', handleError);
+				reject(error);
+			};
+
+			this.on('error', handleError);
+		});
+	}
+
 	async #onEvent(event: EventName, filter?: () => boolean): Promise<void> {
 		return new Promise(resolve => {
 			const listener = () => {

test/basic.ts

@@ -1100,3 +1100,125 @@ test('pause should work when throttled', async () => {
 
 	await delay(2500);
 });
+
+test('.onError() - rejects when task errors', async () => {
+	const queue = new PQueue({concurrency: 1});
+
+	queue.add(async () => delay(50));
+
+	// Set up onError listener first
+	const errorPromise = queue.onError();
+
+	// Add task that will fail
+	const task = queue.add(async () => {
+		throw new Error('Task failed');
+	});
+
+	await assert.rejects(
+		errorPromise,
+		{message: 'Task failed'},
+	);
+
+	// Clean up unhandled rejection
+	await Promise.allSettled([task]);
+});
+
+test('.onError() - captures first error in queue', async () => {
+	const queue = new PQueue({concurrency: 2});
+
+	// Set up onError listener first
+	const errorPromise = queue.onError();
+
+	// Add tasks
+	const task1 = queue.add(async () => {
+		await delay(100);
+		throw new Error('First error');
+	});
+
+	const task2 = queue.add(async () => {
+		await delay(200);
+		throw new Error('Second error');
+	});
+
+	// Set up cleanup
+	const cleanup = Promise.allSettled([task1, task2]);
+
+	// Wait for onError to reject with first error
+	await assert.rejects(
+		errorPromise,
+		{message: 'First error'},
+		'Should reject with first error',
+	);
+
+	// Ensure all tasks completed
+	await cleanup;
+});
+
+test('.onError() - works with Promise.race pattern', async () => {
+	const queue = new PQueue({concurrency: 2});
+
+	queue.add(async () => delay(50));
+	queue.add(async () => delay(100));
+
+	const racePromise = Promise.race([
+		queue.onError(),
+		queue.onIdle(),
+	]);
+
+	const task = queue.add(async () => {
+		throw new Error('Failed task');
+	});
+	queue.add(async () => delay(150));
+
+	// Stop processing on first error
+	try {
+		await racePromise;
+		assert.fail('Should have thrown error');
+	} catch (error) {
+		assert.equal((error as Error).message, 'Failed task');
+		queue.pause(); // Stop processing remaining tasks
+	}
+
+	// Clean up unhandled rejection
+	await Promise.allSettled([task]);
+
+	assert.equal(queue.isPaused, true);
+});
+
+test('.onError() - multiple listeners', async () => {
+	const queue = new PQueue({concurrency: 1});
+
+	queue.add(async () => delay(50));
+
+	const error1 = queue.onError();
+	const error2 = queue.onError();
+
+	const task = queue.add(async () => {
+		throw new Error('Task error');
+	});
+
+	await assert.rejects(error1, {message: 'Task error'});
+	await assert.rejects(error2, {message: 'Task error'});
+
+	// Clean up unhandled rejection
+	await Promise.allSettled([task]);
+});
+
+test('.onError() - works when called before adding tasks', async () => {
+	const queue = new PQueue({concurrency: 1});
+
+	// Call onError() before adding any tasks
+	const errorPromise = queue.onError();
+
+	// Add a task that errors
+	const task = queue.add(async () => {
+		throw new Error('Early error');
+	});
+
+	await Promise.allSettled([task]);
+
+	await assert.rejects(
+		errorPromise,
+		{message: 'Early error'},
+	);
+});