When users clone(), automatically create streams with custom highWaterMark

[bitinn]

Instead of changing default highWaterMark of res.body (see #151), perhaps setting them during clone() would be a better idea. AFAIK, users mostly run into backpressure when they clone() and consume a large response while leaving the other stream idle.

This will likely solve many trivial issues.

[DevBrent]

The https://github.com/mikeal/r2 library is particularly affected. All of their short-hand single-promise resolvers e.g. await r2(url).json, await r2(url).text are affected because they clone() every response before trying to use the node-fetch json/text resolvers.

[bitinn]

@DevBrent PR is welcomed, should be a simple change, but we do want to see a good test to verify it's working as intended.

To truly solve the highWaterMark limit you need a dynamically sizing stream, I am not sure it's possible, but I haven't looked at latest nodejs development.

[DevBrent]

@bitinn I'm not even sure I like what seems is the only solution. On one hand, no reader should be prevented from reading the entire stream if they read faster than another reader. On the other, a sufficiently fast reader or in situations like with r2, dynamically sizing the stream will cause the entire stream to be stored in-memory which is not scalable.

Edit: r2 does intend for this in a way, because in order to afford reading the entire response once while still leaving open the ability to later read it as JSON or an arrayBuffer, you have to have the entire stream in memory.

[nikdo]

I've done some initial research about highWaterMark behavior when one of the streams does not have a consumer:

const highWaterMark = 5

// push byte-sized chunks to reach +1 over buffer limit to trigger backpressure
const input = new RandomBytes({
  // Transform stream has 2 buffers it absorbes twice the buffer before backpressuring
  // https://nodejs.org/api/stream.html#stream_buffering
  pushes: highWaterMark * 2 + 1,
  verbose: false
})

// highWaterMark limit is problem only for the stream without consumer
let p1 = new PassThrough({ highWaterMark })
let p2 = new PassThrough()

// cannot hoook onto p1 data because that would add consumer
p2.on('data', data => {
  logBuffer(p1, 'p1')
})

input.pipe(p1) // no pipe for you
input.pipe(p2).pipe(fs.createWriteStream('/dev/null'))

Full code

Because PassThrough is a Transform stream, it has two buffers. That means the backpressure occurs when there is 16 * 1024 * 2 + 1 on the input.

Now I'll investigate how to properly test stream that does not emit end event leading to unresolved promise. Not really good first issue when you want to see a good test to verify it's working as intended. :) Any help appreciated.