From f86cbdc9ec91c396c3820a3c6227bc6c510aa925 Mon Sep 17 00:00:00 2001 From: hw Date: Sat, 8 Aug 2026 17:48:48 +0900 Subject: [PATCH 1/3] test: reproduce concurrentPool stall, over-prefetch, and post-error pulls Three failing regressions: - buffered items are not delivered once the pull chain drains - next() hangs on a buffered result while the following upstream fetch hangs - every consumer next() launches `length` new fetches regardless of buffered results, so a slow consumer grows the buffer unboundedly (produced == consumed x pool size) - the source is pulled again after it has thrown (the known TODO) Co-Authored-By: Claude Fable 5 --- test/Lazy/concurrentPool.spec.ts | 79 ++++++++++++++++++++++++++++++++ 1 file changed, 79 insertions(+) diff --git a/test/Lazy/concurrentPool.spec.ts b/test/Lazy/concurrentPool.spec.ts index b2f40eb1..cf165f47 100644 --- a/test/Lazy/concurrentPool.spec.ts +++ b/test/Lazy/concurrentPool.spec.ts @@ -115,3 +115,82 @@ describe("concurrentPool", function () { expect(acc).toEqual([1, 2, 3]); }, 2050); }); + +describe("concurrentPool regressions", function () { + const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + + it("should deliver buffered items even when the next upstream fetch hangs", async function () { + async function* fastThenHang() { + yield 1; + yield 2; + yield 3; + await new Promise(() => undefined); // upstream hangs on the 4th item + } + const iter = concurrentPool(3, fastThenHang())[Symbol.asyncIterator](); + + expect((await iter.next()).value).toBe(1); + // let every pending pull-chain continuation drain + await sleep(50); + // items 2 and 3 are already buffered - they must not wait for a new fetch + const result = await Promise.race([ + iter.next(), + sleep(300).then(() => "TIMEOUT" as const), + ]); + expect(result).toEqual({ value: 2, done: false }); + }); + + it("should keep prefetch bounded by the pool size with a slow consumer", async function () { + let produced = 0; + async function* fast() { + for (let i = 0; i < 1000; i++) { + produced++; + yield i; + } + } + const iter = concurrentPool(3, fast())[Symbol.asyncIterator](); + let consumed = 0; + for (let i = 0; i < 5; i++) { + await iter.next(); + consumed++; + await sleep(15); + } + // in-flight + buffered may run ahead of the consumer by at most the pool size + expect(produced).toBeLessThanOrEqual(consumed + 3); + }); + + it("should not pull the source again after it has thrown", async function () { + let pullsAfterThrow = 0; + let thrown = false; + const source: AsyncIterable = { + [Symbol.asyncIterator]() { + let n = 0; + return { + async next() { + if (thrown) { + pullsAfterThrow++; + return { value: undefined, done: true } as IteratorResult; + } + if (n >= 1) { + thrown = true; + throw new Error("boom"); + } + return { value: n++, done: false }; + }, + }; + }, + }; + const iter = concurrentPool(2, source)[Symbol.asyncIterator](); + const results = []; + try { + for (let i = 0; i < 5; i++) { + results.push(await iter.next()); + } + } catch (e) { + // first rejection is expected + } + await sleep(30); + await iter.next().catch(() => undefined); + await sleep(30); + expect(pullsAfterThrow).toBe(0); + }); +}); From 98a96fac4b5661f0ba064b350a6ea758c33310f7 Mon Sep 17 00:00:00 2001 From: hw Date: Sat, 8 Aug 2026 17:55:37 +0900 Subject: [PATCH 2/3] fix: deliver buffered results directly and bound idle prefetch in concurrentPool - next() now serves already-buffered results immediately instead of depending on a future fetch settling, fixing a hang where buffered items were never delivered once the pull chain drained while the next upstream fetch was pending - prefetch is capped at the pool size when no consumer is waiting, so a slow consumer no longer grows the buffer unboundedly (was consumed x pool-size upstream pulls); under active demand the pool stays saturated, preserving p-map-equivalent throughput (verified 455ms == 455ms on the uneven-duration benchmark) - the source is no longer pulled after it has thrown (resolves the long-standing TODO), and post-error consumers receive done once nothing is in flight and the buffer is drained Co-Authored-By: Claude Fable 5 --- src/Lazy/concurrentPool.ts | 31 ++++++++++++++++++++----------- 1 file changed, 20 insertions(+), 11 deletions(-) diff --git a/src/Lazy/concurrentPool.ts b/src/Lazy/concurrentPool.ts index 065aa45c..15fae9f7 100644 --- a/src/Lazy/concurrentPool.ts +++ b/src/Lazy/concurrentPool.ts @@ -69,7 +69,14 @@ function concurrentPool( let prev = Promise.resolve(); async function fetchNextItem() { - if (idleWorkers <= 0) { + if (idleWorkers <= 0 || finished) { + return; + } + // While a consumer is waiting, keep the pool saturated regardless of the + // buffer (order-preserving pools must buffer past a slow head item, as + // p-map does). Only when nobody is waiting is prefetch capped at the + // pool size, so a slow consumer cannot grow the buffer unboundedly. + if (consumer.length === 0 && itemMap.size >= idleWorkers) { return; } @@ -114,7 +121,7 @@ function concurrentPool( } } - async function pullItem(item: Item): Promise { + function deliverFromBuffer() { while (consumer.length > 0) { const id = consumer[0][2]; if (!itemMap.has(id)) { @@ -132,10 +139,11 @@ function concurrentPool( resolve(value.success); } } + } + + async function pullItem(item: Item): Promise { + deliverFromBuffer(); - /** - * @TODO ppeeou: If an error occurs, the next iterator should not be executed - */ if (!item.success.done && consumer.length > 0) { fetchNextItem(); } @@ -152,12 +160,10 @@ function concurrentPool( } function slide() { - if (finished && fillId === consumeId) { - while (consumer.length > 0) { - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - const [resolve] = consumer.shift()!; - resolve({ value: undefined, done: true }); - } + // After a failure, once nothing is in flight and the buffer is drained, + // remaining consumers can only ever receive `done`. + if (finished && idleWorkers === length && itemMap.size === 0) { + doneQueue(); } else { processQueue(); } @@ -174,6 +180,9 @@ function concurrentPool( const task: [Resolve, Reject, number] = [resolve, reject, id]; consumer.push(task); + // Serve already-buffered results immediately - delivery must not + // depend on a future fetch settling. + deliverFromBuffer(); slide(); }); }, From 4b2bfccd86c3b3dabbae27978f99bd740244e513 Mon Sep 17 00:00:00 2001 From: hw Date: Sat, 8 Aug 2026 17:58:25 +0900 Subject: [PATCH 3/3] test: drop the unused catch binding Co-Authored-By: Claude Fable 5 --- test/Lazy/concurrentPool.spec.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/Lazy/concurrentPool.spec.ts b/test/Lazy/concurrentPool.spec.ts index cf165f47..69973d5a 100644 --- a/test/Lazy/concurrentPool.spec.ts +++ b/test/Lazy/concurrentPool.spec.ts @@ -185,7 +185,7 @@ describe("concurrentPool regressions", function () { for (let i = 0; i < 5; i++) { results.push(await iter.next()); } - } catch (e) { + } catch { // first rejection is expected } await sleep(30);