Fix: after first error, future .next() rethrows and no further input is pulled

Author: snatalenko

README.MD

@@ -7,7 +7,7 @@ Async Parallel Pipe
 [![Coverage Status](https://coveralls.io/repos/github/snatalenko/async-parallel-pipe/badge.svg?branch=main)](https://coveralls.io/github/snatalenko/async-parallel-pipe?branch=main)
 [![NPM Downloads](https://img.shields.io/npm/dm/async-parallel-pipe.svg)](https://www.npmjs.com/package/async-parallel-pipe)
 
-The function pulls input values from async iterable input, runs async action, and returns async iterable output in a same order as input was received.
+The function pulls input values from an iterable or async iterable input, runs an async action on each yielded value with a configurable concurrency limit, and returns an async iterable output in the same order as input was received. The final return value of the input iterator (if any) is forwarded as the terminal `{ done: true, value }` of the output without applying the action.
 
 Here is an example that requests random numbers from random.org with not more than 2 concurrent requests at a time:
 

src/parallelPipe.ts

@@ -18,6 +18,7 @@ export function parallelPipe<TInput, TOutput>(
 	const runningTasks: Array<Promise<IteratorResult<TOutput>> | IteratorResult<TOutput>> = [];
 	let inputDone = false;
 	let inputIndex = 0;
+	let iteratorFailure: unknown;
 
 	async function executeActionOnIterableInput(input: IteratorResult<TInput, any>): Promise<
 		{ done: false, value: TOutput } |
@@ -53,20 +54,30 @@ export function parallelPipe<TInput, TOutput>(
 
 	return {
 		async next() {
+			if (iteratorFailure)
+				throw iteratorFailure;
+
 			// start processing input once output is requested
 			if (!runningTasks.length)
 				pullFromInput();
 
-			if (!runningTasks.length)
+			const firstRunningTask = runningTasks.shift();
+			if (!firstRunningTask)
 				return { done: true, value: undefined };
 
-			const firstRunningTask = runningTasks.splice(0, 1)[0];
-			const { done, value } = await firstRunningTask;
+			try {
+				const { done, value } = await firstRunningTask;
 
-			if (!done)
-				pullFromInput();
+				if (!done)
+					pullFromInput();
 
-			return { done, value };
+				return { done, value };
+			}
+			catch (err: unknown) {
+				iteratorFailure = err; // switch to failed state
+				inputDone = true; // stop pulling new tasks
+				throw err;
+			}
 		},
 		[Symbol.asyncIterator]() {
 			return this;

tests/unit/parallelPipe.test.ts

@@ -81,6 +81,40 @@ describe('parallelPipe', () => {
 		expect(caughtError.message).toEqual('fail');
 	});
 
+	it('does not consume remaining input after an error', async () => {
+		const inputValues = [1, 2, 3, 4, 5];
+		let nextCalls = 0;
+		const input: AsyncIterableIterator<number> = {
+			next: async () => {
+				nextCalls++;
+				const value = inputValues[nextCalls - 1];
+				return value === undefined ? { done: true, value: undefined } : { done: false, value };
+			},
+			[Symbol.asyncIterator]() {
+				return this;
+			}
+		};
+
+		const action = async (x: number) => {
+			if (x === 2)
+				throw new Error('fail-2');
+
+			return x;
+		};
+
+		const iterator = parallelPipe(input, 2, action);
+
+		const res1 = await iterator.next();
+		expect(res1).toEqual({ value: 1, done: false });
+
+		await expect(iterator.next()).rejects.toThrow('fail-2');
+
+		// First `next()` fills the queue with `limit` items, then pulls one more after yielding.
+		// After the first rejection, no additional input should be consumed.
+		await Promise.resolve();
+		expect(nextCalls).toBe(3);
+	});
+
 	it('does not exceed the specified parallel limit', async () => {
 
 		const inputValues = [1, 2, 3, 4, 5];