Merge pull request from GHSA-9f24-jqhm-jfcw

* fetch: pull don't push

Signed-off-by: Matteo Collina <hello@matteocollina.com>

* added tests

Signed-off-by: Matteo Collina <hello@matteocollina.com>

---------

Signed-off-by: Matteo Collina <hello@matteocollina.com>

lib/fetch/index.js

@@ -1099,10 +1099,10 @@ function fetchFinale (fetchParams, response) {
 
     const byteStream = new ReadableStream({
       readableStream: transformStream.readable,
-      async start (controller) {
+      async pull (controller) {
         const reader = this.readableStream.getReader()
 
-        while (true) {
+        while (controller.desiredSize >= 0) {
           const { done, value } = await reader.read()
 
           if (done) {
@@ -1113,6 +1113,7 @@ function fetchFinale (fetchParams, response) {
           controller.enqueue(value)
         }
       },
+      queuingStrategy: new ByteLengthQueuingStrategy({ highWaterMark: 16384 }),
       type: 'bytes'
     })
 
@@ -1927,6 +1928,7 @@ async function httpNetworkFetch (
   //     cancelAlgorithm set to cancelAlgorithm.
   const stream = new ReadableStream(
     {
+      highWaterMark: 16384,
       async start (controller) {
         fetchParams.controller.controller = controller
       },
@@ -1936,7 +1938,8 @@ async function httpNetworkFetch (
       async cancel (reason) {
         await cancelAlgorithm(reason)
       },
-      type: 'bytes'
+      type: 'bytes',
+      queuingStrategy: new ByteLengthQueuingStrategy({ highWaterMark: 16384 })
     }
   )
 

test/fetch/pull-dont-push.js

@@ -0,0 +1,53 @@
+'use strict'
+
+const { test } = require('node:test')
+const assert = require('node:assert')
+const { fetch } = require('../..')
+const { createServer } = require('http')
+const { once } = require('events')
+const { Readable, pipeline } = require('stream')
+const { setTimeout: sleep } = require('timers/promises')
+
+const { closeServerAsPromise } = require('../utils/node-http')
+
+test('Allow the usage of custom implementation of AbortController', async (t) => {
+  let count = 0
+  let socket
+  const server = createServer((req, res) => {
+    res.statusCode = 200
+    socket = res.socket
+
+    // infinite stream
+    const stream = new Readable({
+      read () {
+        this.push('a')
+        if (count++ > 1000000) {
+          this.push(null)
+        }
+      }
+    })
+
+    pipeline(stream, res, () => {})
+  })
+
+  t.after(closeServerAsPromise(server))
+
+  server.listen(0)
+  await once(server, 'listening')
+
+  t.diagnostic('server listening on port %d', server.address().port)
+  const res = await fetch(`http://localhost:${server.address().port}`)
+  t.diagnostic('fetched')
+
+  // Some time is needed to fill the buffer
+  await sleep(1000)
+
+  assert.strictEqual(socket.bytesWritten < 1024 * 1024, true) // 1 MB
+  socket.destroy()
+
+  // consume the  stream
+  try {
+    /* eslint-disable-next-line no-empty, no-unused-vars */
+    for await (const chunk of res.body) {}
+  } catch {}
+})