events: implement captureRejections for async handlers

doc/api/events.md

@@ -155,9 +155,66 @@ myEmitter.emit('error', new Error('whoops!'));
 // Prints: whoops! there was an error
 ```
 
+## Capture Rejections of Promises
+
+> Stability: 1 - captureRejections is experimental.
+
+Using `async` functions with event handlers is problematic, because it
+can lead to an unhandled rejection in case of a thrown exception:
+
+```js
+const ee = new EventEmitter();
+ee.on('something', async (value) => {
+  throw new Error('kaboom');
+});
+```
+
+The `captureRejections` option in the `EventEmitter` constructor or the global
+setting change this behavior, installing a `.then(undefined, handler)`
+handler on the `Promise`. This handler routes the exception
+asynchronously to the [`Symbol.for('nodejs.rejection')`][rejection] method
+if there is one, or to [`'error'`][error] event handler if there is none.
+
+```js
+const ee1 = new EventEmitter({ captureRejections: true });
+ee1.on('something', async (value) => {
+  throw new Error('kaboom');
+});
+
+ee1.on('error', console.log);
+
+const ee2 = new EventEmitter({ captureRejections: true });
+ee2.on('something', async (value) => {
+  throw new Error('kaboom');
+});
+
+ee2[Symbol.for('nodejs.rejection')] = console.log;
+```
+
+Setting `EventEmitter.captureRejections = true` will change the default for all
+new instances of `EventEmitter`.
+
+```js
+EventEmitter.captureRejections = true;
+const ee1 = new EventEmitter();
+ee1.on('something', async (value) => {
+  throw new Error('kaboom');
+});
+
+ee1.on('error', console.log);
+```
+
+The `'error'` events that are generated by the `captureRejections` behavior
+do not have a catch handler to avoid infinite error loops: the
+recommendation is to **not use `async` functions as `'error'` event handlers**.
+
 ## Class: EventEmitter
 <!-- YAML
 added: v0.1.26
+changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/27867
+    description: Added captureRejections option.
 -->
 
 The `EventEmitter` class is defined and exposed by the `events` module:
@@ -169,6 +226,12 @@ const EventEmitter = require('events');
 All `EventEmitter`s emit the event `'newListener'` when new listeners are
 added and `'removeListener'` when existing listeners are removed.
 
+It supports the following option:
+
+* `captureRejections` {boolean} It enables
+  [automatic capturing of promise rejection][capturerejections].
+  Default: `false`.
+
 ### Event: 'newListener'
 <!-- YAML
 added: v0.1.26
@@ -694,6 +757,42 @@ newListeners[0]();
 emitter.emit('log');
 ```
 
+### emitter\[Symbol.for('nodejs.rejection')\](err, eventName\[, ...args\])
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - captureRejections is experimental.
+
+* `err` Error
+* `eventName` {string|symbol}
+* `...args` {any}
+
+The `Symbol.for('nodejs.rejection')` method is called in case a
+promise rejection happens when emitting an event and
+[`captureRejections`][capturerejections] is enabled on the emitter.
+It is possible to use [`events.captureRejectionSymbol`][rejectionsymbol] in
+place of `Symbol.for('nodejs.rejection')`.
+
+```js
+const { EventEmitter, captureRejectionSymbol } = require('events');
+
+class MyClass extends EventEmitter {
+  constructor() {
+    super({ captureRejections: true });
+  }
+
+  [captureRejectionSymbol](err, event, ...args) {
+    console.log('rejection happened for', event, 'with', err, ...args);
+    this.destroy(err);
+  }
+
+  destroy(err) {
+    // Tear the resource down here.
+  }
+}
+```
+
 ## events.once(emitter, name)
 <!-- YAML
 added: v11.13.0
