stream: fix _final and 'prefinish' timing

This PR fixes a few different things:

The timing of 'prefinish' depends on whether or not
_final is defined. In on case the event is emitted
synchronously with end() and otherwise asynchronously.

_final is currently unecessarily called asynchronously
which forces implementors to use 'prefinish' as a hack
to emulate synchronous behaviour. Furthermore, this hack
is subtly broken due to the above issue.

The stream should not finish if errored or destroyed
synchronously during the prefinish stage.

Refs: nodejs#31401
Refs: nodejs#32763 (comment)

lib/_stream_transform.js

@@ -99,10 +99,10 @@ function Transform(options) {
       this._flush = options.flush;
   }
 
-  // TODO(ronag): Unfortunately _final is invoked asynchronously.
-  // Use `prefinish` hack. `prefinish` is emitted synchronously when
-  // and only when `_final` is not defined. Implementing `_final`
-  // to a Transform should be an error.
+  // When the writable side finishes, then flush out anything remaining.
+  // Backwards compat. Some Transform streams incorrectly implement _final
+  // instead of or in addition to _flush. By using 'prefinish' instead of
+  // implementing _final we continue supporting this unfortunate use case.
   this.on('prefinish', prefinish);
 }
 

lib/_stream_writable.js

@@ -629,24 +629,33 @@ function needFinish(state) {
 }
 
 function callFinal(stream, state) {
+  state.sync = true;
+  state.pendingcb++;
   stream._final((err) => {
     state.pendingcb--;
     if (err) {
-      errorOrDestroy(stream, err);
+      errorOrDestroy(stream, err, state.sync);
     } else {
+      // Backwards compat. Don't check needFinish() here.
+      // Some streams assume 'finish' will be emitted
+      // even if stream has been destroyed.
       state.prefinished = true;
       stream.emit('prefinish');
-      finishMaybe(stream, state);
+      // Backwards compat. Don't check state.sync here.
+      // Some stream assume 'finish' will be emitted
+      // asynchronously relative to _final callback.
+      state.pendingcb++;
+      process.nextTick(finish, stream, state);
     }
   });
+  state.sync = false;
 }
 
 function prefinish(stream, state) {
   if (!state.prefinished && !state.finalCalled) {
     if (typeof stream._final === 'function' && !state.destroyed) {
-      state.pendingcb++;
       state.finalCalled = true;
-      process.nextTick(callFinal, stream, state);
+      callFinal(stream, state);
     } else {
       state.prefinished = true;
       stream.emit('prefinish');
@@ -655,10 +664,9 @@ function prefinish(stream, state) {
 }
 
 function finishMaybe(stream, state, sync) {
-  const need = needFinish(state);
-  if (need) {
+  if (needFinish(state)) {
     prefinish(stream, state);
-    if (state.pendingcb === 0) {
+    if (state.pendingcb === 0 && needFinish(state)) {
       state.pendingcb++;
       if (sync) {
         process.nextTick(finish, stream, state);
@@ -667,7 +675,6 @@ function finishMaybe(stream, state, sync) {
       }
     }
   }
-  return need;
 }
 
 function finish(stream, state) {

lib/internal/http2/core.js

@@ -1710,11 +1710,14 @@ function streamOnPause() {
 }
 
 function afterShutdown(status) {
+  const stream = this.handle[kOwner];
+  if (stream) {
+    stream.on('finish', () => {
+      stream[kMaybeDestroy]();
+    });
+  }
   // Currently this status value is unused
   this.callback();
-  const stream = this.handle[kOwner];
-  if (stream)
-    stream[kMaybeDestroy]();
 }
 
 function finishSendTrailers(stream, headersList) {

test/parallel/test-stream-transform-final-sync.js

@@ -82,15 +82,15 @@ const t = new stream.Transform({
     process.nextTick(function() {
       state++;
       // fluchCallback part 2
-      assert.strictEqual(state, 15);
+      assert.strictEqual(state, 13);
       done();
     });
   }, 1)
 });
 t.on('finish', common.mustCall(function() {
   state++;
   // finishListener
-  assert.strictEqual(state, 13);
+  assert.strictEqual(state, 14);
 }, 1));
 t.on('end', common.mustCall(function() {
   state++;
@@ -106,5 +106,5 @@ t.write(4);
 t.end(7, common.mustCall(function() {
   state++;
   // endMethodCallback
-  assert.strictEqual(state, 14);
+  assert.strictEqual(state, 15);
 }, 1));

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

@@ -84,15 +84,15 @@ const t = new stream.Transform({
     process.nextTick(function() {
       state++;
       // flushCallback part 2
-      assert.strictEqual(state, 15);
+      assert.strictEqual(state, 13);
       done();
     });
   }, 1)
 });
 t.on('finish', common.mustCall(function() {
   state++;
   // finishListener
-  assert.strictEqual(state, 13);
+  assert.strictEqual(state, 14);
 }, 1));
 t.on('end', common.mustCall(function() {
   state++;
@@ -108,5 +108,5 @@ t.write(4);
 t.end(7, common.mustCall(function() {
   state++;
   // endMethodCallback
-  assert.strictEqual(state, 14);
+  assert.strictEqual(state, 15);
 }, 1));

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

@@ -41,3 +41,59 @@ const assert = require('assert');
   w.end();
   w.on('finish', common.mustCall());
 }
+
+{
+  // Emit prefinish synchronously
+
+  const w = new Writable({
+    write(chunk, encoding, cb) {
+      cb();
+    }
+  });
+
+  let sync = true;
+  w.on('prefinish', common.mustCall(() => {
+    assert.strictEqual(sync, true);
+  }));
+  w.end();
+  sync = false;
+}
+
+{
+  // Emit prefinish synchronously w/ final
+
+  const w = new Writable({
+    write(chunk, encoding, cb) {
+      cb();
+    },
+    final(cb) {
+      cb();
+    }
+  });
+
+  let sync = true;
+  w.on('prefinish', common.mustCall(() => {
+    assert.strictEqual(sync, true);
+  }));
+  w.end();
+  sync = false;
+}
+
+
+{
+  // Call _final synchronouslyl
+
+  let sync = true;
+  const w = new Writable({
+    write(chunk, encoding, cb) {
+      cb();
+    },
+    final: common.mustCall((cb) => {
+      assert.strictEqual(sync, true);
+      cb();
+    })
+  });
+
+  w.end();
+  sync = false;
+}