http2: fix abort when client.destroy inside end event

Backport-PR-URL: #14813
Backport-Reviewed-By: Anna Henningsen <[email protected]>
Backport-Reviewed-By: Timothy Gu <[email protected]>

PR-URL: #14239
Reviewed-By: Anna Henningsen <[email protected]>
Reviewed-By: Colin Ihrig <[email protected]>
Reviewed-By: Matteo Collina <[email protected]>

Author: jasnell

lib/internal/http2/core.js

@@ -289,11 +289,9 @@ function onSessionRead(nread, buf, handle) {
   _unrefActive(this); // Reset the session timeout timer
   _unrefActive(stream); // Reset the stream timeout timer
 
-  if (nread >= 0) {
+  if (nread >= 0 && !stream.destroyed) {
     if (!stream.push(buf)) {
-      assert(this.streamReadStop(id) === undefined,
-             `HTTP/2 Stream ${id} does not exist. Please report this as ' +
-             'a bug in Node.js`);
+      this.streamReadStop(id);
       state.reading = false;
     }
   } else {
@@ -1475,44 +1473,48 @@ class Http2Stream extends Duplex {
       this.once('ready', this._destroy.bind(this, err, callback));
       return;
     }
-    debug(`[${sessionName(session[kType])}] destroying stream ${this[kID]}`);
-
-    // Submit RST-STREAM frame if one hasn't been sent already and the
-    // stream hasn't closed normally...
-    if (!this[kState].rst) {
-      const code =
-        err instanceof Error ?
-          NGHTTP2_INTERNAL_ERROR : NGHTTP2_NO_ERROR;
-      this[kSession].rstStream(this, code);
-    }
-
+    process.nextTick(() => {
+      debug(`[${sessionName(session[kType])}] destroying stream ${this[kID]}`);
+
+      // Submit RST-STREAM frame if one hasn't been sent already and the
+      // stream hasn't closed normally...
+      if (!this[kState].rst && !session.destroyed) {
+        const code =
+          err instanceof Error ?
+            NGHTTP2_INTERNAL_ERROR : NGHTTP2_NO_ERROR;
+        this[kSession].rstStream(this, code);
+      }
 
-    // Remove the close handler on the session
-    session.removeListener('close', this[kState].closeHandler);
+      // Remove the close handler on the session
+      session.removeListener('close', this[kState].closeHandler);
 
-    // Unenroll the timer
-    unenroll(this);
+      // Unenroll the timer
+      unenroll(this);
 
-    setImmediate(finishStreamDestroy.bind(this, handle));
-    session[kState].streams.delete(this[kID]);
-    delete this[kSession];
+      setImmediate(finishStreamDestroy.bind(this, handle));
 
-    // All done
-    const rst = this[kState].rst;
-    const code = rst ? this[kState].rstCode : NGHTTP2_NO_ERROR;
-    if (code !== NGHTTP2_NO_ERROR) {
-      const err = new errors.Error('ERR_HTTP2_STREAM_ERROR', code);
-      process.nextTick(() => this.emit('error', err));
-    }
-    process.nextTick(emit.bind(this, 'streamClosed', code));
-    debug(`[${sessionName(session[kType])}] stream ${this[kID]} destroyed`);
-    callback(err);
+      // All done
+      const rst = this[kState].rst;
+      const code = rst ? this[kState].rstCode : NGHTTP2_NO_ERROR;
+      if (code !== NGHTTP2_NO_ERROR) {
+        const err = new errors.Error('ERR_HTTP2_STREAM_ERROR', code);
+        process.nextTick(() => this.emit('error', err));
+      }
+      process.nextTick(emit.bind(this, 'streamClosed', code));
+      debug(`[${sessionName(session[kType])}] stream ${this[kID]} destroyed`);
+      callback(err);
+    });
   }
 }
 
 function finishStreamDestroy(handle) {
+  const id = this[kID];
+  const session = this[kSession];
+  session[kState].streams.delete(id);
+  delete this[kSession];
   if (handle !== undefined)
-    handle.destroyStream(this[kID]);
+    handle.destroyStream(id);
+  this.emit('destroy');
 }
 
 function processHeaders(headers) {

src/node_http2.cc

@@ -329,7 +329,6 @@ class Http2Session : public AsyncWrap,
     padding_strategy_ = opts.GetPaddingStrategy();
 
     Init(env->event_loop(), type, *opts);
-    stream_buf_.AllocateSufficientStorage(kAllocBufferSize);
   }
 
   ~Http2Session() override {
@@ -456,15 +455,16 @@ class Http2Session : public AsyncWrap,
   }
 
   char* stream_alloc() {
-    return *stream_buf_;
+    return stream_buf_;
   }
 
  private:
   StreamBase* stream_;
   StreamResource::Callback<StreamResource::AllocCb> prev_alloc_cb_;
   StreamResource::Callback<StreamResource::ReadCb> prev_read_cb_;
   padding_strategy_type padding_strategy_ = PADDING_STRATEGY_NONE;
-  MaybeStackBuffer<char, kAllocBufferSize> stream_buf_;
+
+  char stream_buf_[kAllocBufferSize];
 };
 
 class ExternalHeader :