@@ -740,6 +839,28 @@ async function run() {
 run();
 ```
 
+## events.captureRejections
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - captureRejections is experimental.
+
+Value: {boolean}
+
+Change the default `captureRejections` option on all new `EventEmitter` objects.
+
+## events.captureRejectionSymbol
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - captureRejections is experimental.
+
+Value: `Symbol.for('nodejs.rejection')`
+
+See how to write a custom [rejection handler][rejection].
+
 [WHATWG-EventTarget]: https://dom.spec.whatwg.org/#interface-eventtarget
 [`--trace-warnings`]: cli.html#cli_trace_warnings
 [`EventEmitter.defaultMaxListeners`]: #events_eventemitter_defaultmaxlisteners
@@ -751,3 +872,7 @@ run();
 [`net.Server`]: net.html#net_class_net_server
 [`process.on('warning')`]: process.html#process_event_warning
 [stream]: stream.html
+[capturerejections]: #events_capture_rejections_of_promises
+[rejection]: #events_emitter_symbol_for_nodejs_rejection_err_eventname_args
+[rejectionsymbol]: #events_events_capturerejectionsymbol
+[error]: #events_error_events

lib/_http_outgoing.js

@@ -32,6 +32,7 @@ const {
 
 const { getDefaultHighWaterMark } = require('internal/streams/state');
 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');
@@ -884,6 +885,11 @@ OutgoingMessage.prototype.pipe = function pipe() {
   this.emit('error', new ERR_STREAM_CANNOT_PIPE());
 };
 
+OutgoingMessage.prototype[EE.captureRejectionSymbol] =
+function(err, event) {
+  this.destroy(err);
+};
+
 module.exports = {
   OutgoingMessage
 };

lib/_http_server.js

@@ -27,6 +27,7 @@ const {
 } = primordials;
 
 const net = require('net');
+const EE = require('events');
 const assert = require('internal/assert');
 const {
   parsers,
@@ -351,6 +352,28 @@ Server.prototype.setTimeout = function setTimeout(msecs, callback) {
   return this;
 };
 
+Server.prototype[EE.captureRejectionSymbol] = function(
+  err, event, ...args) {
+
+  switch (event) {
+    case 'request':
+      const [ , res] = args;
+      if (!res.headersSent && !res.writableEnded) {
+        // Don't leak headers.
+        for (const name of res.getHeaderNames()) {
+          res.removeHeader(name);
+        }
+        res.statusCode = 500;
+        res.end(STATUS_CODES[500]);
+      } else {
+        res.destroy();
+      }
+      break;
+    default:
+      net.Server.prototype[Symbol.for('nodejs.rejection')]
+        .call(this, err, event, ...args);
+  }
+};
 
 function connectionListener(socket) {
   defaultTriggerAsyncIdScope(

lib/_stream_readable.js

@@ -190,7 +190,7 @@ function Readable(options) {
       this._destroy = options.destroy;
   }
 
-  Stream.call(this);
+  Stream.call(this, options);
 }
 
 ObjectDefineProperty(Readable.prototype, 'destroyed', {
@@ -233,6 +233,14 @@ Readable.prototype._destroy = function(err, cb) {
   cb(err);
 };
 
+Readable.prototype[EE.captureRejectionSymbol] = function(err) {
+  // TODO(mcollina): remove the destroyed if once errorEmitted lands in
+  // Readable.
+  if (!this.destroyed) {
+    this.destroy(err);
+  }
+};
+
 // Manually shove something into the read() buffer.
 // This returns true if the highWaterMark has not been hit yet,
 // similar to how Writable.write() returns true if you should

lib/_stream_writable.js

@@ -35,6 +35,7 @@ module.exports = Writable;
 Writable.WritableState = WritableState;
 
 const internalUtil = require('internal/util');
+const EE = require('events');
 const Stream = require('stream');
 const { Buffer } = require('buffer');
 const destroyImpl = require('internal/streams/destroy');
@@ -250,7 +251,7 @@ function Writable(options) {
       this._final = options.final;
   }
 
-  Stream.call(this);
+  Stream.call(this, options);
 }
 
 // Otherwise people can pipe Writable streams, which is just wrong.
@@ -808,3 +809,7 @@ Writable.prototype._undestroy = destroyImpl.undestroy;
 Writable.prototype._destroy = function(err, cb) {
   cb(err);
 };
+
+Writable.prototype[EE.captureRejectionSymbol] = function(err) {
+  this.destroy(err);
+};

lib/_tls_wrap.js

@@ -37,6 +37,7 @@ assertCrypto();
 const { setImmediate } = require('timers');
 const assert = require('internal/assert');
 const crypto = require('crypto');
+const EE = require('events');
 const net = require('net');
 const tls = require('tls');
 const common = require('_tls_common');
@@ -1282,6 +1283,19 @@ Server.prototype.addContext = function(servername, context) {
   this._contexts.push([re, tls.createSecureContext(context).context]);
 };
 
+Server.prototype[EE.captureRejectionSymbol] = function(
+  err, event, sock) {
+
+  switch (event) {
+    case 'secureConnection':
+      sock.destroy(err);
+      break;
+    default:
+      net.Server.prototype[Symbol.for('nodejs.rejection')]
+        .call(this, err, event, sock);
+  }
+};
+
 function SNICallback(servername, callback) {
   const contexts = this.server._contexts;