fix(web): pull a streamed result behind a demand gate (#3124)

* fix(web): pull a streamed result behind a demand gate

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.

* fix(web): release a parked pull when the stream ends, and say what the gate covers

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.

* test(web): make the teardown guard actually discriminate

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.

* fix(web): check finished before the gate, and make a test that parks

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.

* test(web): say what the resume test actually pins

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.

Author: frenzzy

.changeset/stream-demand-gate.md

@@ -0,0 +1,20 @@
+---
+"@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, 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.

packages/web/server-functions/src/server.ts

@@ -1681,14 +1681,45 @@ 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.
+  //
+  // `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.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;
+  // 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();
+    finishSource();
   };
   if (
     value !== null &&
@@ -1716,8 +1747,17 @@ 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())
+          // Pulls straight through while the queue has room, and parks
+          // 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)
         };
       }
     };
@@ -1727,6 +1767,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();
@@ -1766,12 +1807,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
@@ -1798,6 +1841,9 @@ export function serializeResponseStream(value, codecOptions, signal) {
         }
       });
     },
+    pull() {
+      supplyDemand();
+    },
     cancel() {
       teardown();
     }

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.
  *
@@ -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,115 @@ describe("a streamed result nobody is reading", () => {
 
     expect(produced).toBeLessThan(50);
   });
+
+  // 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-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<never>((_, 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
+  // 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("a codec failure landing while a pull is parked still closes the source", async () => {
+    let cleanedUp = false;
+    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-teardown"));
+    const reader = response.body!.getReader();
+    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);
+  });
+
+  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));
+    }
+
+    // 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);
+  });
 });
 
 describe("the decode depth cap", () => {