stream: pipeline don't destroy Duplex src before 'finish'

[ronag]

pipeline was too agressive with destroying Duplex
streams which were the first argument into pipeline.

Just because it's !writable does not mean that it
is safe to be destroyed, unless it has also emitted
'finish'.

Fixes: #32955

Checklist

• make -j4 test (UNIX), or vcbuild test (Windows) passes
• tests and/or benchmarks are included
• documentation is changed or added
• commit message follows commit guidelines

[BridgeAR]

Nit: it is possible to reduce the code branches a bit:

Suggested change

	const wState = stream._writableState;
	const writableEnded = stream.writableEnded ||
	(wState && wState.ended);
	const writableFinished = stream.writableFinished ||
	(wState && wState.finished);
	const willFinish = stream.writable ||
	(writableEnded && !writableFinished);
	const willEnd = stream.readable;
	if (err || !final || !stream.readable) {
	destroyImpl.destroyer(stream, err);
	if (!err) {
	// First
	if (reading && !writing && willFinish) {
	return callback();
	}
	// Last
	if (!reading && writing && willEnd) {
	return callback();
	}
	if (!err) {
	// Check if the last stream will end:
	if (!reading) {
	if (writing && stream.readable) {
	return callback();
	}
	// First stream
	} else if (!writing) {
	if (stream.writable) {
	return callback();
	}
	const wState = stream._writableState;
	const writableEnded = stream.writableEnded ||
	(wState && wState.ended);
	const writableFinished = stream.writableFinished ||
	(wState && wState.finished);
	if (writableEnded && !writableFinished)
	return callback();
	}
	}

[ronag]

@BridgeAR: Any specific reason? This does make it a bit harder to read in my opinon.

[BridgeAR]

Code should IMO always only run when necessary. Right now some code is executed that is only needed in some situations.

[ronag]

This is not a hot path so readability should IMO come first. I think I can make a compromise 😄. PTAL.

[ronag]

The willFinish part here is the significant change. Before we only checked stream.writable.

The rest of the changes should be logically the same.

[ronag]

blocked pending #32954

[ronag]

closed in favor of #32968