stream: call end(cb) callback after destroy

Don't deadlock calls to end(cb) after destroy(). This only
partly fixes the problem in order to minimize breakage.

Author: ronag

lib/_stream_writable.js

@@ -591,8 +591,20 @@ Writable.prototype.end = function(chunk, encoding, cb) {
     encoding = null;
   }
 
-  if (chunk !== null && chunk !== undefined)
+  if (chunk !== null && chunk !== undefined) {
     this.write(chunk, encoding);
+  } else if (state.destroyed) {
+    // TODO(ronag): Condition is for backwards compat.
+    if (state.errorEmitted || state.finished) {
+      const err = new ERR_STREAM_DESTROYED('end');
+      if (typeof cb === 'function') {
+        process.nextTick(cb, err);
+      }
+      // TODO(ronag): Disabled for backwards compat.
+      // errorOrDestroy(this, err, true);
+      return;
+    }
+  }
 
   // .end() fully uncorks
   if (state.corked) {

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

@@ -292,3 +292,33 @@ const assert = require('assert');
   }));
   write.uncork();
 }
+
+{
+  // Call end(cb) after error & destroy
+
+  const write = new Writable({
+    write(chunk, enc, cb) { cb(new Error('asd')); }
+  });
+  write.on('error', common.mustCall(() => {
+    write.destroy();
+    write.end(common.mustCall((err) => {
+      assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
+    }));
+  }));
+  write.write('asd');
+}
+
+{
+  // Call end(cb) after finish & destroy
+
+  const write = new Writable({
+    write(chunk, enc, cb) { cb(); }
+  });
+  write.on('finish', common.mustCall(() => {
+    write.destroy();
+    write.end(common.mustCall((err) => {
+      assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
+    }));
+  }));
+  write.end();
+}