Duplex stream pipeline regression in 13

[mafintosh]

• Version: 13.13.0
• Platform: Mac
• Subsystem: stream

What steps will reproduce the bug?

pipeline in 13 seems to destroy duplex streams before the get a change to finish their writable flus h (ws.end() -> ws._final) lifecycle.

I managed to boil it down to a pretty straightforward test case below

const stream = require('stream')

// duplex stream similar to a tcp stream etc
const dup = new stream.Duplex({
  write (data, enc, cb) {
    cb()
  },
  destroy () {
    console.log('ws: am getting destroyed')
  },
  final (cb) {
    console.log('ws: flushing writable...')
    setTimeout(function () {
      console.log('ws: done flushing writable...')
      cb()
    }, 1000)
  }
})

// just some sink
const sink = new stream.Writable({
  write (data, enc, cb) {
    cb()
  }
})

// pipe readable side
stream.pipeline(dup, sink, function () { })

dup.write('test')
dup.end()

Running this produces:

ws: am getting destroyed
ws: flushing writable...
ws: done flushing writable...

Notice that the dup stream gets destroyed before it has a chance to finish it's writable flush in the _final lifecycle, due to the pipeline auto destroying it in 13.

[mafintosh]

I think this is related to #31940 as well

[mafintosh]

Equivalent test case using TCP

const net = require('net')
const pipeline = require('stream').pipeline

net.createServer(function (socket) {
  // echo server
  pipeline(socket, socket, () => {})
  // 13 force destroys the socket before it has a chance to emit finish
  socket.on('finish', function () {
    console.log('finished only printed in node 12')
  })
}).listen(10000, function () {
  const socket = net.connect(10000)
  socket.end()
})

[mcollina]

cc @ronag

[ronag]

Seems to work on master. Checking v13-staging.

[ronag]

This is a problem on v13 only as far as I can tell. Basically pre v14 net.Socket had a bit strange semantics regarding writable/readable which kind of short-circuits the writable/readable check which pipeline depends on.

[ronag]

Looking into ways to get around this.

[mafintosh]

This happens in master on all streams that don't have autoDestroy enabled as well (ie any readable-stream stream or any stream that opt out)

const stream = require('stream')

// duplex stream similar to a tcp stream etc
const dup = new stream.Duplex({
  autoDestroy: false,
  write (data, enc, cb) {
    cb()
  },
  read () {
    this.push(null)
  },
  destroy () {
    console.log('ws: am getting destroyed')
  },
  final (cb) {
    console.log('ws: flushing writable...')
    setTimeout(function () {
      console.log('ws: done flushing writable...')
      cb()
    }, 1000)
  }
})

// just some sink
const sink = new stream.Writable({
  write (data, enc, cb) {
    cb()
  }
})

// pipe readable side
stream.pipeline(dup, sink, function () { })

dup.write('test')
dup.end()

Above prints the following

ws: flushing writable...
ws: am getting destroyed
ws: done flushing writable...

cc @ronag @mcollina

[ronag]

Yes, I see the problem. Will try to prepare a PR by the end of the day. Thanks a lot @mafintosh!

[mafintosh]

@ronag i think this is related to #32965

[ronag]

@ronag i think this is related to #32965

I think it's unrelated. See comment in #32965.