http2: fix end without read

Adjust http2 behaviour to allow ending a stream even after some
data comes in (when the user has no intention of reading that
data). Also correctly end a stream when trailers are present.
PR-URL: #20621Fixes: #20060
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>

Author: apapirovski

lib/internal/http2/compat.js

@@ -260,8 +260,6 @@ class Http2ServerRequest extends Readable {
 stream[kRequest]=this;
 
 // Pause the stream..
-stream.pause();
-stream.on('data',onStreamData);
 stream.on('trailers',onStreamTrailers);
 stream.on('end',onStreamEnd);
 stream.on('error',onStreamError);
@@ -328,8 +326,12 @@ class Http2ServerRequest extends Readable {
 _read(nread){
 conststate=this[kState];
 if(!state.closed){
-state.didRead=true;
-process.nextTick(resumeStream,this[kStream]);
+if(!state.didRead){
+state.didRead=true;
+this[kStream].on('data',onStreamData);
+}else{
+process.nextTick(resumeStream,this[kStream]);
+}
 }else{
 this.emit('error',newERR_HTTP2_INVALID_STREAM());
 }

lib/internal/http2/core.js

@@ -349,11 +349,11 @@ function onStreamClose(code) {
 // Push a null so the stream can end whenever the client consumes
 // it completely.
 stream.push(null);
-
-// Same as net.
-if(stream.readableLength===0){
-stream.read(0);
-}
+// If the client hasn't tried to consume the stream and there is no
+// resume scheduled (which would indicate they would consume in the future),
+// then just dump the incoming data so that the stream can be destroyed.
+if(!stream[kState].didRead&&!stream._readableState.resumeScheduled)
+stream.resume();
 }
 }
 
@@ -1795,6 +1795,8 @@ class Http2Stream extends Duplex {
 constret=this[kHandle].trailers(headersList);
 if(ret<0)
 this.destroy(newNghttpError(ret));
+else
+this[kMaybeDestroy]();
 }
 
 getclosed(){

test/parallel/test-http2-client-upload-reject.js

@@ -20,12 +20,15 @@ fs.readFile(loc, common.mustCall((err, data) => {
 constserver=http2.createServer();
 
 server.on('stream',common.mustCall((stream)=>{
-stream.on('close',common.mustCall(()=>{
-assert.strictEqual(stream.rstCode,0);
-}));
-
-stream.respond({':status': 400});
-stream.end();
+// Wait for some data to come through.
+setImmediate(()=>{
+stream.on('close',common.mustCall(()=>{
+assert.strictEqual(stream.rstCode,0);
+}));
+
+stream.respond({':status': 400});
+stream.end();
+});
 }));
 
 server.listen(0,common.mustCall(()=>{

test/parallel/test-http2-compat-client-upload-reject.js

@@ -0,0 +1,44 @@
+'use strict';
+
+// Verifies that uploading data from a client works
+
+constcommon=require('../common');
+if(!common.hasCrypto)
+common.skip('missing crypto');
+constassert=require('assert');
+consthttp2=require('http2');
+constfs=require('fs');
+constfixtures=require('../common/fixtures');
+
+constloc=fixtures.path('person-large.jpg');
+
+assert(fs.existsSync(loc));
+
+fs.readFile(loc,common.mustCall((err,data)=>{
+assert.ifError(err);
+
+constserver=http2.createServer(common.mustCall((req,res)=>{
+setImmediate(()=>{
+res.writeHead(400);
+res.end();
+});
+}));
+
+server.listen(0,common.mustCall(()=>{
+constclient=http2.connect(`http://localhost:${server.address().port}`);
+
+constreq=client.request({':method': 'POST'});
+req.on('response',common.mustCall((headers)=>{
+assert.strictEqual(headers[':status'],400);
+}));
+
+req.resume();
+req.on('end',common.mustCall(()=>{
+server.close();
+client.close();
+}));
+
+conststr=fs.createReadStream(loc);
+str.pipe(req);
+}));
+}));