stream: don't emit finish after destroy

PR-URL: https://github.com/nodejs/node/pull/40852
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

lib/internal/streams/writable.js

@@ -650,6 +650,7 @@ Writable.prototype.end = function(chunk, encoding, cb) {
 
 function needFinish(state) {
   return (state.ending &&
+          !state.destroyed &&
           state.constructed &&
           state.length === 0 &&
           !state.errored &&
@@ -732,11 +733,18 @@ function prefinish(stream, state) {
 function finishMaybe(stream, state, sync) {
   if (needFinish(state)) {
     prefinish(stream, state);
-    if (state.pendingcb === 0 && needFinish(state)) {
-      state.pendingcb++;
+    if (state.pendingcb === 0) {
       if (sync) {
-        process.nextTick(finish, stream, state);
-      } else {
+        state.pendingcb++;
+        process.nextTick((stream, state) => {
+          if (needFinish(state)) {
+            finish(stream, state);
+          } else {
+            state.pendingcb--;
+          }
+        }, stream, state);
+      } else if (needFinish(state)) {
+        state.pendingcb++;
         finish(stream, state);
       }
     }

test/parallel/test-http2-client-upload.js

@@ -22,7 +22,7 @@ fs.readFile(loc, common.mustSucceed((data) => {
   const server = http2.createServer();
   let client;
 
-  const countdown = new Countdown(3, () => {
+  const countdown = new Countdown(2, () => {
     server.close();
     client.close();
   });
@@ -50,7 +50,6 @@ fs.readFile(loc, common.mustSucceed((data) => {
     req.resume();
     req.on('end', common.mustCall());
 
-    req.on('finish', () => countdown.dec());
     const str = fs.createReadStream(loc);
     str.on('end', common.mustCall());
     str.on('close', () => countdown.dec());

test/parallel/test-http2-compat-socket-set.js

@@ -73,16 +73,6 @@ server.on('request', common.mustCall(function(request, response) {
   assert.throws(() => request.socket.pause = noop, errMsg);
   assert.throws(() => request.socket.resume = noop, errMsg);
 
-  request.stream.on('finish', common.mustCall(() => {
-    setImmediate(() => {
-      request.socket.setTimeout = noop;
-      assert.strictEqual(request.stream.setTimeout, noop);
-
-      assert.strictEqual(request.stream._isProcessing, undefined);
-      request.socket._isProcessing = true;
-      assert.strictEqual(request.stream._isProcessing, true);
-    });
-  }));
   response.stream.destroy();
 }));
 

test/parallel/test-stream-duplex-destroy.js

@@ -125,7 +125,7 @@ const assert = require('assert');
   duplex.removeListener('end', fail);
   duplex.removeListener('finish', fail);
   duplex.on('end', common.mustNotCall());
-  duplex.on('finish', common.mustCall());
+  duplex.on('finish', common.mustNotCall());
   assert.strictEqual(duplex.destroyed, true);
 }
 

test/parallel/test-stream-transform-destroy.js

@@ -117,7 +117,7 @@ const assert = require('assert');
   transform.removeListener('end', fail);
   transform.removeListener('finish', fail);
   transform.on('end', common.mustCall());
-  transform.on('finish', common.mustCall());
+  transform.on('finish', common.mustNotCall());
 }
 
 {

test/parallel/test-stream-writable-destroy.js

@@ -124,8 +124,6 @@ const assert = require('assert');
 
   write.destroy();
 
-  write.removeListener('finish', fail);
-  write.on('finish', common.mustCall());
   assert.strictEqual(write.destroyed, true);
 }
 

test/parallel/test-stream-writable-finish-destroyed.js

@@ -31,3 +31,13 @@ const { Writable } = require('stream');
   w.write('asd');
   w.destroy();
 }
+
+{
+  const w = new Writable({
+    write() {
+    }
+  });
+  w.on('finish', common.mustNotCall());
+  w.end();
+  w.destroy();
+}

test/parallel/test-stream-write-destroy.js

@@ -20,9 +20,7 @@ for (const withPendingData of [ false, true ]) {
 
     let chunksWritten = 0;
     let drains = 0;
-    let finished = false;
     w.on('drain', () => drains++);
-    w.on('finish', () => finished = true);
 
     function onWrite(err) {
       if (err) {
@@ -60,9 +58,5 @@ for (const withPendingData of [ false, true ]) {
     assert.strictEqual(chunksWritten, useEnd && !withPendingData ? 1 : 2);
     assert.strictEqual(callbacks.length, 0);
     assert.strictEqual(drains, 1);
-
-    // When we used `.end()`, we see the 'finished' event if and only if
-    // we actually finished processing the write queue.
-    assert.strictEqual(finished, !withPendingData && useEnd);
   }
 }