Repository: denoland/deno
Pull request: fix(ext/web): fix promise chain leak of ReadableStream

Closes #30298

This problem occurs because promise chain keep growing(like a stack overflow caused by a recursive function) while the stream is running. To solve this, used a loop.

Changed files:
filename: ext/web/06_streams.js
patch: @@ -787,18 +787,18 @@ class ResourceStreamResourceSink {
 async function readableStreamWriteChunkFn(reader, sink, chunk) {
   // Empty chunk. Re-read.
   if (chunk.length == 0) {
-    await readableStreamReadFn(reader, sink);
-    return;
+    return true;
   }
 
   const res = op_readable_stream_resource_write_sync(sink.external, chunk);
   if (res == 0) {
     // Closed
     await reader.cancel("resource closed");
     sink.close();
+
+    return false;
   } else if (res == 1) {
-    // Successfully written (synchronous). Re-read.
-    await readableStreamReadFn(reader, sink);
+    return true;
   } else if (res == 2) {
     // Full. If the channel is full, we perform an async await until we can write, and then return
     // to a synchronous loop.
@@ -808,10 +808,12 @@ async function readableStreamWriteChunkFn(reader, sink, chunk) {
         chunk,
       )
     ) {
-      await readableStreamReadFn(reader, sink);
+      return true;
     } else {
       await reader.cancel("resource closed");
       sink.close();
+
+      return false;
     }
   }
 }
@@ -820,60 +822,66 @@ async function readableStreamWriteChunkFn(reader, sink, chunk) {
  * @param {ReadableStreamDefaultReader<Uint8Array>} reader
  * @param {any} sink
  */
-function readableStreamReadFn(reader, sink) {
-  // The ops here look like op_write_all/op_close, but we're not actually writing to a
-  // real resource.
-  let reentrant = true;
-  let gotChunk = undefined;
-  const promise = new Deferred();
-  readableStreamDefaultReaderRead(reader, {
-    chunkSteps(chunk) {
-      // If the chunk has non-zero length, write it
-      if (reentrant) {
-        gotChunk = chunk;
-      } else {
-        PromisePrototypeThen(
-          readableStreamWriteChunkFn(reader, sink, chunk),
-          () => promise.resolve(),
-          (e) => promise.reject(e),
+async function readableStreamReadFn(reader, sink) {
+  let loop = true;
+
+  while (loop) {
+    // The ops here look like op_write_all/op_close, but we're not actually writing to a
+    // real resource.
+    let reentrant = true;
+    let gotChunk = undefined;
+    const promise = new Deferred();
+
+    readableStreamDefaultReaderRead(reader, {
+      chunkSteps(chunk) {
+        // If the chunk has non-zero length, write it
+        if (reentrant) {
+          gotChunk = chunk;
+        } else {
+          PromisePrototypeThen(
+            readableStreamWriteChunkFn(reader, sink, chunk),
+            (loop) => promise.resolve(loop),
+            (e) => promise.reject(e),
+          );
+        }
+      },
+      closeSteps() {
+        sink.close();
+        promise.resolve(false);
+      },
+      errorSteps(error) {
+        const success = op_readable_stream_resource_write_error(
+          sink.external,
+          extractStringErrorFromError(error),
         );
-      }
-    },
-    closeSteps() {
-      sink.close();
-      promise.resolve();
-    },
-    errorSteps(error) {
-      const success = op_readable_stream_resource_write_error(
-        sink.external,
-        extractStringErrorFromError(error),
+        // We don't cancel the reader if there was an error reading. We'll let the downstream
+        // consumer close the resource after it receives the error.
+        if (!success) {
+          PromisePrototypeThen(
+            reader.cancel("resource closed"),
+            () => {
+              sink.close();
+              promise.resolve(false);
+            },
+            (e) => promise.reject(e),
+          );
+        } else {
+          sink.close();
+          promise.resolve(false);
+        }
+      },
+    });
+    reentrant = false;
+    if (gotChunk) {
+      PromisePrototypeThen(
+        readableStreamWriteChunkFn(reader, sink, gotChunk),
+        (loop) => promise.resolve(loop),
+        (e) => promise.reject(e),
       );
-      // We don't cancel the reader if there was an error reading. We'll let the downstream
-      // consumer close the resource after it receives the error.
-      if (!success) {
-        PromisePrototypeThen(
-          reader.cancel("resource closed"),
-          () => {
-            sink.close();
-            promise.resolve();
-          },
-          (e) => promise.reject(e),
-        );
-      } else {
-        sink.close();
-        promise.resolve();
-      }
-    },
-  });
-  reentrant = false;
-  if (gotChunk) {
-    PromisePrototypeThen(
-      readableStreamWriteChunkFn(reader, sink, gotChunk),
-      () => promise.resolve(),
-      (e) => promise.reject(e),
-    );
+    }
+
+    loop = await promise.promise;
   }
-  return promise.promise;
 }
 
 /**

filename: tests/unit/streams_test.ts
patch: @@ -651,3 +651,72 @@ Deno.test(async function readableStreamFromWithStringThrows() {
   stopSignal.abort();
   await p;
 });
+
+Deno.test(async function readableStreamEmittingManyChunks() {
+  const code = `
+    const serverPort = 4594;
+    const stopSignal = new AbortController();
+    let count = 0;
+    let before = 0;
+    let after = 0;
+
+    function startServer() {
+      Deno.serve({ port: serverPort, signal: stopSignal.signal }, (_) => {
+        return new Response(
+          new ReadableStream({
+            start() {
+              before = Deno.memoryUsage().heapUsed;
+            },
+            pull(controller) {
+              const used = Deno.memoryUsage().heapUsed;
+
+              if (used > after) {
+                after = used;
+              }
+
+              if (count < 30_000) {
+                controller.enqueue(new Uint8Array([0]));
+              } else {
+                controller.close();
+              }
+
+              count += 1;
+            },
+          }),
+        );
+      });
+    }
+
+    async function startClient() {
+      const response = await fetch(\`http://localhost:\${serverPort}\`);
+      const reader = response.body?.getReader();
+      if (!reader) {
+        throw new Error("client: failed to get reader from response");
+      }
+
+      while (true) {
+        const { done } = await reader.read();
+        if (done) break;
+      }
+    }
+
+    startServer();
+    await startClient();
+    stopSignal.abort();
+    console.log(\`\${after} / \${before} = \${after / before}\`);
+    if (after / before > 1.5) {
+      Deno.exit(1);
+    }
+  `;
+  const command = new Deno.Command(Deno.execPath(), {
+    args: ["run", "-N", "-"],
+    stdin: "piped",
+  });
+
+  await using child = command.spawn();
+  await ReadableStream.from([code])
+    .pipeThrough(new TextEncoderStream())
+    .pipeTo(child.stdin);
+
+  assertEquals((await child.status).code, 0, "memory leak");
+});