http2: force through RST_STREAM in destroy

If still needed, force through RST_STREAM in Http2Stream#destroy
calls, so that nghttp2 can wrap up properly and doesn't continue
trying to read & write data to the stream.

PR-URL: https://github.com/nodejs/node/pull/21016
Fixes: https://github.com/nodejs/node/issues/21008
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>

lib/internal/http2/core.js

@@ -337,7 +337,7 @@ function onStreamClose(code) {
         `${sessionName(stream[kSession][kType])}]: closed with code ${code}`);
 
   if (!stream.closed)
-    closeStream(stream, code, false);
+    closeStream(stream, code, kNoRstStream);
 
   stream[kState].fd = -1;
   // Defer destroy we actually emit end.
@@ -1476,7 +1476,11 @@ function finishSendTrailers(stream, headersList) {
     stream[kMaybeDestroy]();
 }
 
-function closeStream(stream, code, shouldSubmitRstStream = true) {
+const kNoRstStream = 0;
+const kSubmitRstStream = 1;
+const kForceRstStream = 2;
+
+function closeStream(stream, code, rstStreamStatus = kSubmitRstStream) {
   const state = stream[kState];
   state.flags |= STREAM_FLAGS_CLOSED;
   state.rstCode = code;
@@ -1499,9 +1503,10 @@ function closeStream(stream, code, shouldSubmitRstStream = true) {
     stream.end();
   }
 
-  if (shouldSubmitRstStream) {
+  if (rstStreamStatus !== kNoRstStream) {
     const finishFn = finishCloseStream.bind(stream, code);
-    if (!ending || finished || code !== NGHTTP2_NO_ERROR)
+    if (!ending || finished || code !== NGHTTP2_NO_ERROR ||
+        rstStreamStatus === kForceRstStream)
       finishFn();
     else
       stream.once('finish', finishFn);
@@ -1852,7 +1857,7 @@ class Http2Stream extends Duplex {
     const hasHandle = handle !== undefined;
 
     if (!this.closed)
-      closeStream(this, code, hasHandle);
+      closeStream(this, code, hasHandle ? kForceRstStream : kNoRstStream);
     this.push(null);
 
     if (hasHandle) {

src/node_http2.cc

@@ -1762,6 +1762,8 @@ void Http2Stream::Destroy() {
   // Do nothing if this stream instance is already destroyed
   if (IsDestroyed())
     return;
+  if (session_->HasPendingRstStream(id_))
+    FlushRstStream();
   flags_ |= NGHTTP2_STREAM_FLAG_DESTROYED;
 
   Debug(this, "destroying stream");

src/node_http2.h

@@ -9,6 +9,7 @@
 #include "stream_base-inl.h"
 #include "string_bytes.h"
 
+#include <algorithm>
 #include <queue>
 
 namespace node {
@@ -803,6 +804,12 @@ class Http2Session : public AsyncWrap, public StreamListener {
     pending_rst_streams_.emplace_back(stream_id);
   }
 
+  inline bool HasPendingRstStream(int32_t stream_id) {
+    return pending_rst_streams_.end() != std::find(pending_rst_streams_.begin(),
+                                                   pending_rst_streams_.end(),
+                                                   stream_id);
+  }
+
   // Handle reads/writes from the underlying network transport.
   void OnStreamRead(ssize_t nread, const uv_buf_t& buf) override;
   void OnStreamAfterWrite(WriteWrap* w, int status) override;

test/parallel/test-http2-large-write-destroy.js

@@ -0,0 +1,40 @@
+'use strict';
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+const fixtures = require('../common/fixtures');
+const http2 = require('http2');
+
+// This test will result in a crash due to a missed CHECK in C++ or
+// a straight-up segfault if the C++ doesn't send RST_STREAM through
+// properly when calling destroy.
+
+const content = Buffer.alloc(60000, 0x44);
+
+const server = http2.createSecureServer({
+  key: fixtures.readKey('agent1-key.pem'),
+  cert: fixtures.readKey('agent1-cert.pem')
+});
+server.on('stream', common.mustCall((stream) => {
+  stream.respond({
+    'Content-Type': 'application/octet-stream',
+    'Content-Length': (content.length.toString() * 2),
+    'Vary': 'Accept-Encoding'
+  }, { waitForTrailers: true });
+
+  stream.write(content);
+  stream.destroy();
+}));
+
+server.listen(0, common.mustCall(() => {
+  const client = http2.connect(`https://localhost:${server.address().port}`,
+                               { rejectUnauthorized: false });
+
+  const req = client.request({ ':path': '/' });
+  req.end();
+
+  req.on('close', common.mustCall(() => {
+    client.close();
+    server.close();
+  }));
+}));