Revert: stop failing iterator on first task error

Author: snatalenko

src/parallelPipe.ts

@@ -18,7 +18,6 @@ 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 } |
@@ -54,9 +53,6 @@ export function parallelPipe<TInput, TOutput>(
 
 	return {
 		async next() {
-			if (iteratorFailure)
-				throw iteratorFailure;
-
 			// start processing input once output is requested
 			if (!runningTasks.length)
 				pullFromInput();
@@ -65,19 +61,12 @@ export function parallelPipe<TInput, TOutput>(
 			if (!firstRunningTask)
 				return { done: true, value: undefined };
 
-			try {
-				const { done, value } = await firstRunningTask;
+			const { done, value } = await firstRunningTask;
 
-				if (!done)
-					pullFromInput();
+			if (!done)
+				pullFromInput();
 
-				return { done, value };
-			}
-			catch (err: unknown) {
-				iteratorFailure = err; // switch to failed state
-				inputDone = true; // stop pulling new tasks
-				throw err;
-			}
+			return { done, value };
 		},
 		[Symbol.asyncIterator]() {
 			return this;

tests/unit/parallelPipe.test.ts

@@ -78,41 +78,61 @@ describe('parallelPipe', () => {
 			caughtError = err;
 		}
 		expect(caughtError).toBeTruthy();
-		expect(caughtError.message).toEqual('fail');
+		expect(caughtError).toHaveProperty('message', '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;
-			}
-		};
+	it('propagates upstream pipe errors to downstream consumption', async () => {
+		const input = asyncGenerator([1, 2, 3]);
+		const iterator1 = parallelPipe(input, 1, async x => {
+			if (x === 2)
+				throw new Error('upstream-fail');
+
+			return x;
+		});
+
+		const action2 = jest.fn(async x => x * 10);
+		const iterator2 = parallelPipe(iterator1, 1, action2);
+
+		let caughtError;
+		try {
+			// eslint-disable-next-line @typescript-eslint/no-unused-vars
+			for await (const _ of iterator2) { /* no-op */ }
+		}
+		catch (err) {
+			caughtError = err;
+		}
+		expect(caughtError).toBeTruthy();
+		expect(caughtError).toHaveProperty('message', 'upstream-fail');
+		expect(action2).toHaveBeenCalledTimes(1);
+	});
 
-		const action = async (x: number) => {
+	it('continues processing after handled upstream error when consuming via .next()', async () => {
+		const input = asyncGenerator([1, 2, 3]);
+		const iterator1 = parallelPipe(input, 1, async x => {
 			if (x === 2)
-				throw new Error('fail-2');
+				throw new Error('upstream-fail');
 
 			return x;
-		};
+		});
 
-		const iterator = parallelPipe(input, 2, action);
+		const iterator2 = parallelPipe(iterator1, 1, async x => x * 10);
 
-		const res1 = await iterator.next();
-		expect(res1).toEqual({ value: 1, done: false });
+		const res1 = await iterator2.next();
+		expect(res1).toEqual({ value: 10, done: false });
 
-		await expect(iterator.next()).rejects.toThrow('fail-2');
+		let caughtError;
+		try {
+			await iterator2.next();
+		}
+		catch (err) {
+			caughtError = err;
+		}
+		expect(caughtError).toBeTruthy();
+		expect(caughtError).toHaveProperty('message', 'upstream-fail');
 
-		// 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);
+		// once the error is handled by the consumer, subsequent inputs can still be processed.
+		const res3 = await iterator2.next();
+		expect(res3).toEqual({ value: 30, done: false });
 	});
 
 	it('does not exceed the specified parallel limit', async () => {