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

.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

@@ -1676,14 +1676,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 &&
@@ -1711,8 +1742,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)
         };
       }
     };
@@ -1722,6 +1762,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();
@@ -1761,12 +1802,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
@@ -1793,6 +1836,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", () => {