http2: fix responses to long payload reqs

When a request with a long payload is received, http2 does
not allow a response that does not process all the incoming
payload. Add a conditional Http2Stream.close call that runs
only if the user hasn't attempted to read the stream.
PR-URL: #20084Fixes: #20060
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

Author: apapirovski

lib/internal/http2/core.js

@@ -206,6 +206,7 @@ const STREAM_FLAGS_CLOSED = 0x2;
 constSTREAM_FLAGS_HEADERS_SENT=0x4;
 constSTREAM_FLAGS_HEAD_REQUEST=0x8;
 constSTREAM_FLAGS_ABORTED=0x10;
+constSTREAM_FLAGS_HAS_TRAILERS=0x20;
 
 constSESSION_FLAGS_PENDING=0x0;
 constSESSION_FLAGS_READY=0x1;
@@ -330,26 +331,13 @@ function onStreamClose(code) {
 if(stream.destroyed)
 return;
 
-conststate=stream[kState];
-
 debug(`Http2Stream ${stream[kID]} [Http2Session `+
 `${sessionName(stream[kSession][kType])}]: closed with code ${code}`);
 
-if(!stream.closed){
-// Clear timeout and remove timeout listeners
-stream.setTimeout(0);
-stream.removeAllListeners('timeout');
+if(!stream.closed)
+closeStream(stream,code,false);
 
-// Set the state flags
-state.flags|=STREAM_FLAGS_CLOSED;
-state.rstCode=code;
-
-// Close the writable side of the stream
-abort(stream);
-stream.end();
-}
-
-state.fd=-1;
+stream[kState].fd=-1;
 // Defer destroy we actually emit end.
 if(stream._readableState.endEmitted||code!==NGHTTP2_NO_ERROR){
 // If errored or ended, we can destroy immediately.
@@ -504,7 +492,7 @@ function requestOnConnect(headers, options) {
 
 // At this point, the stream should have already been destroyed during
 // the session.destroy() method. Do nothing else.
-if(session.destroyed)
+if(session===undefined||session.destroyed)
 return;
 
 // If the session was closed while waiting for the connect, destroy
@@ -1412,6 +1400,9 @@ class ClientHttp2Session extends Http2Session {
 if(options.endStream)
 stream.end();
 
+if(options.waitForTrailers)
+stream[kState].flags|=STREAM_FLAGS_HAS_TRAILERS;
+
 constonConnect=requestOnConnect.bind(stream,headersList,options);
 if(this.connecting){
 this.on('connect',onConnect);
@@ -1445,32 +1436,70 @@ function afterDoStreamWrite(status, handle) {
 }
 
 functionstreamOnResume(){
-if(!this.destroyed&&!this.pending)
+if(!this.destroyed&&!this.pending){
+if(!this[kState].didRead)
+this[kState].didRead=true;
 this[kHandle].readStart();
+}
 }
 
 functionstreamOnPause(){
 if(!this.destroyed&&!this.pending)
 this[kHandle].readStop();
 }
 
-// If the writable side of the Http2Stream is still open, emit the
-// 'aborted' event and set the aborted flag.
-functionabort(stream){
-if(!stream.aborted&&
-!(stream._writableState.ended||stream._writableState.ending)){
-stream[kState].flags|=STREAM_FLAGS_ABORTED;
-stream.emit('aborted');
-}
-}
-
 functionafterShutdown(){
 this.callback();
 conststream=this.handle[kOwner];
 if(stream)
 stream[kMaybeDestroy]();
 }
 
+functioncloseStream(stream,code,shouldSubmitRstStream=true){
+conststate=stream[kState];
+state.flags|=STREAM_FLAGS_CLOSED;
+state.rstCode=code;
+
+// Clear timeout and remove timeout listeners
+stream.setTimeout(0);
+stream.removeAllListeners('timeout');
+
+const{ ending, finished }=stream._writableState;
+
+if(!ending){
+// If the writable side of the Http2Stream is still open, emit the
+// 'aborted' event and set the aborted flag.
+if(!stream.aborted){
+state.flags|=STREAM_FLAGS_ABORTED;
+stream.emit('aborted');
+}
+
+// Close the writable side.
+stream.end();
+}
+
+if(shouldSubmitRstStream){
+constfinishFn=finishCloseStream.bind(stream,code);
+if(!ending||finished||code!==NGHTTP2_NO_ERROR)
+finishFn();
+else
+stream.once('finish',finishFn);
+}
+}
+
+functionfinishCloseStream(code){
+constrstStreamFn=submitRstStream.bind(this,code);
+// If the handle has not yet been assigned, queue up the request to
+// ensure that the RST_STREAM frame is sent after the stream ID has
+// been determined.
+if(this.pending){
+this.push(null);
+this.once('ready',rstStreamFn);
+return;
+}
+rstStreamFn();
+}
+
 // An Http2Stream is a Duplex stream that is backed by a
 // node::http2::Http2Stream handle implementing StreamBase.
 classHttp2StreamextendsDuplex{
@@ -1490,6 +1519,7 @@ class Http2Stream extends Duplex {
 this[kTimeout]=null;
 
 this[kState]={
+didRead: false,
 flags: STREAM_FLAGS_PENDING,
 rstCode: NGHTTP2_NO_ERROR,
 writeQueueSize: 0,
@@ -1756,6 +1786,8 @@ class Http2Stream extends Duplex {
 throwheadersList;
 this[kSentTrailers]=headers;
 
+this[kState].flags&=~STREAM_FLAGS_HAS_TRAILERS;
+
 constret=this[kHandle].trailers(headersList);
 if(ret<0)
 this.destroy(newNghttpError(ret));
@@ -1786,38 +1818,13 @@ class Http2Stream extends Duplex {
 if(callback!==undefined&&typeofcallback!=='function')
 thrownewERR_INVALID_CALLBACK();
 
-// Clear timeout and remove timeout listeners
-this.setTimeout(0);
-this.removeAllListeners('timeout');
-
-// Close the writable
-abort(this);
-this.end();
-
 if(this.closed)
 return;
 
-conststate=this[kState];
-state.flags|=STREAM_FLAGS_CLOSED;
-state.rstCode=code;
-
-if(callback!==undefined){
+if(callback!==undefined)
 this.once('close',callback);
-}
-
-if(this[kHandle]===undefined)
-return;
 
-constrstStreamFn=submitRstStream.bind(this,code);
-// If the handle has not yet been assigned, queue up the request to
-// ensure that the RST_STREAM frame is sent after the stream ID has
-// been determined.
-if(this.pending){
-this.push(null);
-this.once('ready',rstStreamFn);
-return;
-}
-rstStreamFn();
+closeStream(this,code);
 }
 
 // Called by this.destroy().
@@ -1832,26 +1839,19 @@ class Http2Stream extends Duplex {
 debug(`Http2Stream ${this[kID]||'<pending>'} [Http2Session `+
 `${sessionName(session[kType])}]: destroying stream`);
 conststate=this[kState];
-constcode=state.rstCode=
-err!=null ?
-NGHTTP2_INTERNAL_ERROR :
-state.rstCode||NGHTTP2_NO_ERROR;
-if(handle!==undefined){
-// If the handle exists, we need to close, then destroy the handle
-this.close(code);
-if(!this._readableState.ended&&!this._readableState.ending)
-this.push(null);
+constcode=err!=null ?
+NGHTTP2_INTERNAL_ERROR : (state.rstCode||NGHTTP2_NO_ERROR);
+
+consthasHandle=handle!==undefined;
+
+if(!this.closed)
+closeStream(this,code,hasHandle);
+this.push(null);
+
+if(hasHandle){
 handle.destroy();
 session[kState].streams.delete(id);
 }else{
-// Clear timeout and remove timeout listeners
-this.setTimeout(0);
-this.removeAllListeners('timeout');
-
-state.flags|=STREAM_FLAGS_CLOSED;
-abort(this);
-this.end();
-this.push(null);
 session[kState].pendingStreams.delete(this);
 }
 
@@ -1884,13 +1884,23 @@ class Http2Stream extends Duplex {
 }
 
 // TODO(mcollina): remove usage of _*State properties
-if(this._readableState.ended&&
-this._writableState.ended&&
-this._writableState.pendingcb===0&&
-this.closed){
-this.destroy();
-// This should return, but eslint complains.
-// return
+if(this._writableState.ended&&this._writableState.pendingcb===0){
+if(this._readableState.ended&&this.closed){
+this.destroy();
+return;
+}
+
+// We've submitted a response from our server session, have not attempted
+// to process any incoming data, and have no trailers. This means we can
+// attempt to gracefully close the session.
+conststate=this[kState];
+if(this.headersSent&&
+this[kSession][kType]===NGHTTP2_SESSION_SERVER&&
+!(state.flags&STREAM_FLAGS_HAS_TRAILERS)&&
+!state.didRead&&
+!this._readableState.resumeScheduled){
+this.close();
+}
 }
 }
 }
@@ -2095,7 +2105,6 @@ function afterOpen(session, options, headers, streamOptions, err, fd) {
 }
 if(this.destroyed||this.closed){
 tryClose(fd);
-abort(this);
 return;
 }
 state.fd=fd;
@@ -2224,8 +2233,10 @@ class ServerHttp2Stream extends Http2Stream {
 if(options.endStream)
 streamOptions|=STREAM_OPTION_EMPTY_PAYLOAD;
 
-if(options.waitForTrailers)
+if(options.waitForTrailers){
 streamOptions|=STREAM_OPTION_GET_TRAILERS;
+state.flags|=STREAM_FLAGS_HAS_TRAILERS;
+}
 
 headers=processHeaders(headers);
 conststatusCode=headers[HTTP2_HEADER_STATUS]|=0;
@@ -2285,8 +2296,10 @@ class ServerHttp2Stream extends Http2Stream {
 }
 
 letstreamOptions=0;
-if(options.waitForTrailers)
+if(options.waitForTrailers){
 streamOptions|=STREAM_OPTION_GET_TRAILERS;
+this[kState].flags|=STREAM_FLAGS_HAS_TRAILERS;
+}
 
 if(typeoffd!=='number')
 thrownewERR_INVALID_ARG_TYPE('fd','number',fd);
@@ -2346,8 +2359,10 @@ class ServerHttp2Stream extends Http2Stream {
 }
 
 letstreamOptions=0;
-if(options.waitForTrailers)
+if(options.waitForTrailers){
 streamOptions|=STREAM_OPTION_GET_TRAILERS;
+this[kState].flags|=STREAM_FLAGS_HAS_TRAILERS;
+}
 
 constsession=this[kSession];
 debug(`Http2Stream ${this[kID]} [Http2Session `+