Not sure if this should be a semver-major.

CI: https://ci.nodejs.org/job/node-test-pull-request/30805/

This looks great if it does not cause issues in user-land (it will almost certainly break ws tests).

This looks great if it does not cause issues in user-land (it will almost certainly break ws tests).

I'll run a CITGM.

CI: https://ci.nodejs.org/job/node-test-pull-request/30810/
CITGM: https://ci.nodejs.org/view/Node.js-citgm/job/citgm-smoker/2351/

CITGM looks good apart from ws. Sorry @lpinca. I'm fine with skipping this PR if it's a big effort from you part.

It should be just a matter of fixing those two failing tests (not sure how), core functionality should not be broken.

Perhaps wait until #32780 lands so I don't have to fix them twice.

@nodejs/streams

I tagged this as "dont-land" on everything but v14.x.

I've fixed the failing tests: websockets/ws@18d773d.

CI: https://ci.nodejs.org/job/node-test-pull-request/30972/

Landed in 003fb53

@ronag This PR (found after bisecting on v14.x) seems to be causing issues for ssh2/ssh2-streams when large amounts of data are being passed through a stream. The same regression still exists on master as of this writing. Can we at least revert this on v14.x, especially if there is no guidance on how to deal with the breakage in a backwards-compatible manner?

@mscdex +1 in reverting this. I would love to have a regression test added (even in ssh2 and we can add it to citgm) so that we do not regress again.

I'm ok with revert + a comment in the code. Just a note here, reverting this might have performance impact e.g. when piping from a file stream (which has a larger highwater mark) to a sync transform stream (which has a smaller highwatermark) e.g. for hashing.

It's a bit strange to me that this would cause breakage. Is ssh2 doing something funky with streams?

Hi,
If it can help, as part of troubleshooting a related issue @theophilusx made this script that will hang forever if the passed data is large enough as discussed here.

None of those examples are self-contained :/, they need an active ssh server to run.

I've stripped down the relevant ssh2 internals so that it is a completely standalone test that exhibits the problem. The code here doesn't exactly represent what is happening internally in most cases (mostly referring to the write happening before piping, that's done here to simulate backpressure and buffered data), but it should sufficiently demonstrate the same issue. Also it's possible the test code could be simplified further if someone wanted to have a go at it.

'use strict';

const { Duplex, Transform } = require('stream');
const { connect, createServer } = require('net');
const { inherits } = require('util');

function Channel(protoStream, socket) {
  const streamOpts = {
    highWaterMark: 2 * 1024 * 1024,
    allowHalfOpen: false
  };

  Duplex.call(this, streamOpts);

  socket.on('drain', () => {
    console.log('ondrain()', 'waitClientDrain', this._waitSocketDrain);
    if (this._waitSocketDrain) {
      this._waitSocketDrain = false;
      if (this._chunk)
        this._write(this._chunk, null, this._chunkcb);
      else if (this._chunkcb)
        this._chunkcb();
    }
  });

  this._protoStream = protoStream;

  // outgoing data
  this._waitSocketDrain = false;

  this._chunk = undefined;
  this._chunkcb = undefined;
}
inherits(Channel, Duplex);

Channel.prototype._read = function(n) {};

Channel.prototype._write = function(data, encoding, cb) {
  const protoStream = this._protoStream;
  const packetSize = 64 * 1024;
  const len = data.length;
  let p = 0;

  while (len - p > 0) {
    let sliceLen = len - p;
    if (sliceLen > packetSize)
      sliceLen = packetSize;

    const slice = data.slice(p, p + sliceLen);
    const ret = protoStream.sendData(slice);
    console.log(`Channel._write() ret = ${ret} after writing ${slice.length} byte(s)`);

    p += sliceLen;

    if (!ret) {
      this._waitSocketDrain = true;
      this._chunk = undefined;
      this._chunkcb = cb;
      break;
    }
  }

  console.log(`Channel._write() outside of loop; p=${p} len=${len} waitSocketDrain=${this._waitSocketDrain}`);

  if (len - p > 0) {
    if (p > 0) {
      // partial
      let buf = Buffer.allocUnsafe(len - p);
      data.copy(buf, 0, p);
      this._chunk = buf;
    } else {
      this._chunk = data;
    }
    this._chunkcb = cb;
    return;
  }

  if (!this._waitSocketDrain)
    cb();
};

function ProtocolStream() {
  Transform.call(this, { highWaterMark: 32 * 1024 });
}
inherits(ProtocolStream, Transform);

ProtocolStream.prototype.sendData = function(data) {
  return this.push(data);
};

createServer(function(socket) {
  this.close();
}).listen(0, '127.0.0.1', function() {
  const { address, port } = this.address();
  connect(port, address, function() {
    console.log('Client connected');
    const protoStream = new ProtocolStream();
    const channel = new Channel(protoStream, this);
    channel.on('finish', () => {
      console.log('============ Channel finish');
      this.destroy();
    });
    channel.write(Buffer.alloc(128 * 1024));
    protoStream.pipe(this).pipe(protoStream);
    channel.end(Buffer.alloc(128 * 1024));
  });
});

Executing the test on node v10.19.0 results in the following output:

$ node-v10.19.0 stream-issue.js
Client connected
Channel._write() ret = false after writing 65536 byte(s)
Channel._write() outside of loop; p=65536 len=131072 waitSocketDrain=true
ondrain() waitClientDrain true
Channel._write() ret = false after writing 65536 byte(s)
Channel._write() outside of loop; p=65536 len=65536 waitSocketDrain=true
ondrain() waitClientDrain true
Channel._write() ret = false after writing 65536 byte(s)
Channel._write() outside of loop; p=65536 len=131072 waitSocketDrain=true
ondrain() waitClientDrain true
Channel._write() ret = false after writing 65536 byte(s)
Channel._write() outside of loop; p=65536 len=65536 waitSocketDrain=true
ondrain() waitClientDrain true
============ Channel finish
$

Executing the test on node v14.1.0 or later results in the following output:

$ node-v14.7.0 stream-issue.js
Client connected
Channel._write() ret = false after writing 65536 byte(s)
Channel._write() outside of loop; p=65536 len=131072 waitSocketDrain=true

(... and the process never exits)

Looks like it ends up making an incorrect assumption regarding whether or not 'drain' will be emitted. Maybe something with _write being called in the drain handler which can set _waitSocketDrain to true without the stream actually needing a drain.

I think this is weird/incorrect/bug usage and node streams are behaving correctly. However, I guess it is is breaking change so if no one feels strongly about the performance loss I think we just revert it.

@mscdex @ronag how hard would it be to fix on the ssh2 side? What puzzles me is that all of ssh2 tests were passing, so it would have been extremely hard to spot this.

Given that we were on the fence on this being semver-major we can revert on v14 and keep it on master/v15.

Not to do the "+1" thing here, but we've seen this regression in the wild with the (quite popular) ssh2 library.

Is there any chance of getting this reverted?

@schmod, the changes reverted in #35941. @ronag, maybe this PR should be reopened?