http: cleaning up redispatch logic (#9208)

Signed-off-by: Alyssa Wilk <alyssar@chromium.org>

source/common/http/conn_manager_impl.cc

@@ -296,53 +296,40 @@ Network::FilterStatus ConnectionManagerImpl::onData(Buffer::Instance& data, bool
     }
   }
 
-  // TODO(alyssawilk) clean this up after #8352 is well vetted.
-  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;
+  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());
     }
 
-    // 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;
-      }
+    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;
+  }
 
-      if (!streams_.empty() && streams_.front()->state_.remote_complete_) {
-        read_callbacks_->connection().readDisable(true);
-      }
+  // Processing incoming data may release outbound data so check for closure here as well.
+  checkForDeferredClose();
+
+  // 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);
     }
-  } while (redispatch);
+  }
 
   return Network::FilterStatus::StopIteration;
 }

test/common/http/conn_manager_impl_test.cc

@@ -442,6 +442,10 @@ 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

@@ -490,6 +490,35 @@ TEST_P(IntegrationTest, Pipeline) {
   connection.close();
 }
 
+// Add a pipeline test where complete request headers in the first request merit
+// an inline sendLocalReply to make sure the "kick" works under the call stack
+// of dispatch as well as when a response is proxied from upstream.
+TEST_P(IntegrationTest, PipelineInline) {
+  autonomous_upstream_ = true;
+  initialize();
+  std::string response;
+
+  Buffer::OwnedImpl buffer("GET / HTTP/1.1\r\n\r\nGET / HTTP/1.1\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) {
+    connection.run(Event::Dispatcher::RunType::NonBlock);
+  }
+  EXPECT_THAT(response, HasSubstr("HTTP/1.1 400 Bad Request\r\n"));
+  connection.close();
+}
+
 TEST_P(IntegrationTest, NoHost) {
   initialize();
   codec_client_ = makeHttpConnection(lookupPort("http"));