From e1ce9a85949a63a06cb19b460cb4983bf566fda5 Mon Sep 17 00:00:00 2001 From: Vladimir Kutepov Date: Sun, 30 Aug 2026 18:45:30 +0700 Subject: [PATCH 1/5] fix(web): pull a streamed result behind a demand gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The stream was built with no `pull` and no queuing strategy, and every codec node is enqueued as soon as it is parsed, so the producer ran as fast as it could resolve whether or not anyone read. One slow consumer buffered the entire result in server memory — unbounded, and invisible to application code. The gate sits in the iterator wrapper the runtime already installs, which its own comment calls "the only seam where a dropped consumer can stop the producer": the same seam lets a SLOW consumer slow it. A read drives `pull`, `pull` releases one source pull. Teardown releases a parked pull too, or an aborted stream would hang on it. Measured with an async generator and no reads for 200 event-loop turns: the producer advanced by 1, where it previously tracked the turn count (and reached ~17,000 items, +19 MiB, in the wall-clock shape the issue reports). Reading 20 chunks advances it by 20, and cancelling stops it within the pull in flight. The #3112 marker for this gap comes off, per the convention the other two followed when they closed. --- .changeset/stream-demand-gate.md | 14 ++++++ packages/web/server-functions/src/server.ts | 25 +++++++++- .../server-functions-open-gaps.spec.tsx | 47 +++++++++++++++---- 3 files changed, 75 insertions(+), 11 deletions(-) create mode 100644 .changeset/stream-demand-gate.md diff --git a/.changeset/stream-demand-gate.md b/.changeset/stream-demand-gate.md new file mode 100644 index 000000000..4c830879c --- /dev/null +++ b/.changeset/stream-demand-gate.md @@ -0,0 +1,14 @@ +--- +"@solidjs/web": patch +--- + +Pull a streamed server-function result behind a demand gate (#3118). The +response stream was built with no `pull` and no queuing strategy, and +every codec node is enqueued the moment it is parsed, so the producer ran +as fast as it could resolve whether or not anyone was reading: one slow +consumer buffered the whole result in server memory, unbounded and +invisible to application code. The consumer's reads now drive `pull`, +which releases one source pull at a time — measured over 200 idle +event-loop turns, an async generator advanced by one step instead of +running away. Teardown releases a parked pull, so an aborted or cancelled +stream still ends. diff --git a/packages/web/server-functions/src/server.ts b/packages/web/server-functions/src/server.ts index c0c1bdb61..88bdb55d1 100644 --- a/packages/web/server-functions/src/server.ts +++ b/packages/web/server-functions/src/server.ts @@ -1676,6 +1676,20 @@ export function serializeResponseStream(value, codecOptions, signal) { value = guardFailures(value); let closeIterator = null; let closed = false; + // Demand gate. seroval's pump pulls the source as fast as it resolves and + // enqueues every node the moment it is parsed, so without this a slow + // consumer never slows the producer: the whole result accumulates in the + // stream's queue, in server memory, unbounded. The consumer's reads drive + // `pull`, which releases one source pull at a time. + let streamController = null; + let releaseDemand = null; + const wantsMore = () => !streamController || streamController.desiredSize > 0; + const awaitDemand = () => new Promise(resolve => (releaseDemand = resolve)); + const supplyDemand = () => { + const resolve = releaseDemand; + releaseDemand = null; + if (resolve) resolve(); + }; let cancelSerialize = null; let onAbort = null; const teardown = () => { @@ -1684,6 +1698,7 @@ export function serializeResponseStream(value, codecOptions, signal) { if (onAbort) signal.removeEventListener("abort", onAbort); if (cancelSerialize) cancelSerialize(); if (closeIterator) closeIterator(); + supplyDemand(); }; if ( value !== null && @@ -1711,8 +1726,12 @@ export function serializeResponseStream(value, codecOptions, signal) { // torn down before the codec opened the value (abort raced the // codec load): close the source immediately, never pull if (closed) closeIterator(); + const step = () => (finished ? { done: true, value: undefined } : it.next()); return { - next: () => (finished ? Promise.resolve({ done: true, value: undefined }) : it.next()) + // Waits for the consumer to want a chunk before pulling the next + // one; `finished` is re-read after the wait, since teardown can + // land while it is parked. + next: () => (wantsMore() ? Promise.resolve(step()) : awaitDemand().then(step)) }; } }; @@ -1722,6 +1741,7 @@ export function serializeResponseStream(value, codecOptions, signal) { // the top of shared.js), and a ReadableStream start may return a // promise — reads wait for it, so the stream's contract is unchanged async start(controller) { + streamController = controller; if (signal) { if (signal.aborted) { teardown(); @@ -1793,6 +1813,9 @@ export function serializeResponseStream(value, codecOptions, signal) { } }); }, + pull() { + supplyDemand(); + }, cancel() { teardown(); } diff --git a/packages/web/test/server/server-functions-open-gaps.spec.tsx b/packages/web/test/server/server-functions-open-gaps.spec.tsx index f3853e968..5219df90f 100644 --- a/packages/web/test/server/server-functions-open-gaps.spec.tsx +++ b/packages/web/test/server/server-functions-open-gaps.spec.tsx @@ -161,17 +161,18 @@ describe("a result the codec cannot encode", () => { }); describe("a streamed result nobody is reading", () => { - // GAP (#3118): the producer runs unboundedly ahead. The response stream is built - // with no `pull` and no queuing strategy, and every codec node is - // enqueued the moment it is parsed, so the producer runs as fast as it - // can resolve whether or not anyone reads. On a large or infinite stream - // one slow client buffers the whole result in server memory, invisibly - // to application code. + // Closed (#3118): the source is pulled behind a demand gate. The stream + // was built with no `pull` and no queuing strategy, and every codec node + // is enqueued the moment it is parsed, so the producer ran as fast as it + // could resolve whether or not anyone read — one slow client buffered + // the whole result in server memory, invisibly to application code. The + // consumer's reads now drive `pull`, which releases one source pull at a + // time. Ordinary guard now. // - // Counted in event-loop turns rather than wall-clock: a bounded producer - // stays near the queue size whatever the machine, an unbounded one - // tracks the turn count. - test.fails("does not let the producer run ahead of the consumer", async () => { + // Counted in event-loop turns rather than wall-clock, so the assertion + // means the same thing on any machine: a gated producer stays near the + // queue size, an ungated one tracks the turn count. + test("does not let the producer run ahead of the consumer", async () => { let produced = 0; registerServerFunction("gap-backpressure", async function* () { while (produced < 100_000) { @@ -191,6 +192,32 @@ describe("a streamed result nobody is reading", () => { expect(produced).toBeLessThan(50); }); + + test("resumes as the consumer reads, and stops when it leaves", async () => { + let produced = 0; + registerServerFunction("gap-backpressure-resume", async function* () { + while (produced < 100_000) { + produced++; + yield { n: produced }; + await new Promise(resolve => setImmediate(resolve)); + } + }); + + const response = await handleServerFunctionRequest(scriptedPost("gap-backpressure-resume")); + const reader = response.body!.getReader(); + for (let read = 0; read < 20; read++) await reader.read(); + const whileReading = produced; + await reader.cancel(); + const atCancel = produced; + for (let turn = 0; turn < 50; turn++) { + await new Promise(resolve => setImmediate(resolve)); + } + + // it kept up with the reads rather than stalling behind them... + expect(whileReading).toBeGreaterThan(1); + // ...and a departed consumer stops it, give or take the pull in flight + expect(produced).toBeLessThanOrEqual(atCancel + 1); + }); }); describe("the decode depth cap", () => { From c26415b8036c3365e52a1e7427b5bf1b03a1354d Mon Sep 17 00:00:00 2001 From: Vladimir Kutepov Date: Sun, 30 Aug 2026 19:51:32 +0700 Subject: [PATCH 2/5] fix(web): release a parked pull when the stream ends, and say what the gate covers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of the first shape found a regression I introduced. `onDone` and `onError` set `closed` directly rather than through `teardown()`, so a pull parked on the demand gate was stranded: `desiredSize` is 0 after close and null after error, both failing the `> 0` check, so the gate never reopened and the source's `finally` never ran — leaking generator, cursor and file-handle cleanup once per failed request. Every path that ends the stream now runs `finishSource()`, which closes the source and releases the parked pull. Also from review: - The gate rests on the default high-water mark of 1, and `releaseDemand` holds a single resolver, safe only because the pump is sequential. Both were load-bearing and unstated; both are now in the comment. - `!streamController` in `wantsMore` was unreachable — `start()` assigns the controller before the iterator is ever opened. - The spec header still called #3118 open, and the changeset claimed the producer "advanced by one step" where the test asserts it stays near the queue size. - The liveness assertion was `> 1` under a comment claiming it tracked the reads; a gate that resumed twice in twenty reads would have passed. Scope is now stated rather than implied: only the result itself is gated. A nested `{ items: rows() }` is pumped by the codec directly and still runs ahead — measured at 200 items over 200 idle turns against 1 for the top-level shape. --- .changeset/stream-demand-gate.md | 14 +++-- packages/web/server-functions/src/server.ts | 30 ++++++++-- .../server-functions-open-gaps.spec.tsx | 59 ++++++++++++++++++- 3 files changed, 90 insertions(+), 13 deletions(-) diff --git a/.changeset/stream-demand-gate.md b/.changeset/stream-demand-gate.md index 4c830879c..3a6bcdc1c 100644 --- a/.changeset/stream-demand-gate.md +++ b/.changeset/stream-demand-gate.md @@ -8,7 +8,13 @@ every codec node is enqueued the moment it is parsed, so the producer ran as fast as it could resolve whether or not anyone was reading: one slow consumer buffered the whole result in server memory, unbounded and invisible to application code. The consumer's reads now drive `pull`, -which releases one source pull at a time — measured over 200 idle -event-loop turns, an async generator advanced by one step instead of -running away. Teardown releases a parked pull, so an aborted or cancelled -stream still ends. +which releases one source pull at a time, so an unread stream stays near +the queue size instead of running away. + +Scope: the gate sits on the source the runtime wraps, which is the +result itself. An async iterable nested inside the result — `{ items: +rows() }` — is pumped by the codec directly and is not yet gated. Ending +the stream releases a parked pull, so an aborted, cancelled or failed +stream still closes its source; a consumer that abandons a stream without +cancelling it now leaves the producer parked rather than running it to +completion. diff --git a/packages/web/server-functions/src/server.ts b/packages/web/server-functions/src/server.ts index 88bdb55d1..78c51c688 100644 --- a/packages/web/server-functions/src/server.ts +++ b/packages/web/server-functions/src/server.ts @@ -1681,9 +1681,18 @@ export function serializeResponseStream(value, codecOptions, signal) { // consumer never slows the producer: the whole result accumulates in the // stream's queue, in server memory, unbounded. The consumer's reads drive // `pull`, which releases one source pull at a time. + // + // `desiredSize > 0` means "fewer than one chunk queued": the stream takes + // no queuing strategy, so it runs on the default high-water mark of 1. + // That default is what sets the depth, and raising it is how you would + // trade memory for fewer round trips. + // + // One resolver is enough because the pump is sequential — the same reason + // the wrapper below only exposes `next()`. Two concurrent pulls would + // overwrite it and strand the first. let streamController = null; let releaseDemand = null; - const wantsMore = () => !streamController || streamController.desiredSize > 0; + const wantsMore = () => streamController.desiredSize > 0; const awaitDemand = () => new Promise(resolve => (releaseDemand = resolve)); const supplyDemand = () => { const resolve = releaseDemand; @@ -1692,13 +1701,20 @@ export function serializeResponseStream(value, codecOptions, signal) { }; let cancelSerialize = null; let onAbort = null; + // Ends the source and releases a pull parked on the demand gate. Every + // path that stops the stream has to run this: a parked pull holds the + // source open and nothing else will resolve it — `desiredSize` is 0 after + // close and null after error, so the gate never reopens on its own. + const finishSource = () => { + if (closeIterator) closeIterator(); + supplyDemand(); + }; const teardown = () => { if (closed) return; closed = true; if (onAbort) signal.removeEventListener("abort", onAbort); if (cancelSerialize) cancelSerialize(); - if (closeIterator) closeIterator(); - supplyDemand(); + finishSource(); }; if ( value !== null && @@ -1728,9 +1744,9 @@ export function serializeResponseStream(value, codecOptions, signal) { if (closed) closeIterator(); const step = () => (finished ? { done: true, value: undefined } : it.next()); return { - // Waits for the consumer to want a chunk before pulling the next - // one; `finished` is re-read after the wait, since teardown can - // land while it is parked. + // Pulls straight through while the queue has room, and parks + // until a read makes room when it does not. `finished` is re-read + // after the wait, since teardown can land while it is parked. next: () => (wantsMore() ? Promise.resolve(step()) : awaitDemand().then(step)) }; } @@ -1781,12 +1797,14 @@ export function serializeResponseStream(value, codecOptions, signal) { if (closed) return; closed = true; if (onAbort) signal.removeEventListener("abort", onAbort); + finishSource(); controller.close(); }, onError(error) { if (closed) return; closed = true; if (onAbort) signal.removeEventListener("abort", onAbort); + finishSource(); // The head is committed by the time an encode failure arrives, so // the status is spent and no error tag can be added — and merely // erroring the stream truncates the body over a socket, which the diff --git a/packages/web/test/server/server-functions-open-gaps.spec.tsx b/packages/web/test/server/server-functions-open-gaps.spec.tsx index 5219df90f..4b2a73dc1 100644 --- a/packages/web/test/server/server-functions-open-gaps.spec.tsx +++ b/packages/web/test/server/server-functions-open-gaps.spec.tsx @@ -4,7 +4,7 @@ * `test/lifecycle-matrix/MATRIX.md`: the marker is the point — the suite * stays green while the gap is open and turns red the day it closes, at * which point the marker comes off and the test becomes an ordinary guard. - * Each carries the issue that tracks it: #3118 (open); #3117 (closed by + * Each carries the issue that tracks it: #3118 (closed); #3117 (closed by * the error trailer frame) and #3119 (closed by #3115's request bounds) * remain as ordinary guards, with the trailer's full matrix alongside. * @@ -193,6 +193,57 @@ describe("a streamed result nobody is reading", () => { expect(produced).toBeLessThan(50); }); + test("drains to completion, so the gate cannot swallow the tail", async () => { + registerServerFunction("gap-backpressure-drain", async function* () { + for (let n = 0; n < 40; n++) { + yield { n }; + await new Promise(resolve => setImmediate(resolve)); + } + }); + + const response = await handleServerFunctionRequest(scriptedPost("gap-backpressure-drain")); + // a gate that never reopened, or one that missed the last chunk, hangs + // here instead of failing an assertion — which is the point + expect((await response.text()).length).toBeGreaterThan(0); + }); + + // A pull parked on the gate holds the source open, and `desiredSize` is 0 + // after close and null after error — so the gate never reopens on its own + // and every path that ends the stream has to release it. Without that the + // source's cleanup silently never runs, once per failed request. + test("releases a parked pull when the codec ends the stream", async () => { + let cleanedUp = false; + registerServerFunction("gap-backpressure-cleanup", async () => ({ + rows: (async function* () { + try { + for (let n = 0; n < 20; n++) { + yield { n }; + await new Promise(resolve => setImmediate(resolve)); + } + } finally { + cleanedUp = true; + } + })(), + unencodable: { + get boom(): never { + throw new Error("a deferred encode failure, landing while a pull is parked"); + } + } + })); + + const response = await handleServerFunctionRequest(scriptedPost("gap-backpressure-cleanup")); + const reader = response.body!.getReader(); + // ONE read, then stop: the source parks on the gate, and the encode + // failure on the sibling branch ends the stream while it is parked. + // Draining instead would never park, and would never reach the defect. + await reader.read(); + for (let turn = 0; turn < 20; turn++) { + await new Promise(resolve => setTimeout(resolve, 0)); + } + + expect(cleanedUp).toBe(true); + }); + test("resumes as the consumer reads, and stops when it leaves", async () => { let produced = 0; registerServerFunction("gap-backpressure-resume", async function* () { @@ -213,8 +264,10 @@ describe("a streamed result nobody is reading", () => { await new Promise(resolve => setImmediate(resolve)); } - // it kept up with the reads rather than stalling behind them... - expect(whileReading).toBeGreaterThan(1); + // liveness: test 1 passes just as well if the gate never reopens, so + // this is the case that would catch a deadlock. A healthy run tracks + // the reads almost exactly; the bound is loose enough for a slow CI. + expect(whileReading).toBeGreaterThanOrEqual(10); // ...and a departed consumer stops it, give or take the pull in flight expect(produced).toBeLessThanOrEqual(atCancel + 1); }); From 6e3f01058125f2336059e08b49ac5b0ec5067fa0 Mon Sep 17 00:00:00 2001 From: Vladimir Kutepov Date: Sun, 30 Aug 2026 19:56:08 +0700 Subject: [PATCH 3/5] test(web): make the teardown guard actually discriminate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first version of this test passed with finishSource() removed from both codec-end paths — it never reached the defect. Two reasons, and the second is the one worth writing down: - the source was NESTED (`{ rows: gen(), … }`), and a nested iterable never enters the wrapper, so nothing parked; - a top-level source needs no sibling branch to carry the failure. The deferred failure rides INSIDE a yielded chunk: a pending promise in the first value resolving to an object whose getter throws. The pump enqueues the chunk, asks for the second, parks because the consumer stopped reading, and only then does the promise settle and onError fire against a parked pull. `produced` is asserted alongside the cleanup so the nested-shape mistake cannot quietly recur: if the source never parks, the test says so instead of passing for the wrong reason. Verified both ways — with finishSource() in the codec-end paths the cleanup runs, without it the assertion fails. --- .../server-functions-open-gaps.spec.tsx | 45 ++++++++++--------- 1 file changed, 25 insertions(+), 20 deletions(-) diff --git a/packages/web/test/server/server-functions-open-gaps.spec.tsx b/packages/web/test/server/server-functions-open-gaps.spec.tsx index 4b2a73dc1..1c48c0883 100644 --- a/packages/web/test/server/server-functions-open-gaps.spec.tsx +++ b/packages/web/test/server/server-functions-open-gaps.spec.tsx @@ -211,36 +211,41 @@ describe("a streamed result nobody is reading", () => { // after close and null after error — so the gate never reopens on its own // and every path that ends the stream has to release it. Without that the // source's cleanup silently never runs, once per failed request. - test("releases a parked pull when the codec ends the stream", async () => { + test("a codec failure landing while a pull is parked still closes the source", async () => { let cleanedUp = false; - registerServerFunction("gap-backpressure-cleanup", async () => ({ - rows: (async function* () { - try { - for (let n = 0; n < 20; n++) { - yield { n }; - await new Promise(resolve => setImmediate(resolve)); - } - } finally { - cleanedUp = true; - } - })(), - unencodable: { - get boom(): never { - throw new Error("a deferred encode failure, landing while a pull is parked"); + let produced = 0; + // The failure has to ride INSIDE a yielded chunk of the top-level + // source. Putting it on a sibling branch makes the generator nested, + // and a nested iterable never reaches the gate at all — which is why + // `produced` is asserted too: it proves the source really parked, so + // this cannot quietly decay into testing nothing. + registerServerFunction("gap-backpressure-teardown", async function* () { + try { + produced++; + yield { + late: Promise.resolve().then(() => ({ + get boom(): never { + throw new Error("unencodable, discovered after the gate parked"); + } + })) + }; + for (let n = 0; n < 20; n++) { + produced++; + yield { n }; } + } finally { + cleanedUp = true; } - })); + }); - const response = await handleServerFunctionRequest(scriptedPost("gap-backpressure-cleanup")); + const response = await handleServerFunctionRequest(scriptedPost("gap-backpressure-teardown")); const reader = response.body!.getReader(); - // ONE read, then stop: the source parks on the gate, and the encode - // failure on the sibling branch ends the stream while it is parked. - // Draining instead would never park, and would never reach the defect. await reader.read(); for (let turn = 0; turn < 20; turn++) { await new Promise(resolve => setTimeout(resolve, 0)); } + expect(produced).toBe(1); expect(cleanedUp).toBe(true); }); From f2824f6306906ad6e617d40493260a9745c2d7d4 Mon Sep 17 00:00:00 2001 From: Vladimir Kutepov Date: Sun, 30 Aug 2026 21:51:58 +0700 Subject: [PATCH 4/5] fix(web): check finished before the gate, and make a test that parks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Third review round, both findings real. The gate was checked before `finished`, so teardown landing while a pull was in flight stranded the codec's pump: the release it fires finds nothing parked, the in-flight pull then resolves, the pump asks for the next item, `wantsMore()` is false — 0 after close, null after error — and it parks on a resolver nobody will ever call. `push()` never returns. Impossible before this PR, where `finished` was checked first. One token, and it mirrors the defect the previous round fixed: that one stranded the source, this one stranded the pump. Worth stating plainly: this half is not guarded by a test. The failure is a leaked pending promise inside seroval's pump with no outward symptom — the source still runs its `finally`, the response still ends, nothing observable differs. It was found with instrumented park/release counters, and that is the evidence it rests on. The tests also did not park at all. Both read in a tight loop, so a read request is always pending, `desiredSize` never drops and the producer never reaches the gate — deleting `pull()` outright left all eight green, including the one whose comment claimed it would catch a deadlock. The new test pauses between reads, which is what puts the producer on the gate, and reads to completion with a deadline so a gate that never reopens fails fast instead of hanging. It counts arrivals by its own key rather than by the codec's node shapes. --- packages/web/server-functions/src/server.ts | 11 +++-- .../server-functions-open-gaps.spec.tsx | 43 ++++++++++++++----- 2 files changed, 41 insertions(+), 13 deletions(-) diff --git a/packages/web/server-functions/src/server.ts b/packages/web/server-functions/src/server.ts index 78c51c688..ff08922d9 100644 --- a/packages/web/server-functions/src/server.ts +++ b/packages/web/server-functions/src/server.ts @@ -1745,9 +1745,14 @@ export function serializeResponseStream(value, codecOptions, signal) { const step = () => (finished ? { done: true, value: undefined } : it.next()); return { // Pulls straight through while the queue has room, and parks - // until a read makes room when it does not. `finished` is re-read - // after the wait, since teardown can land while it is parked. - next: () => (wantsMore() ? Promise.resolve(step()) : awaitDemand().then(step)) + // until a read makes room when it does not. `finished` is checked + // FIRST: teardown can land while a pull is in flight, and the + // release it fires then finds nothing parked — so a gate checked + // first would park the next pull on a resolver nobody will ever + // call, stranding the codec's pump. `finished` is re-read after + // the wait for the same reason from the other direction. + next: () => + finished || wantsMore() ? Promise.resolve(step()) : awaitDemand().then(step) }; } }; diff --git a/packages/web/test/server/server-functions-open-gaps.spec.tsx b/packages/web/test/server/server-functions-open-gaps.spec.tsx index 1c48c0883..b5a5afa2f 100644 --- a/packages/web/test/server/server-functions-open-gaps.spec.tsx +++ b/packages/web/test/server/server-functions-open-gaps.spec.tsx @@ -193,18 +193,41 @@ describe("a streamed result nobody is reading", () => { expect(produced).toBeLessThan(50); }); - test("drains to completion, so the gate cannot swallow the tail", async () => { - registerServerFunction("gap-backpressure-drain", async function* () { - for (let n = 0; n < 40; n++) { - yield { n }; - await new Promise(resolve => setImmediate(resolve)); - } + // The gate has to REOPEN, not merely close, and nothing above proves it: + // a consumer that reads in a tight loop always has a read request + // pending, so `desiredSize` never drops and the producer never parks. + // Pausing between reads is what puts it on the gate. Deleting `pull()` + // outright leaves every other test in this file green and deadlocks this + // one, which is the whole point of it. + test("keeps delivering after the consumer pauses long enough to park it", async () => { + registerServerFunction("gap-backpressure-park", async function* () { + for (let n = 0; n < 12; n++) yield { n }; }); - const response = await handleServerFunctionRequest(scriptedPost("gap-backpressure-drain")); - // a gate that never reopened, or one that missed the last chunk, hangs - // here instead of failing an assertion — which is the point - expect((await response.text()).length).toBeGreaterThan(0); + const response = await handleServerFunctionRequest(scriptedPost("gap-backpressure-park")); + const reader = response.body!.getReader(); + const decoder = new TextDecoder(); + let body = ""; + + for (;;) { + // long enough for the queue to drain and the next pull to park + for (let turn = 0; turn < 5; turn++) { + await new Promise(resolve => setTimeout(resolve, 0)); + } + const next = await Promise.race([ + reader.read(), + new Promise((_, reject) => + setTimeout(() => reject(new Error("the gate parked and never reopened")), 2000) + ) + ]); + if (next.done) break; + body += decoder.decode(next.value as Uint8Array); + } + + // every item arrived, so the gate released each park in turn. Counted + // by this test's own key rather than by the codec's node shapes, which + // are not what it is about. + expect(body.split('["n"]').length - 1).toBe(12); }); // A pull parked on the gate holds the source open, and `desiredSize` is 0 From 5d292778278dee60e80307110ee1e51bc6633d2e Mon Sep 17 00:00:00 2001 From: Vladimir Kutepov Date: Sun, 30 Aug 2026 23:45:49 +0700 Subject: [PATCH 5/5] test(web): say what the resume test actually pins MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Its comment claimed it was the case that would catch a deadlock. It is not: it reads in a tight loop, so a read request is always pending, `desiredSize` never drops and nothing parks — the same blindness review found in the original pair. What it pins is the cancel half. The pausing test is the liveness one. --- .../web/test/server/server-functions-open-gaps.spec.tsx | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/packages/web/test/server/server-functions-open-gaps.spec.tsx b/packages/web/test/server/server-functions-open-gaps.spec.tsx index b5a5afa2f..0595d4e0b 100644 --- a/packages/web/test/server/server-functions-open-gaps.spec.tsx +++ b/packages/web/test/server/server-functions-open-gaps.spec.tsx @@ -292,9 +292,11 @@ describe("a streamed result nobody is reading", () => { await new Promise(resolve => setImmediate(resolve)); } - // liveness: test 1 passes just as well if the gate never reopens, so - // this is the case that would catch a deadlock. A healthy run tracks - // the reads almost exactly; the bound is loose enough for a slow CI. + // This pins the CANCEL half — that a departed consumer stops the + // producer. It does not catch a gate that never reopens: reading in a + // tight loop keeps a read request pending, so `desiredSize` never + // drops and nothing ever parks. The pausing test above is the one that + // catches that, and it took two attempts to learn the difference. expect(whileReading).toBeGreaterThanOrEqual(10); // ...and a departed consumer stops it, give or take the pull in flight expect(produced).toBeLessThanOrEqual(atCancel + 1);