http2: fix stream reading resumption

_read should always resume the underlying code that is attempting
to push data to a readable stream. Adjust http2 core code to
resume its reading appropriately.

Fixes: nodejs#16578

Author: apapirovski

lib/internal/http2/core.js

@@ -1276,8 +1276,6 @@ function onStreamClosed(code) {
 }
 
 function streamOnResume() {
-  if (this._paused)
-    return this.pause();
   if (this[kID] === undefined) {
     this.once('ready', streamOnResume);
     return;
@@ -1301,8 +1299,7 @@ function streamOnPause() {
 
 function streamOnDrain() {
   const needPause = 0 > this._writableState.highWaterMark;
-  if (this._paused && !needPause) {
-    this._paused = false;
+  if (!needPause) {
     this.resume();
   }
 }

src/node_http2.cc

@@ -756,7 +756,7 @@ void Http2Session::FlushData(const FunctionCallbackInfo<Value>& args) {
   if (!(stream = session->FindStream(id))) {
     return args.GetReturnValue().Set(NGHTTP2_ERR_INVALID_STREAM_ID);
   }
-  stream->FlushDataChunks();
+  stream->ReadResume();
 }
 
 void Http2Session::UpdateChunksSent(const FunctionCallbackInfo<Value>& args) {

src/node_http2_core-inl.h

@@ -896,6 +896,14 @@ inline void Nghttp2Stream::ReadStart() {
   FlushDataChunks();
 }
 
+inline void Nghttp2Stream::ReadResume() {
+  DEBUG_HTTP2("Nghttp2Stream %d: resume reading\n", id_);
+  flags_ &= ~NGHTTP2_STREAM_FLAG_READ_PAUSED;
+
+  // Flush any queued data chunks immediately out to the JS layer
+  FlushDataChunks();
+}
+
 inline void Nghttp2Stream::ReadStop() {
   DEBUG_HTTP2("Nghttp2Stream %d: stop reading\n", id_);
   if (!IsReading())

src/node_http2_core.h

@@ -384,6 +384,9 @@ class Nghttp2Stream {
   // the session to be emitted at the JS side
   inline void ReadStart();
 
+  // Resume Reading
+  inline void ReadResume();
+
   // Stop/Pause Reading.
   inline void ReadStop();
 

test/parallel/test-http2-pipe.js

@@ -0,0 +1,49 @@
+'use strict';
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+const fixtures = require('../common/fixtures');
+const assert = require('assert');
+const http2 = require('http2');
+const fs = require('fs');
+const path = require('path');
+
+// piping should work as expected with createWriteStream
+
+const loc = fixtures.path('url-tests.js');
+const fn = path.join(common.tmpDir, 'http2-url-tests.js');
+common.refreshTmpDir();
+
+const server = http2.createServer();
+
+server.on('stream', common.mustCall((stream) => {
+  const dest = stream.pipe(fs.createWriteStream(fn));
+  dest.on('finish', common.mustCall(() => {
+    assert.deepStrictEqual(fs.readFileSync(loc), fs.readFileSync(fn));
+    fs.unlinkSync(fn);
+    stream.respond();
+    stream.end();
+  }));
+}));
+
+server.listen(0, common.mustCall(() => {
+  const port = server.address().port;
+  const client = http2.connect(`http://localhost:${port}`);
+
+  let remaining = 2;
+  function maybeClose() {
+    if (--remaining === 0) {
+      server.close();
+      client.destroy();
+    }
+  }
+
+  const req = client.request({ ':method': 'POST' });
+  req.on('response', common.mustCall());
+  req.resume();
+  req.on('end', common.mustCall(maybeClose));
+  const str = fs.createReadStream(loc);
+  str.on('end', common.mustCall(maybeClose));
+  str.pipe(req);
+}));