http/2: correctly handle reset after complete response

source/common/http/http2/codec_impl.cc

@@ -194,14 +194,25 @@ void ConnectionImpl::StreamImpl::encodeData(const Buffer::Instance& data, bool e
 void ConnectionImpl::StreamImpl::resetStream(StreamResetReason reason) {
   // Higher layers expect calling resetStream() to immediately raise reset callbacks.
   runResetCallbacks(reason);
+
+  // If we submit a reset, nghttp2 will cancel outbound frames that have not yet been sent.
+  // We want these frames to go out so we defer the reset until we send all of the frames that
+  // end the local stream.
+  if (local_end_stream_ && !local_end_stream_sent_) {
+    deferred_reset_.value(reason);
+  } else {
+    resetStreamWorker(reason);
+    parent_.sendPendingFrames();
+  }
+}
+
+void ConnectionImpl::StreamImpl::resetStreamWorker(StreamResetReason reason) {
   int rc = nghttp2_submit_rst_stream(parent_.session_, NGHTTP2_FLAG_NONE, stream_id_,
                                      reason == StreamResetReason::LocalRefusedStreamReset
                                          ? NGHTTP2_REFUSED_STREAM
                                          : NGHTTP2_NO_ERROR);
   ASSERT(rc == 0);
   UNREFERENCED_PARAMETER(rc);
-
-  parent_.sendPendingFrames();
 }
 
 void ConnectionImpl::StreamImpl::runResetCallbacks(StreamResetReason reason) {
@@ -295,15 +306,26 @@ int ConnectionImpl::onFrameReceived(const nghttp2_frame* frame) {
     } else {
       ASSERT(frame->headers.cat == NGHTTP2_HCAT_HEADERS);
       ASSERT(stream->remote_end_stream_);
-      stream->decoder_->decodeTrailers(std::move(stream->headers_));
+
+      // It's possible that we are waiting to send a deferred reset, so only raise trailers if local
+      // is not complete.
+      if (!stream->deferred_reset_.valid()) {
+        stream->decoder_->decodeTrailers(std::move(stream->headers_));
+      }
     }
 
     stream->headers_.reset();
     break;
   }
   case NGHTTP2_DATA: {
     stream->remote_end_stream_ = frame->hd.flags & NGHTTP2_FLAG_END_STREAM;
-    stream->decoder_->decodeData(stream->pending_recv_data_, stream->remote_end_stream_);
+
+    // It's possible that we are waiting to send a deferred reset, so only raise data if local
+    // is not complete.
+    if (!stream->deferred_reset_.valid()) {
+      stream->decoder_->decodeData(stream->pending_recv_data_, stream->remote_end_stream_);
+    }
+
     stream->pending_recv_data_.drain(stream->pending_recv_data_.length());
     break;
   }
@@ -323,13 +345,29 @@ int ConnectionImpl::onFrameSend(const nghttp2_frame* frame) {
   // In all cases however it will attempt to send a GOAWAY frame with an error status. If we see
   // an outgoing frame of this type, we will return an error code so that we can abort execution.
   conn_log_trace("sent frame type={}", connection_, static_cast<uint64_t>(frame->hd.type));
-  if (frame->hd.type == NGHTTP2_GOAWAY && frame->goaway.error_code != NGHTTP2_NO_ERROR) {
-    return NGHTTP2_ERR_CALLBACK_FAILURE;
+  switch (frame->hd.type) {
+  case NGHTTP2_GOAWAY: {
+    if (frame->goaway.error_code != NGHTTP2_NO_ERROR) {
+      return NGHTTP2_ERR_CALLBACK_FAILURE;
+    }
+    break;
   }
 
-  if (frame->hd.type == NGHTTP2_RST_STREAM) {
+  case NGHTTP2_RST_STREAM: {
     conn_log_debug("sent reset code={}", connection_, frame->rst_stream.error_code);
     stats_.tx_reset_.inc();
+    break;
+  }
+
+  case NGHTTP2_HEADERS:
+  case NGHTTP2_DATA: {
+    StreamImpl* stream = getStream(frame->hd.stream_id);
+    stream->local_end_stream_sent_ = frame->hd.flags & NGHTTP2_FLAG_END_STREAM;
+    if (stream->local_end_stream_sent_ && stream->deferred_reset_.valid()) {
+      stream->resetStreamWorker(stream->deferred_reset_.value());
+    }
+    break;
+  }
   }
 
   return 0;

source/common/http/http2/codec_impl.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include "envoy/common/optional.h"
 #include "envoy/event/deferred_deletable.h"
 #include "envoy/http/codec.h"
 #include "envoy/network/connection.h"
@@ -100,6 +101,7 @@ class ConnectionImpl : public virtual Connection, Logger::Loggable<Logger::Id::h
     StreamImpl* base() { return this; }
     ssize_t onDataSourceRead(size_t length, uint32_t* data_flags);
     int onDataSourceSend(const uint8_t* framehd, size_t length);
+    void resetStreamWorker(StreamResetReason reason);
     void runResetCallbacks(StreamResetReason reason);
     void buildHeaders(std::vector<nghttp2_nv>& final_headers, const HeaderMap& headers);
     virtual void submitHeaders(const std::vector<nghttp2_nv>& final_headers,
@@ -130,12 +132,14 @@ class ConnectionImpl : public virtual Connection, Logger::Loggable<Logger::Id::h
     StreamDecoder* decoder_{};
     int32_t stream_id_{-1};
     bool local_end_stream_{};
+    bool local_end_stream_sent_{};
     bool remote_end_stream_{};
     Buffer::OwnedImpl pending_recv_data_;
     Buffer::OwnedImpl pending_send_data_;
     bool data_deferred_{};
     bool reset_callbacks_run_{};
     HeaderMapPtr pending_trailers_;
+    Optional<StreamResetReason> deferred_reset_;
   };
 
   typedef std::unique_ptr<StreamImpl> StreamImplPtr;

test/integration/http2_integration_test.cc

@@ -10,9 +10,7 @@
 TEST_F(Http2IntegrationTest, RouterNotFound) { testRouterNotFound(Http::CodecClient::Type::HTTP2); }
 
 TEST_F(Http2IntegrationTest, RouterNotFoundBodyNoBuffer) {
-  // TODO: Right now this test does not work IMO because of an issue in nghttp2.
-  // https://github.com/nghttp2/nghttp2/issues/692
-  // testRouterNotFoundWithBody(HTTP_PORT, Http::CodecClient::Type::HTTP2);
+  testRouterNotFoundWithBody(HTTP_PORT, Http::CodecClient::Type::HTTP2);
 }
 
 TEST_F(Http2IntegrationTest, RouterNotFoundBodyBuffer) {