fix response before request handling

source/common/http/conn_manager_impl.cc

@@ -92,6 +92,9 @@ void ConnectionManagerImpl::destroyStream(ActiveStream& stream) {
   // deleted first.
   bool reset_stream = false;
   if (!stream.state_.remote_complete_ || !stream.state_.local_complete_) {
+    // Indicate local is complete at this point so that if we reset during a continuation, we don't
+    // raise further data or trailers.
+    stream.state_.local_complete_ = true;
     stream.response_encoder_->getStream().resetStream(StreamResetReason::LocalReset);
     reset_stream = true;
   }
@@ -416,6 +419,12 @@ void ConnectionManagerImpl::ActiveStream::decodeData(const Buffer::Instance& dat
 
 void ConnectionManagerImpl::ActiveStream::decodeData(ActiveStreamDecoderFilter* filter,
                                                      Buffer::Instance& data, bool end_stream) {
+  // If a response is complete or a reset has been sent, filters do not care about further body
+  // data. Just drop it.
+  if (state_.local_complete_) {
+    return;
+  }
+
   std::list<ActiveStreamDecoderFilterPtr>::iterator entry;
   if (!filter) {
     entry = decoder_filters_.begin();
@@ -442,6 +451,11 @@ void ConnectionManagerImpl::ActiveStream::decodeTrailers(HeaderMapPtr&& trailers
 
 void ConnectionManagerImpl::ActiveStream::decodeTrailers(ActiveStreamDecoderFilter* filter,
                                                          HeaderMap& trailers) {
+  // See decodeData() above for why we check local_complete_ here.
+  if (state_.local_complete_) {
+    return;
+  }
+
   std::list<ActiveStreamDecoderFilterPtr>::iterator entry;
   if (!filter) {
     entry = decoder_filters_.begin();
@@ -478,11 +492,6 @@ ConnectionManagerImpl::ActiveStream::commonEncodePrefix(ActiveStreamEncoderFilte
 
 void ConnectionManagerImpl::ActiveStream::encodeHeaders(ActiveStreamEncoderFilter* filter,
                                                         Http::HeaderMap& headers, bool end_stream) {
-  if (filter == nullptr) {
-    ASSERT(!state_.local_started_);
-    state_.local_started_ = true;
-  }
-
   std::list<ActiveStreamEncoderFilterPtr>::iterator entry = commonEncodePrefix(filter, end_stream);
   for (; entry != encoder_filters_.end(); entry++) {
     Http::FilterHeadersStatus status = (*entry)->handle_->encodeHeaders(headers, end_stream);

source/common/http/conn_manager_impl.h

@@ -370,7 +370,6 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     struct State {
       bool remote_complete_{};
       bool local_complete_{};
-      bool local_started_{};
     };
 
     ConnectionManagerImpl& connection_manager_;

source/common/router/router.cc

@@ -547,7 +547,8 @@ void Filter::UpstreamRequest::onPerTryTimeout() {
       ->stats()
       .upstream_rq_per_try_timeout_.inc();
   upstream_encoder_->resetStream();
-  parent_.onUpstreamReset(UpstreamResetType::PerTryTimeout, Optional<Http::StreamResetReason>());
+  parent_.onUpstreamReset(UpstreamResetType::PerTryTimeout,
+                          Optional<Http::StreamResetReason>(Http::StreamResetReason::LocalReset));
 }
 
 RetryStatePtr

test/common/http/conn_manager_impl_test.cc

@@ -461,7 +461,8 @@ TEST_F(HttpConnectionManagerImplTest, IntermediateBufferingEarlyResponse) {
         return Http::FilterHeadersStatus::StopIteration;
       }));
 
-  EXPECT_CALL(*decoder_filter2, decodeData(_, true));
+  // Response is already complete so we drop buffered body data when we continue.
+  EXPECT_CALL(*decoder_filter2, decodeData(_, _)).Times(0);
   decoder_filter1->callbacks_->continueDecoding();
 }
 

test/integration/http2_integration_test.cc

@@ -9,6 +9,16 @@
 
 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);
+}
+
+TEST_F(Http2IntegrationTest, RouterNotFoundBodyBuffer) {
+  testRouterNotFoundWithBody(HTTP_BUFFER_PORT, Http::CodecClient::Type::HTTP2);
+}
+
 TEST_F(Http2IntegrationTest, RouterRedirect) { testRouterRedirect(Http::CodecClient::Type::HTTP2); }
 
 TEST_F(Http2IntegrationTest, DrainClose) { testDrainClose(Http::CodecClient::Type::HTTP2); }

test/integration/integration.h

@@ -172,6 +172,7 @@ class BaseIntegrationTest : Logger::Loggable<Logger::Id::testing> {
 protected:
   void testRouterRedirect(Http::CodecClient::Type type);
   void testRouterNotFound(Http::CodecClient::Type type);
+  void testRouterNotFoundWithBody(uint32_t port, Http::CodecClient::Type type);
   void testRouterRequestAndResponseWithBody(Network::ClientConnectionPtr&& conn,
                                             Http::CodecClient::Type type, uint64_t request_size,
                                             uint64_t response_size);

test/integration/integration_admin_test.cc

@@ -5,72 +5,72 @@
 
 TEST_F(IntegrationTest, HealthCheck) {
   BufferingStreamDecoderPtr response = IntegrationUtil::makeSingleRequest(
-      HTTP_PORT, "GET", "/healthcheck", Http::CodecClient::Type::HTTP1);
+      HTTP_PORT, "GET", "/healthcheck", "", Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
 
-  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/healthcheck/fail",
+  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/healthcheck/fail", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
 
-  response = IntegrationUtil::makeSingleRequest(HTTP_PORT, "GET", "/healthcheck",
+  response = IntegrationUtil::makeSingleRequest(HTTP_PORT, "GET", "/healthcheck", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("503", response->headers().get(":status"));
 
-  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/healthcheck/ok",
+  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/healthcheck/ok", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
 
-  response = IntegrationUtil::makeSingleRequest(HTTP_PORT, "GET", "/healthcheck",
+  response = IntegrationUtil::makeSingleRequest(HTTP_PORT, "GET", "/healthcheck", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
 
-  response = IntegrationUtil::makeSingleRequest(HTTP_BUFFER_PORT, "GET", "/healthcheck",
+  response = IntegrationUtil::makeSingleRequest(HTTP_BUFFER_PORT, "GET", "/healthcheck", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
 }
 
 TEST_F(IntegrationTest, AdminLogging) {
   BufferingStreamDecoderPtr response = IntegrationUtil::makeSingleRequest(
-      ADMIN_PORT, "GET", "/logging", Http::CodecClient::Type::HTTP1);
+      ADMIN_PORT, "GET", "/logging", "", Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("404", response->headers().get(":status"));
 
   // Bad level
-  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/logging?level=blah",
+  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/logging?level=blah", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("404", response->headers().get(":status"));
 
   // Bad logger
-  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/logging?blah=info",
+  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/logging?blah=info", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("404", response->headers().get(":status"));
 
   // This is going to stomp over custom log levels that are set on the command line.
-  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/logging?level=warning",
+  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/logging?level=warning", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
   for (const Logger::Logger& logger : Logger::Registry::loggers()) {
     EXPECT_EQ("warning", logger.levelString());
   }
 
-  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/logging?assert=trace",
+  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/logging?assert=trace", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
   EXPECT_EQ(spdlog::level::trace, Logger::Registry::getLog(Logger::Id::assert).level());
 
   const char* level_name = spdlog::level::level_names[default_log_level_];
   response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET",
-                                                fmt::format("/logging?level={}", level_name),
+                                                fmt::format("/logging?level={}", level_name), "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
@@ -79,53 +79,60 @@ TEST_F(IntegrationTest, AdminLogging) {
   }
 }
 
+TEST_F(IntegrationTest, AdminCertEndpoint) {
+  BufferingStreamDecoderPtr response = IntegrationUtil::makeSingleRequest(
+      ADMIN_PORT, "GET", "/certs", "", Http::CodecClient::Type::HTTP1);
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().get(":status"));
+}
+
 TEST_F(IntegrationTest, Admin) {
-  BufferingStreamDecoderPtr response =
-      IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/", Http::CodecClient::Type::HTTP1);
+  BufferingStreamDecoderPtr response = IntegrationUtil::makeSingleRequest(
+      ADMIN_PORT, "GET", "/", "", Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("404", response->headers().get(":status"));
 
-  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/server_info",
+  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/server_info", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
 
-  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/stats",
+  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/stats", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
 
-  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/clusters",
+  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/clusters", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
 
-  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/cpuprofiler",
+  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/cpuprofiler", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("400", response->headers().get(":status"));
 
-  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/cpuprofiler?enable=y",
+  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/cpuprofiler?enable=y", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
 
-  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/cpuprofiler?enable=n",
+  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/cpuprofiler?enable=n", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
 
-  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/hot_restart_version",
+  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/hot_restart_version", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
 
-  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/reset_counters",
+  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/reset_counters", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
 
-  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/certs",
+  response = IntegrationUtil::makeSingleRequest(ADMIN_PORT, "GET", "/certs", "",
                                                 Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));

test/integration/integration_test.cc

@@ -23,7 +23,22 @@ TEST_F(IntegrationTest, RouterNotFound) { testRouterNotFound(Http::CodecClient::
 
 void BaseIntegrationTest::testRouterNotFound(Http::CodecClient::Type type) {
   BufferingStreamDecoderPtr response =
-      IntegrationUtil::makeSingleRequest(HTTP_PORT, "GET", "/notfound", type);
+      IntegrationUtil::makeSingleRequest(HTTP_PORT, "GET", "/notfound", "", type);
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("404", response->headers().get(":status"));
+}
+
+TEST_F(IntegrationTest, RouterNotFoundBodyNoBuffer) {
+  testRouterNotFoundWithBody(HTTP_PORT, Http::CodecClient::Type::HTTP1);
+}
+
+TEST_F(IntegrationTest, RouterNotFoundBodyBuffer) {
+  testRouterNotFoundWithBody(HTTP_BUFFER_PORT, Http::CodecClient::Type::HTTP1);
+}
+
+void BaseIntegrationTest::testRouterNotFoundWithBody(uint32_t port, Http::CodecClient::Type type) {
+  BufferingStreamDecoderPtr response =
+      IntegrationUtil::makeSingleRequest(port, "POST", "/notfound", "foo", type);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("404", response->headers().get(":status"));
 }
@@ -32,7 +47,7 @@ TEST_F(IntegrationTest, RouterRedirect) { testRouterRedirect(Http::CodecClient::
 
 void BaseIntegrationTest::testRouterRedirect(Http::CodecClient::Type type) {
   BufferingStreamDecoderPtr response =
-      IntegrationUtil::makeSingleRequest(HTTP_PORT, "GET", "/foo", type, "www.redirect.com");
+      IntegrationUtil::makeSingleRequest(HTTP_PORT, "GET", "/foo", "", type, "www.redirect.com");
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("301", response->headers().get(":status"));
   EXPECT_EQ("https://www.redirect.com/foo", response->headers().get("location"));

test/integration/server.cc

@@ -40,7 +40,7 @@ IntegrationTestServer::~IntegrationTestServer() {
   log().info("stopping integration test server");
 
   BufferingStreamDecoderPtr response = IntegrationUtil::makeSingleRequest(
-      IntegrationTest::ADMIN_PORT, "GET", "/quitquitquit", Http::CodecClient::Type::HTTP1);
+      IntegrationTest::ADMIN_PORT, "GET", "/quitquitquit", "", Http::CodecClient::Type::HTTP1);
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().get(":status"));
 

test/integration/ssl_integration_test.cc

@@ -109,11 +109,4 @@ TEST_F(SslIntegrationTest, AltAlpn) {
   checkStats();
 }
 
-TEST_F(SslIntegrationTest, AdminCertEndpoint) {
-  BufferingStreamDecoderPtr response = IntegrationUtil::makeSingleRequest(
-      ADMIN_PORT, "GET", "/certs", Http::CodecClient::Type::HTTP1);
-  EXPECT_TRUE(response->complete());
-  EXPECT_EQ("200", response->headers().get(":status"));
-}
-
 } // Ssl

test/integration/utility.cc

@@ -4,6 +4,7 @@
 #include "envoy/network/connection.h"
 
 #include "common/api/api_impl.h"
+#include "common/buffer/buffer_impl.h"
 #include "common/common/assert.h"
 #include "common/http/header_map_impl.h"
 #include "common/http/headers.h"
@@ -38,10 +39,10 @@ void BufferingStreamDecoder::onComplete() {
 
 void BufferingStreamDecoder::onResetStream(Http::StreamResetReason) { ADD_FAILURE(); }
 
-BufferingStreamDecoderPtr IntegrationUtil::makeSingleRequest(uint32_t port, std::string method,
-                                                             std::string url,
-                                                             Http::CodecClient::Type type,
-                                                             std::string host) {
+BufferingStreamDecoderPtr
+IntegrationUtil::makeSingleRequest(uint32_t port, const std::string& method, const std::string& url,
+                                   const std::string& body, Http::CodecClient::Type type,
+                                   const std::string& host) {
   Api::Impl api(std::chrono::milliseconds(9000));
   Event::DispatcherPtr dispatcher(api.allocateDispatcher());
   Stats::IsolatedStoreImpl stats_store;
@@ -54,11 +55,14 @@ BufferingStreamDecoderPtr IntegrationUtil::makeSingleRequest(uint32_t port, std:
   encoder.getStream().addCallbacks(*response);
 
   Http::HeaderMapImpl headers;
-  headers.addViaMoveValue(Http::Headers::get().Method, std::move(method));
-  headers.addViaMoveValue(Http::Headers::get().Path, std::move(url));
-  headers.addViaMoveValue(Http::Headers::get().Host, std::move(host));
+  headers.addViaCopy(Http::Headers::get().Method, method);
+  headers.addViaCopy(Http::Headers::get().Path, url);
+  headers.addViaCopy(Http::Headers::get().Host, host);
   headers.addViaMoveValue(Http::Headers::get().Scheme, "http");
-  encoder.encodeHeaders(headers, true);
+  encoder.encodeHeaders(headers, body.empty());
+  if (!body.empty()) {
+    encoder.encodeData(Buffer::OwnedImpl(body), true);
+  }
 
   dispatcher->run(Event::Dispatcher::RunType::Block);
   return response;

test/integration/utility.h

@@ -79,12 +79,15 @@ class IntegrationUtil {
    * @param port supplies the port to connect to on localhost.
    * @param method supplies the request method.
    * @param url supplies the request url.
+   * @param body supplies the optional request body to send.
    * @param type supplies the codec to use for the request.
    * @param host supplies the host header to use for the request.
    * @return BufferingStreamDecoderPtr the complete request or a partial request if there was
    *         remote easly disconnection.
    */
-  static BufferingStreamDecoderPtr makeSingleRequest(uint32_t port, std::string method,
-                                                     std::string url, Http::CodecClient::Type type,
-                                                     std::string host = "host");
+  static BufferingStreamDecoderPtr makeSingleRequest(uint32_t port, const std::string& method,
+                                                     const std::string& url,
+                                                     const std::string& body,
+                                                     Http::CodecClient::Type type,
+                                                     const std::string& host = "host");
 };