http2: implement capture rection for 'request' and 'stream' events

PR-URL: #27867
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Jeremiah Senkpiel <fishrock123@rocketmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Michaël Zasso <targos@protonmail.com>

Author: mcollina

lib/internal/http2/core.js

@@ -1414,6 +1414,17 @@ class Http2Session extends EventEmitter {
 this[kMaybeDestroy]();
 }
 
+[EventEmitter.captureRejectionSymbol](err,event, ...args){
+switch(event){
+case'stream':
+const[stream]=args;
+stream.destroy(err);
+break;
+default:
+this.destroy(err);
+}
+}
+
 // Destroy the session if:
 // * error is not undefined/null
 // * session is closed and there are no more pending or open streams
@@ -2905,6 +2916,40 @@ class Http2Server extends NETServer {
 }
 }
 
+Http2Server.prototype[EventEmitter.captureRejectionSymbol]=function(
+err,event, ...args){
+
+switch(event){
+case'stream':
+// TODO(mcollina): we might want to match this with what we do on
+// the compat side.
+const[stream]=args;
+if(stream.sentHeaders){
+stream.destroy(err);
+}else{
+stream.respond({[HTTP2_HEADER_STATUS]: 500});
+stream.end();
+}
+break;
+case'request':
+const[,res]=args;
+if(!res.headersSent&&!res.finished){
+// Don't leak headers.
+for(constnameofres.getHeaderNames()){
+res.removeHeader(name);
+}
+res.statusCode=500;
+res.end(http.STATUS_CODES[500]);
+}else{
+res.destroy();
+}
+break;
+default:
+net.Server.prototype[EventEmitter.captureRejectionSymbol]
+.call(this,err,event, ...args);
+}
+};
+
 functionsetupCompat(ev){
 if(ev==='request'){
 this.removeListener('newListener',setupCompat);

test/parallel/test-http2-capture-rejection.js

@@ -0,0 +1,152 @@
+'use strict';
+
+constcommon=require('../common');
+if(!common.hasCrypto)
+common.skip('missing crypto');
+
+constassert=require('assert');
+constevents=require('events');
+const{ createServer, connect }=require('http2');
+
+events.captureRejections=true;
+
+{
+// Test error thrown in the server 'stream' event,
+// after a respond()
+
+constserver=createServer();
+server.on('stream',common.mustCall(async(stream)=>{
+server.close();
+
+stream.respond({':status': 200});
+
+const_err=newError('kaboom');
+stream.on('error',common.mustCall((err)=>{
+assert.strictEqual(err,_err);
+}));
+throw_err;
+}));
+
+server.listen(0,common.mustCall(()=>{
+const{ port }=server.address();
+constsession=connect(`http://localhost:${port}`);
+
+constreq=session.request();
+
+req.on('error',common.mustCall((err)=>{
+assert.strictEqual(err.code,'ERR_HTTP2_STREAM_ERROR');
+}));
+
+req.on('close',common.mustCall(()=>{
+session.close();
+}));
+}));
+}
+
+{
+// Test error thrown in the server 'stream' event,
+// before a respond().
+
+constserver=createServer();
+server.on('stream',common.mustCall(async(stream)=>{
+server.close();
+
+stream.on('error',common.mustNotCall());
+
+thrownewError('kaboom');
+}));
+
+server.listen(0,common.mustCall(()=>{
+const{ port }=server.address();
+constsession=connect(`http://localhost:${port}`);
+
+constreq=session.request();
+
+req.on('response',common.mustCall((headers)=>{
+assert.strictEqual(headers[':status'],500);
+}));
+
+req.on('close',common.mustCall(()=>{
+session.close();
+}));
+}));
+}
+
+
+{
+// Test error thrown in 'request' event
+
+constserver=createServer(common.mustCall(async(req,res)=>{
+server.close();
+res.setHeader('content-type','application/json');
+const_err=newError('kaboom');
+throw_err;
+}));
+
+server.listen(0,common.mustCall(()=>{
+const{ port }=server.address();
+constsession=connect(`http://localhost:${port}`);
+
+constreq=session.request();
+
+req.on('response',common.mustCall((headers)=>{
+assert.strictEqual(headers[':status'],500);
+assert.strictEqual(Object.hasOwnProperty.call(headers,'content-type'),
+false);
+}));
+
+req.on('close',common.mustCall(()=>{
+session.close();
+}));
+
+req.resume();
+}));
+}
+
+{
+// Test error thrown in the client 'stream' event
+
+constserver=createServer();
+server.on('stream',common.mustCall(async(stream)=>{
+const{ port }=server.address();
+
+server.close();
+
+stream.pushStream({
+':scheme': 'http',
+':path': '/foobar',
+':authority': `localhost:${port}`,
+},common.mustCall((err,push)=>{
+push.respond({
+'content-type': 'text/html',
+':status': 200
+});
+push.end('pushed by the server');
+
+stream.end('test');
+}));
+
+stream.respond({
+':status': 200
+});
+}));
+
+server.listen(0,common.mustCall(()=>{
+const{ port }=server.address();
+constsession=connect(`http://localhost:${port}`);
+
+constreq=session.request();
+
+session.on('stream',common.mustCall(async(stream)=>{
+session.close();
+
+const_err=newError('kaboom');
+stream.on('error',common.mustCall((err)=>{
+assert.strictEqual(err,_err);
+}));
+throw_err;
+}));
+
+req.end();
+}));
+}