http: rolling back the dispatcher cleanup

source/common/http/conn_manager_impl.cc

@@ -296,40 +296,52 @@ Network::FilterStatus ConnectionManagerImpl::onData(Buffer::Instance& data, bool
     }
   }
 
-  try {
-    codec_->dispatch(data);
-  } catch (const FrameFloodException& e) {
-    // TODO(mattklein123): This is an emergency substitute for the lack of connection level
-    // logging in the HCM. In a public follow up change we will add full support for connection
-    // level logging in the HCM, similar to what we have in tcp_proxy. This will allow abuse
-    // indicators to be stored in the connection level stream info, and then matched, sampled,
-    // etc. when logged.
-    const envoy::type::FractionalPercent default_value; // 0
-    if (runtime_.snapshot().featureEnabled("http.connection_manager.log_flood_exception",
-                                           default_value)) {
-      ENVOY_CONN_LOG(warn, "downstream HTTP flood from IP '{}': {}", read_callbacks_->connection(),
-                     read_callbacks_->connection().remoteAddress()->asString(), e.what());
-    }
+  bool redispatch;
+  do {
+    redispatch = false;
+
+    try {
+      codec_->dispatch(data);
+    } catch (const FrameFloodException& e) {
+      // TODO(mattklein123): This is an emergency substitute for the lack of connection level
+      // logging in the HCM. In a public follow up change we will add full support for connection
+      // level logging in the HCM, similar to what we have in tcp_proxy. This will allow abuse
+      // indicators to be stored in the connection level stream info, and then matched, sampled,
+      // etc. when logged.
+      const envoy::type::FractionalPercent default_value; // 0
+      if (runtime_.snapshot().featureEnabled("http.connection_manager.log_flood_exception",
+                                             default_value)) {
+        ENVOY_CONN_LOG(warn, "downstream HTTP flood from IP '{}': {}",
+                       read_callbacks_->connection(),
+                       read_callbacks_->connection().remoteAddress()->asString(), e.what());
+      }
 
-    handleCodecException(e.what());
-    return Network::FilterStatus::StopIteration;
-  } catch (const CodecProtocolException& e) {
-    stats_.named_.downstream_cx_protocol_error_.inc();
-    handleCodecException(e.what());
-    return Network::FilterStatus::StopIteration;
-  }
+      handleCodecException(e.what());
+      return Network::FilterStatus::StopIteration;
+    } catch (const CodecProtocolException& e) {
+      stats_.named_.downstream_cx_protocol_error_.inc();
+      handleCodecException(e.what());
+      return Network::FilterStatus::StopIteration;
+    }
 
-  // Processing incoming data may release outbound data so check for closure here as well.
-  checkForDeferredClose();
+    // Processing incoming data may release outbound data so check for closure here as well.
+    checkForDeferredClose();
+
+    // The HTTP/1 codec will pause dispatch after a single message is complete. We want to
+    // either redispatch if there are no streams and we have more data. If we have a single
+    // complete non-WebSocket stream but have not responded yet we will pause socket reads
+    // to apply back pressure.
+    if (codec_->protocol() < Protocol::Http2) {
+      if (read_callbacks_->connection().state() == Network::Connection::State::Open &&
+          data.length() > 0 && streams_.empty()) {
+        redispatch = true;
+      }
 
-  // The HTTP/1 codec will pause parsing after a single message is complete. If we have a single
-  // complete non-WebSocket stream but have not responded yet we will pause socket reads
-  // to apply back pressure.
-  if (codec_->protocol() < Protocol::Http2) {
-    if (!streams_.empty() && streams_.front()->state_.remote_complete_) {
-      read_callbacks_->connection().readDisable(true);
+      if (!streams_.empty() && streams_.front()->state_.remote_complete_) {
+        read_callbacks_->connection().readDisable(true);
+      }
     }
-  }
+  } while (redispatch);
 
   return Network::FilterStatus::StopIteration;
 }

test/common/http/conn_manager_impl_test.cc

@@ -442,10 +442,6 @@ TEST_F(HttpConnectionManagerImplTest, HeaderOnlyRequestAndResponse) {
   // Kick off the incoming data. Use extra data which should cause a redispatch.
   Buffer::OwnedImpl fake_input("1234");
   conn_manager_->onData(fake_input, false);
-  // If this were actual HTTP input, the codec would kick off a fake event after
-  // the first request was finished, triggering the second onData. We do this
-  // manually.
-  conn_manager_->onData(fake_input, false);
 
   EXPECT_EQ(1U, stats_.named_.downstream_rq_2xx_.value());
   EXPECT_EQ(1U, listener_stats_.downstream_rq_2xx_.value());

test/integration/integration_test.cc

@@ -498,24 +498,23 @@ TEST_P(IntegrationTest, PipelineInline) {
   initialize();
   std::string response;
 
-  Buffer::OwnedImpl buffer("GET / HTTP/1.1\r\n\r\nGET / HTTP/1.1\r\n\r\n");
+  Buffer::OwnedImpl buffer("GET / HTTP/1.1\r\n\r\nGET / HTTP/1.0\r\n\r\n");
   RawConnectionDriver connection(
       lookupPort("http"), buffer,
       [&](Network::ClientConnection&, const Buffer::Instance& data) -> void {
         response.append(data.toString());
       },
       version_);
-  // First is an error: no host.
+
   while (response.find("400") == std::string::npos) {
     connection.run(Event::Dispatcher::RunType::NonBlock);
   }
   EXPECT_THAT(response, HasSubstr("HTTP/1.1 400 Bad Request\r\n"));
 
-  // Second response should be 400 (no host)
-  while (response.find("400") == std::string::npos) {
+  while (response.find("426") == std::string::npos) {
     connection.run(Event::Dispatcher::RunType::NonBlock);
   }
-  EXPECT_THAT(response, HasSubstr("HTTP/1.1 400 Bad Request\r\n"));
+  EXPECT_THAT(response, HasSubstr("HTTP/1.1 426 Upgrade Required\r\n"));
   connection.close();
 }