stream: call _final synchronously

Author: ronag

lib/_stream_writable.js

@@ -665,24 +665,26 @@ function needFinish(state) {
 }
 
 function callFinal(stream, state) {
+  state.sync = true;
   stream._final((err) => {
     state.pendingcb--;
     if (err) {
-      errorOrDestroy(stream, err);
+      errorOrDestroy(stream, err, state.sync);
     } else {
       state.prefinished = true;
       stream.emit('prefinish');
-      finishMaybe(stream, state);
+      finishMaybe(stream, state, state.sync);
     }
   });
+  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');
@@ -691,10 +693,11 @@ function prefinish(stream, state) {
 }
 
 function finishMaybe(stream, state, sync) {
-  const need = needFinish(state);
+  let need = needFinish(state);
   if (need) {
     prefinish(stream, state);
-    if (state.pendingcb === 0) {
+    need = needFinish(state);
+    if (state.pendingcb === 0 && need) {
       state.pendingcb++;
       if (sync) {
         process.nextTick(finish, stream, state);

lib/internal/fs/streams.js

@@ -366,7 +366,7 @@ ObjectSetPrototypeOf(WriteStream, Writable);
 WriteStream.prototype._final = function(callback) {
   if (typeof this.fd !== 'number') {
     return this.once('open', function() {
-      this._final(callback);
+      process.nextTick(() => this._final(callback));
     });
   }
 

lib/internal/http2/core.js

@@ -2004,16 +2004,21 @@ class Http2Stream extends Duplex {
   _final(cb) {
     const handle = this[kHandle];
     if (this.pending) {
-      this.once('ready', () => this._final(cb));
+      this.once('ready', function() {
+        process.nextTick(() => this._final(cb));
+      });
     } else if (handle !== undefined) {
       debugStreamObj(this, '_final shutting down');
-      const req = new ShutdownWrap();
-      req.oncomplete = afterShutdown;
-      req.callback = cb;
-      req.handle = handle;
-      const err = handle.shutdown(req);
-      if (err === 1)  // synchronous finish
-        return afterShutdown.call(req, 0);
+      // TODO: Why does this need to be in nextTick?
+      process.nextTick(() => {
+        const req = new ShutdownWrap();
+        req.oncomplete = afterShutdown;
+        req.callback = cb;
+        req.handle = handle;
+        const err = handle.shutdown(req);
+        if (err === 1)  // synchronous finish
+          return afterShutdown.call(req, 0);
+      });
     } else {
       cb();
     }

lib/internal/streams/destroy.js

@@ -119,7 +119,7 @@ function undestroy() {
   }
 }
 
-function errorOrDestroy(stream, err) {
+function errorOrDestroy(stream, err, sync) {
   // We have tests that rely on errors being emitted
   // in the same tick, so changing this is semver major.
   // For now when you opt-in to autoDestroy we allow
@@ -138,7 +138,12 @@ function errorOrDestroy(stream, err) {
     if (r) {
       r.errored = true;
     }
-    emitErrorNT(stream, err);
+
+    if (sync) {
+      process.nextTick(emitErrorNT, stream, err);
+    } else {
+      emitErrorNT(stream, err);
+    }
   }
 }
 

lib/net.js

@@ -398,7 +398,9 @@ Socket.prototype._final = function(cb) {
   // If still connecting - defer handling `_final` until 'connect' will happen
   if (this.pending) {
     debug('_final: not yet connected');
-    return this.once('connect', () => this._final(cb));
+    return this.once('connect', function() {
+      process.nextTick(() => this._final(cb));
+    });
   }
 
   if (!this._handle)

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));