src/node_http2.h

@@ -221,10 +221,7 @@ inline int Nghttp2Session::Free() {
     Nghttp2Session* session =
         ContainerOf(&Nghttp2Session::prep_,
                     reinterpret_cast<uv_prepare_t*>(handle));
-
     session->OnFreeSession();
-    DEBUG_HTTP2("Nghttp2Session %d: session is free\n",
-            session->session_type_);
   };
   uv_close(reinterpret_cast<uv_handle_t*>(&prep_), PrepClose);
 
@@ -302,9 +299,9 @@ inline void Nghttp2Stream::ResetState(
 inline void Nghttp2Stream::Destroy() {
   DEBUG_HTTP2("Nghttp2Stream %d: destroying stream\n", id_);
   // Do nothing if this stream instance is already destroyed
-  if (IsDestroyed() || IsDestroying())
+  if (IsDestroyed())
     return;
-  flags_ |= NGHTTP2_STREAM_DESTROYING;
+  flags_ |= NGHTTP2_STREAM_DESTROYED;
   Nghttp2Session* session = this->session_;
 
   if (session != nullptr) {

src/node_http2_core-inl.h

@@ -65,9 +65,7 @@ enum nghttp2_stream_flags {
   // Stream is closed
   NGHTTP2_STREAM_CLOSED = 0x8,
   // Stream is destroyed
-  NGHTTP2_STREAM_DESTROYED = 0x10,
-  // Stream is being destroyed
-  NGHTTP2_STREAM_DESTROYING = 0x20
+  NGHTTP2_STREAM_DESTROYED = 0x10
 };
 
 
@@ -290,10 +288,6 @@ class Nghttp2Stream {
     return (flags_ & NGHTTP2_STREAM_DESTROYED) == NGHTTP2_STREAM_DESTROYED;
   }
 
-  inline bool IsDestroying() const {
-    return (flags_ & NGHTTP2_STREAM_DESTROYING) == NGHTTP2_STREAM_DESTROYING;
-  }
-
   // Queue outbound chunks of data to be sent on this stream
   inline int Write(
       nghttp2_stream_write_t* req,

src/node_http2_core.h

@@ -51,6 +51,14 @@ server.on('listening', common.mustCall(() => {
   const client = h2.connect(`http://localhost:${server.address().port}`,
                             options);
 
+  let remaining = 2;
+  function maybeClose() {
+    if (--remaining === 0) {
+      server.close();
+      client.destroy();
+    }
+  }
+
   const req = client.request({ ':path': '/' });
 
   // Because maxReservedRemoteStream is 1, the stream event
@@ -59,15 +67,12 @@ server.on('listening', common.mustCall(() => {
   client.on('stream', common.mustCall((stream) => {
     stream.resume();
     stream.on('end', common.mustCall());
+    stream.on('streamClosed', common.mustCall(maybeClose));
   }));
 
   req.on('response', common.mustCall());
 
   req.resume();
-  req.on('end', common.mustCall(() => {
-    server.close();
-    client.destroy();
-  }));
-  req.end();
-
+  req.on('end', common.mustCall());
+  req.on('streamClosed', common.mustCall(maybeClose));
 }));

test/parallel/test-http2-options-max-reserved-streams.js

@@ -65,7 +65,8 @@ server.listen(0, common.mustCall(() => {
       assert.strictEqual(headers.location, undefined);
     }));
     req.resume();
-    req.on('end', common.mustCall(maybeClose));
+    req.on('end', common.mustCall());
+    req.on('streamClosed', common.mustCall(maybeClose));
   }
 
   doTest(str, 'location', str);

test/parallel/test-http2-response-splitting.js

@@ -36,7 +36,9 @@ function onStream(stream) {
 
   assert.notStrictEqual(stream.session, undefined);
   socket.destroy();
-  assert.strictEqual(stream.session, undefined);
+  stream.on('destroy', common.mustCall(() => {
+    assert.strictEqual(stream.session, undefined);
+  }));
 }
 
 server.listen(0);
@@ -49,9 +51,8 @@ server.on('listening', common.mustCall(() => {
     [HTTP2_HEADER_METHOD]: HTTP2_METHOD_POST });
 
   req.on('aborted', common.mustCall());
+  req.resume();
   req.on('end', common.mustCall());
-  req.on('response', common.mustCall());
-  req.on('data', common.mustCall());
 
   client.on('close', common.mustCall());
 }));