http: don't emit error for stream destroyed

lib/_http_client.js

@@ -49,7 +49,7 @@ const Agent = require('_http_agent');
 const { Buffer } = require('buffer');
 const { defaultTriggerAsyncIdScope } = require('internal/async_hooks');
 const { URL, urlToOptions, searchParamsSymbol } = require('internal/url');
-const { kOutHeaders, kNeedDrain } = require('internal/http');
+const { kOutHeaders, kNeedDrain, kClosed } = require('internal/http');
 const { connResetException, codes } = require('internal/errors');
 const {
   ERR_HTTP_HEADERS_SENT,
@@ -385,6 +385,7 @@ function _destroy(req, socket, err) {
     if (err) {
       req.emit('error', err);
     }
+    req[kClosed] = true;
     req.emit('close');
   }
 }
@@ -427,6 +428,7 @@ function socketCloseListener() {
         res.emit('error', connResetException('aborted'));
       }
     }
+    req[kClosed] = true;
     req.emit('close');
     if (!res.aborted && res.readable) {
       res.on('end', function() {
@@ -446,6 +448,7 @@ function socketCloseListener() {
       req.socket._hadError = true;
       req.emit('error', connResetException('socket hang up'));
     }
+    req[kClosed] = true;
     req.emit('close');
   }
 
@@ -550,6 +553,7 @@ function socketOnData(d) {
 
       req.emit(eventName, res, socket, bodyHead);
       req.destroyed = true;
+      req[kClosed] = true;
       req.emit('close');
     } else {
       // Requested Upgrade or used CONNECT method, but have no handler.
@@ -721,6 +725,7 @@ function requestOnPrefinish() {
 }
 
 function emitFreeNT(req) {
+  req[kClosed] = true;
   req.emit('close');
   if (req.res) {
     req.res.emit('close');

lib/_http_outgoing.js

@@ -36,7 +36,7 @@ const assert = require('internal/assert');
 const EE = require('events');
 const Stream = require('stream');
 const internalUtil = require('internal/util');
-const { kOutHeaders, utcDate, kNeedDrain } = require('internal/http');
+const { kOutHeaders, utcDate, kNeedDrain, kClosed } = require('internal/http');
 const { Buffer } = require('buffer');
 const common = require('_http_common');
 const checkIsHttpToken = common._checkIsHttpToken;
@@ -117,6 +117,7 @@ function OutgoingMessage() {
   this.finished = false;
   this._headerSent = false;
   this[kCorked] = 0;
+  this[kClosed] = false;
 
   this.socket = null;
   this._header = null;
@@ -663,7 +664,9 @@ function onError(msg, err, callback) {
 
 function emitErrorNt(msg, err, callback) {
   callback(err);
-  if (typeof msg.emit === 'function') msg.emit('error', err);
+  if (typeof msg.emit === 'function' && !msg[kClosed]) {
+    msg.emit('error', err);
+  }
 }
 
 function write_(msg, chunk, encoding, callback, fromEnd) {
@@ -690,7 +693,11 @@ function write_(msg, chunk, encoding, callback, fromEnd) {
   }
 
   if (err) {
-    onError(msg, err, callback);
+    if (!msg.destroyed) {
+      onError(msg, err, callback);
+    } else {
+      process.nextTick(callback, err);
+    }
     return false;
   }
 

lib/_http_server.js

@@ -49,6 +49,7 @@ const { OutgoingMessage } = require('_http_outgoing');
 const {
   kOutHeaders,
   kNeedDrain,
+  kClosed,
   emitStatistics
 } = require('internal/http');
 const {
@@ -212,6 +213,7 @@ function onServerResponseClose() {
   // Fortunately, that requires only a single if check. :-)
   if (this._httpMessage) {
     this._httpMessage.destroyed = true;
+    this._httpMessage[kClosed] = true;
     this._httpMessage.emit('close');
   }
 }
@@ -729,6 +731,7 @@ function resOnFinish(req, res, socket, state, server) {
 
 function emitCloseNT(self) {
   self.destroyed = true;
+  self[kClosed] = true;
   self.emit('close');
 }
 

lib/internal/http.js

@@ -51,6 +51,7 @@ function emitStatistics(statistics) {
 module.exports = {
   kOutHeaders: Symbol('kOutHeaders'),
   kNeedDrain: Symbol('kNeedDrain'),
+  kClosed: Symbol('kClosed'),
   nowDate,
   utcDate,
   emitStatistics

test/parallel/test-http-outgoing-destroy.js

@@ -10,13 +10,8 @@ const OutgoingMessage = http.OutgoingMessage;
   assert.strictEqual(msg.destroyed, false);
   msg.destroy();
   assert.strictEqual(msg.destroyed, true);
-  let callbackCalled = false;
   msg.write('asd', common.mustCall((err) => {
     assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
-    callbackCalled = true;
-  }));
-  msg.on('error', common.mustCall((err) => {
-    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
-    assert.strictEqual(callbackCalled, true);
   }));
+  msg.on('error', common.mustNotCall());
 }

test/parallel/test-http-outgoing-destroyed.js

@@ -0,0 +1,24 @@
+'use strict';
+const common = require('../common');
+const http = require('http');
+
+const server = http.createServer(common.mustCall((req, res) => {
+  req.pipe(res);
+  res.on('error', common.mustNotCall());
+  res.on('close', common.mustCall(() => {
+    res.end('asd');
+    process.nextTick(() => {
+      server.close();
+    });
+  }));
+})).listen(0, () => {
+  http
+    .request({
+      port: server.address().port,
+      method: 'PUT'
+    })
+    .on('response', (res) => {
+      res.destroy();
+    })
+    .write('asd');
+});

test/parallel/test-http-server-write-after-end.js

@@ -8,19 +8,19 @@ const http = require('http');
 const server = http.createServer(handle);
 
 function handle(req, res) {
-  res.on('error', common.mustCall((err) => {
-    common.expectsError({
-      code: 'ERR_STREAM_WRITE_AFTER_END',
-      name: 'Error'
-    })(err);
-    server.close();
-  }));
+  res.on('error', common.mustNotCall());
 
   res.write('hello');
   res.end();
 
   setImmediate(common.mustCall(() => {
-    res.write('world');
+    res.write('world', common.mustCall((err) => {
+      common.expectsError({
+        code: 'ERR_STREAM_WRITE_AFTER_END',
+        name: 'Error'
+      })(err);
+      server.close();
+    }));
   }));
 }
 

test/parallel/test-http-server-write-end-after-end.js

@@ -6,19 +6,23 @@ const http = require('http');
 const server = http.createServer(handle);
 
 function handle(req, res) {
-  res.on('error', common.mustCall((err) => {
-    common.expectsError({
-      code: 'ERR_STREAM_WRITE_AFTER_END',
-      name: 'Error'
-    })(err);
-    server.close();
-  }));
+  res.on('error', common.mustNotCall());
 
   res.write('hello');
   res.end();
 
   setImmediate(common.mustCall(() => {
     res.end('world');
+    process.nextTick(() => {
+      server.close();
+    });
+    res.write('world', common.mustCall((err) => {
+      common.expectsError({
+        code: 'ERR_STREAM_WRITE_AFTER_END',
+        name: 'Error'
+      })(err);
+      server.close();
+    }));
   }));
 }