http: don't emit error after close

ronag · ronag · commit 299b2fdb1068 · 2020-06-20T13:55:42.000+02:00
Refs: #33591
diff --git a/lib/_http_client.js b/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');
diff --git a/lib/_http_outgoing.js b/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;
   }
 
@@ -780,7 +787,9 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
   }
 
   if (chunk) {
-    write_(this, chunk, encoding, null, true);
+    if (!write_(this, chunk, encoding, null, true)) {
+      return;
+    }
   } else if (this.finished) {
     if (typeof callback === 'function') {
       if (!this.writableFinished) {
diff --git a/lib/_http_server.js b/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');
 }
 
diff --git a/lib/internal/http.js b/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
diff --git a/test/parallel/test-http-outgoing-destroy.js b/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());
 }
diff --git a/test/parallel/test-http-outgoing-destroyed.js b/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');
+});
diff --git a/test/parallel/test-http-server-write-after-end.js b/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();
+    }));
   }));
 }
 
diff --git a/test/parallel/test-http-server-write-end-after-end.js b/test/parallel/test-http-server-write-end-after-end.js
@@ -6,20 +6,26 @@ 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(() => {
+  setImmediate(() => {
+    // TODO: Known issues. OutgoingMessage does not always invoke callback
+    // in end. Also streams don't propagate write after end to callback.
     res.end('world');
-  }));
+    process.nextTick(() => {
+      server.close();
+    });
+    // res.end('world', common.mustCall((err) => {
+    //   common.expectsError({
+    //     code: 'ERR_STREAM_WRITE_AFTER_END',
+    //     name: 'Error'
+    //   })(err);
+    //   server.close();
+    // }));
+  });
 }
 
 server.listen(0, common.mustCall(() => {

Original file line number	Diff line number	Diff line change
`@@ -49,7 +49,7 @@ const Agent = require('_http_agent');`
`49`	`49`	`const { Buffer } = require('buffer');`
`50`	`50`	`const { defaultTriggerAsyncIdScope } = require('internal/async_hooks');`
`51`	`51`	`const { URL, urlToOptions, searchParamsSymbol } = require('internal/url');`
`52`		`-const { kOutHeaders, kNeedDrain } = require('internal/http');`
	`52`	`+const { kOutHeaders, kNeedDrain, kClosed } = require('internal/http');`
`53`	`53`	`const { connResetException, codes } = require('internal/errors');`
`54`	`54`	`const {`
`55`	`55`	`ERR_HTTP_HEADERS_SENT,`
`@@ -385,6 +385,7 @@ function _destroy(req, socket, err) {`
`385`	`385`	`if (err) {`
`386`	`386`	`req.emit('error', err);`
`387`	`387`	`}`
	`388`	`+ req[kClosed] = true;`
`388`	`389`	`req.emit('close');`
`389`	`390`	`}`
`390`	`391`	`}`
`@@ -427,6 +428,7 @@ function socketCloseListener() {`
`427`	`428`	`res.emit('error', connResetException('aborted'));`
`428`	`429`	`}`
`429`	`430`	`}`
	`431`	`+ req[kClosed] = true;`
`430`	`432`	`req.emit('close');`
`431`	`433`	`if (!res.aborted && res.readable) {`
`432`	`434`	`res.on('end', function() {`
`@@ -446,6 +448,7 @@ function socketCloseListener() {`
`446`	`448`	`req.socket._hadError = true;`
`447`	`449`	`req.emit('error', connResetException('socket hang up'));`
`448`	`450`	`}`
	`451`	`+ req[kClosed] = true;`
`449`	`452`	`req.emit('close');`
`450`	`453`	`}`
`451`	`454`
`@@ -550,6 +553,7 @@ function socketOnData(d) {`
`550`	`553`
`551`	`554`	`req.emit(eventName, res, socket, bodyHead);`
`552`	`555`	`req.destroyed = true;`
	`556`	`+ req[kClosed] = true;`
`553`	`557`	`req.emit('close');`
`554`	`558`	`} else {`
`555`	`559`	`// Requested Upgrade or used CONNECT method, but have no handler.`
`@@ -721,6 +725,7 @@ function requestOnPrefinish() {`
`721`	`725`	`}`
`722`	`726`
`723`	`727`	`function emitFreeNT(req) {`
	`728`	`+ req[kClosed] = true;`
`724`	`729`	`req.emit('close');`
`725`	`730`	`if (req.res) {`
`726`	`731`	`req.res.emit('close');`