Common: Introduce StopAllIteration filter status for decoding and encoding filters

include/envoy/http/filter.h

@@ -33,7 +33,30 @@ enum class FilterHeadersStatus {
   StopIteration,
   // Continue iteration to remaining filters, but ignore any subsequent data or trailers. This
   // results in creating a header only request/response.
-  ContinueAndEndStream
+  ContinueAndEndStream,
+  // Do not iterate for headers as well as data and trailers for the current filter and the filters
+  // following, and buffer body data for later dispatching. ContinueDecoding() MUST
+  // be called if continued filter iteration is desired.
+  //
+  // Used when a filter wants to stop iteration on data and trailers while waiting for headers'
+  // iteration to resume.
+  //
+  // If buffering the request causes buffered data to exceed the configured buffer limit, a 413 will
+  // be sent to the user. On the response path exceeding buffer limits will result in a 500.
+  //
+  // TODO(soya3129): stop metadata parsing when StopAllIterationAndBuffer is set.
+  StopAllIterationAndBuffer,
+  // Do not iterate for headers as well as data and trailers for the current filter and the filters
+  // following, and buffer body data for later dispatching. continueDecoding() MUST
+  // be called if continued filter iteration is desired.
+  //
+  // Used when a filter wants to stop iteration on data and trailers while waiting for headers'
+  // iteration to resume.
+  //
+  // This will cause the flow of incoming data to cease until continueDecoding() function is called.
+  //
+  // TODO(soya3129): stop metadata parsing when StopAllIterationAndWatermark is set.
+  StopAllIterationAndWatermark,
 };
 
 /**

source/common/http/conn_manager_impl.h

@@ -95,16 +95,26 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
    */
   struct ActiveStreamFilterBase : public virtual StreamFilterCallbacks {
     ActiveStreamFilterBase(ActiveStream& parent, bool dual_filter)
-        : parent_(parent), headers_continued_(false), continue_headers_continued_(false),
-          stopped_(false), end_stream_(false), dual_filter_(dual_filter) {}
+        : iteration_state_(IterationState::Continue), iterate_from_current_filter_(false),
+          parent_(parent), headers_continued_(false), continue_headers_continued_(false),
+          end_stream_(false), dual_filter_(dual_filter) {}
 
+    // Functions in the following block are called after the filter finishes processing
+    // corresponding data. Those functions handle state updates and data storage (if needed)
+    // according to the status returned by filter's callback functions.
     bool commonHandleAfter100ContinueHeadersCallback(FilterHeadersStatus status);
     bool commonHandleAfterHeadersCallback(FilterHeadersStatus status, bool& headers_only);
-    void commonHandleBufferData(Buffer::Instance& provided_data);
     bool commonHandleAfterDataCallback(FilterDataStatus status, Buffer::Instance& provided_data,
                                        bool& buffer_was_streaming);
     bool commonHandleAfterTrailersCallback(FilterTrailersStatus status);
 
+    // Buffers provided_data.
+    void commonHandleBufferData(Buffer::Instance& provided_data);
+
+    // If iteration has stopped for all frame types, calls this function to buffer the data before
+    // the filter processes data. The function also updates streaming state.
+    void commonBufferDataIfStopAll(Buffer::Instance& provided_data, bool& buffer_was_streaming);
+
     void commonContinue();
     virtual bool canContinue() PURE;
     virtual Buffer::WatermarkBufferPtr createBuffer() PURE;
@@ -128,10 +138,35 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     Tracing::Span& activeSpan() override;
     Tracing::Config& tracingConfig() override;
 
+    // Functions to set or get iteration state.
+    bool canIterate() { return iteration_state_ == IterationState::Continue; }
+    bool stoppedAll() {
+      return iteration_state_ == IterationState::StopAllBuffer ||
+             iteration_state_ == IterationState::StopAllWatermark;
+    }
+    void allowIteration() {
+      ASSERT(iteration_state_ != IterationState::Continue);
+      iteration_state_ = IterationState::Continue;
+    }
+
+    // The state of iteration.
+    enum class IterationState {
+      Continue,            // Iteration has not stopped for any frame type.
+      StopSingleIteration, // Iteration has stopped for headers, 100-continue, or data.
+      StopAllBuffer,       // Iteration has stopped for all frame types, and following data should
+                           // be buffered.
+      StopAllWatermark,    // Iteration has stopped for all frame types, and following data should
+                           // be buffered until high watermark is reached.
+    };
+    IterationState iteration_state_;
+    // If the filter resumes iteration from a StopAllBuffer/Watermark state, the current filter
+    // hasn't parsed data and trailers. As a result, the filter iteration should start with the
+    // current filter instead of the next one. If true, filter iteration starts with the current
+    // filter. Otherwise, starts with the next filter in the chain.
+    bool iterate_from_current_filter_;
     ActiveStream& parent_;
     bool headers_continued_ : 1;
     bool continue_headers_continued_ : 1;
-    bool stopped_ : 1;
     // If true, end_stream is called for this filter.
     bool end_stream_ : 1;
     const bool dual_filter_ : 1;
@@ -290,11 +325,22 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     ActiveStream(ConnectionManagerImpl& connection_manager);
     ~ActiveStream();
 
+    // Indicates which filter to start the iteration with.
+    enum class FilterIterationStartState { Always_start_from_next, Can_start_from_current };
+
     void addStreamDecoderFilterWorker(StreamDecoderFilterSharedPtr filter, bool dual_filter);
     void addStreamEncoderFilterWorker(StreamEncoderFilterSharedPtr filter, bool dual_filter);
     void chargeStats(const HeaderMap& headers);
+    // Returns the encoder filter to start iteration with. If the function is called from a filter
+    // that should always iterate from the next filter, always_start_next should be set to true.
     std::list<ActiveStreamEncoderFilterPtr>::iterator
-    commonEncodePrefix(ActiveStreamEncoderFilter* filter, bool end_stream);
+    commonEncodePrefix(ActiveStreamEncoderFilter* filter, bool end_stream,
+                       FilterIterationStartState filter_iteration_start_state);
+    // Returns the decoder filter to start iteration with. If the function is called from a filter
+    // that should always iterate from the next filter, always_start_next should be set to true.
+    std::list<ActiveStreamDecoderFilterPtr>::iterator
+    commonDecodePrefix(ActiveStreamDecoderFilter* filter,
+                       FilterIterationStartState filter_iteration_start_state);
     const Network::Connection* connection();
     void addDecodedData(ActiveStreamDecoderFilter& filter, Buffer::Instance& data, bool streaming);
     HeaderMap& addDecodedTrailers();
@@ -316,6 +362,10 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     void encodeMetadata(ActiveStreamEncoderFilter* filter, MetadataMapPtr&& metadata_map_ptr);
     void maybeEndEncode(bool end_stream);
     uint64_t streamId() { return stream_id_; }
+    // Returns true if filter has stopped iteration for all frame types. Otherwise, returns false.
+    // filter_streaming is the variable to indicate if stream is streaming.
+    bool handleDataIfStopAll(ActiveStreamFilterBase& filter, Buffer::Instance& data,
+                             bool& filter_streaming);
 
     // Http::StreamCallbacks
     void onResetStream(StreamResetReason reason,

test/common/http/conn_manager_impl_test.cc

@@ -168,22 +168,40 @@ class HttpConnectionManagerImplTest : public testing::Test, public ConnectionMan
     EXPECT_CALL(stream_, bufferLimit()).WillOnce(Return(initial_buffer_limit_));
   }
 
-  void setUpEncoderAndDecoder() {
+  // If request_with_data_and_trailers is true, includes data and trailers in the request. If
+  // decode_headers_stop_all is true, decoder_filters_[0]'s callback decodeHeaders() returns
+  // StopAllIterationAndBuffer.
+  void setUpEncoderAndDecoder(bool request_with_data_and_trailers, bool decode_headers_stop_all) {
     setUpBufferLimits();
-    EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> void {
-      StreamDecoder* decoder = &conn_manager_->newStream(response_encoder_);
-      HeaderMapPtr headers{
-          new TestHeaderMapImpl{{":authority", "host"}, {":path", "/"}, {":method", "GET"}}};
-      decoder->decodeHeaders(std::move(headers), true);
-    }));
+    EXPECT_CALL(*codec_, dispatch(_))
+        .WillOnce(Invoke([&, request_with_data_and_trailers](Buffer::Instance&) -> void {
+          StreamDecoder* decoder = &conn_manager_->newStream(response_encoder_);
+          HeaderMapPtr headers{
+              new TestHeaderMapImpl{{":authority", "host"}, {":path", "/"}, {":method", "GET"}}};
+          if (request_with_data_and_trailers) {
+            decoder->decodeHeaders(std::move(headers), false);
+
+            Buffer::OwnedImpl fake_data("12345");
+            decoder->decodeData(fake_data, false);
+
+            HeaderMapPtr trailers{new TestHeaderMapImpl{{"foo", "bar"}}};
+            decoder->decodeTrailers(std::move(trailers));
+          } else {
+            decoder->decodeHeaders(std::move(headers), true);
+          }
+        }));
 
     setupFilterChain(2, 2);
 
-    EXPECT_CALL(*decoder_filters_[0], decodeHeaders(_, true))
-        .WillOnce(InvokeWithoutArgs([&]() -> FilterHeadersStatus {
+    EXPECT_CALL(*decoder_filters_[0], decodeHeaders(_, _))
+        .WillOnce(InvokeWithoutArgs([&, decode_headers_stop_all]() -> FilterHeadersStatus {
           Buffer::OwnedImpl data("hello");
           decoder_filters_[0]->callbacks_->addDecodedData(data, true);
-          return FilterHeadersStatus::Continue;
+          if (decode_headers_stop_all) {
+            return FilterHeadersStatus::StopAllIterationAndBuffer;
+          } else {
+            return FilterHeadersStatus::Continue;
+          }
         }));
     EXPECT_CALL(*decoder_filters_[0], decodeComplete());
   }
@@ -434,7 +452,7 @@ TEST_F(HttpConnectionManagerImplTest, 100ContinueResponse) {
 TEST_F(HttpConnectionManagerImplTest, 100ContinueResponseWithEncoderFiltersProxyingDisabled) {
   proxy_100_continue_ = false;
   setup(false, "envoy-custom-server", false);
-  setUpEncoderAndDecoder();
+  setUpEncoderAndDecoder(false, false);
   sendRequestHeadersAndData();
 
   // Akin to 100ContinueResponseWithEncoderFilters below, but with
@@ -458,7 +476,7 @@ TEST_F(HttpConnectionManagerImplTest, 100ContinueResponseWithEncoderFiltersProxy
 TEST_F(HttpConnectionManagerImplTest, 100ContinueResponseWithEncoderFilters) {
   proxy_100_continue_ = true;
   setup(false, "envoy-custom-server", false);
-  setUpEncoderAndDecoder();
+  setUpEncoderAndDecoder(false, false);
   sendRequestHeadersAndData();
 
   EXPECT_CALL(*encoder_filters_[0], encode100ContinueHeaders(_))
@@ -481,7 +499,7 @@ TEST_F(HttpConnectionManagerImplTest, 100ContinueResponseWithEncoderFilters) {
 TEST_F(HttpConnectionManagerImplTest, PauseResume100Continue) {
   proxy_100_continue_ = true;
   setup(false, "envoy-custom-server", false);
-  setUpEncoderAndDecoder();
+  setUpEncoderAndDecoder(false, false);
   sendRequestHeadersAndData();
 
   // Stop the 100-Continue at encoder filter 1. Encoder filter 0 should not yet receive the
@@ -2646,7 +2664,7 @@ TEST_F(HttpConnectionManagerImplTest, FilterClearRouteCache) {
 
 TEST_F(HttpConnectionManagerImplTest, UpstreamWatermarkCallbacks) {
   setup(false, "");
-  setUpEncoderAndDecoder();
+  setUpEncoderAndDecoder(false, false);
   sendRequestHeadersAndData();
 
   // Mimic the upstream connection backing up. The router would call
@@ -2811,7 +2829,7 @@ TEST_F(HttpConnectionManagerImplTest, UnderlyingConnectionWatermarksUnwoundWithL
 TEST_F(HttpConnectionManagerImplTest, AlterFilterWatermarkLimits) {
   initial_buffer_limit_ = 100;
   setup(false, "");
-  setUpEncoderAndDecoder();
+  setUpEncoderAndDecoder(false, false);
   sendRequestHeadersAndData();
 
   // Check initial limits.
@@ -2840,7 +2858,7 @@ TEST_F(HttpConnectionManagerImplTest, HitFilterWatermarkLimits) {
   initial_buffer_limit_ = 1;
   streaming_filter_ = true;
   setup(false, "");
-  setUpEncoderAndDecoder();
+  setUpEncoderAndDecoder(false, false);
 
   // The filter is a streaming filter. Sending 4 bytes should hit the
   // watermark limit and disable reads on the stream.
@@ -2880,7 +2898,7 @@ TEST_F(HttpConnectionManagerImplTest, HitRequestBufferLimits) {
   initial_buffer_limit_ = 10;
   streaming_filter_ = false;
   setup(false, "");
-  setUpEncoderAndDecoder();
+  setUpEncoderAndDecoder(false, false);
   sendRequestHeadersAndData();
 
   // Set the filter to be a buffering filter. Sending any data will hit the
@@ -2941,7 +2959,7 @@ TEST_F(HttpConnectionManagerImplTest, HitRequestBufferLimitsIntermediateFilter)
 TEST_F(HttpConnectionManagerImplTest, HitResponseBufferLimitsBeforeHeaders) {
   initial_buffer_limit_ = 10;
   setup(false, "");
-  setUpEncoderAndDecoder();
+  setUpEncoderAndDecoder(false, false);
   sendRequestHeadersAndData();
 
   // Start the response without processing the request headers through all
@@ -2975,7 +2993,7 @@ TEST_F(HttpConnectionManagerImplTest, HitResponseBufferLimitsBeforeHeaders) {
 TEST_F(HttpConnectionManagerImplTest, HitResponseBufferLimitsAfterHeaders) {
   initial_buffer_limit_ = 10;
   setup(false, "");
-  setUpEncoderAndDecoder();
+  setUpEncoderAndDecoder(false, false);
   sendRequestHeadersAndData();
 
   // Start the response, and make sure the request headers are fully processed.
@@ -3810,5 +3828,90 @@ TEST_F(HttpConnectionManagerImplTest, OverlyLongHeadersAcceptedIfConfigured) {
   conn_manager_->onData(fake_input, false); // kick off request
 }
 
+TEST_F(HttpConnectionManagerImplTest, TestStopAllIterationAndBufferOnDecodingPathFirstFilter) {
+  setup(false, "envoy-custom-server", false);
+  setUpEncoderAndDecoder(true, true);
+
+  // Kick off the incoming data.
+  Buffer::OwnedImpl fake_input("1234");
+  conn_manager_->onData(fake_input, false);
+
+  // Verify that once the decoder_filters_[0]'s contineDecoding() is called, decoder_filters_[1]'s
+  // decodeHeaders() is called, and both filters receive data and trailers consequently.
+  EXPECT_CALL(*decoder_filters_[1], decodeHeaders(_, _))
+      .WillOnce(Return(FilterHeadersStatus::Continue));
+  EXPECT_CALL(*decoder_filters_[0], decodeData(_, _)).WillOnce(Return(FilterDataStatus::Continue));
+  EXPECT_CALL(*decoder_filters_[1], decodeData(_, _)).WillOnce(Return(FilterDataStatus::Continue));
+  EXPECT_CALL(*decoder_filters_[0], decodeTrailers(_))
+      .WillOnce(Return(FilterTrailersStatus::Continue));
+  EXPECT_CALL(*decoder_filters_[1], decodeTrailers(_))
+      .WillOnce(Return(FilterTrailersStatus::Continue));
+  EXPECT_CALL(*decoder_filters_[1], decodeComplete());
+  decoder_filters_[0]->callbacks_->continueDecoding();
+}
+
+TEST_F(HttpConnectionManagerImplTest, TestStopAllIterationAndBufferOnDecodingPathSecondFilter) {
+  setup(false, "envoy-custom-server", false);
+  setUpEncoderAndDecoder(true, false);
+
+  // Verify headers go through both filters, and data and trailers go through the first filter only.
+  EXPECT_CALL(*decoder_filters_[1], decodeHeaders(_, _))
+      .WillOnce(Return(FilterHeadersStatus::StopAllIterationAndBuffer));
+  EXPECT_CALL(*decoder_filters_[0], decodeData(_, _)).WillOnce(Return(FilterDataStatus::Continue));
+  EXPECT_CALL(*decoder_filters_[0], decodeTrailers(_))
+      .WillOnce(Return(FilterTrailersStatus::Continue));
+  // Kick off the incoming data.
+  Buffer::OwnedImpl fake_input("1234");
+  conn_manager_->onData(fake_input, false);
+
+  // Verify that once the decoder_filters_[1]'s contineDecoding() is called, both data and trailers
+  // go through the second filter.
+  EXPECT_CALL(*decoder_filters_[1], decodeData(_, _)).WillOnce(Return(FilterDataStatus::Continue));
+  EXPECT_CALL(*decoder_filters_[1], decodeTrailers(_))
+      .WillOnce(Return(FilterTrailersStatus::Continue));
+  EXPECT_CALL(*decoder_filters_[1], decodeComplete());
+  decoder_filters_[1]->callbacks_->continueDecoding();
+}
+
+TEST_F(HttpConnectionManagerImplTest, TestStopAllIterationAndBufferOnEncodingPath) {
+  setup(false, "envoy-custom-server", false);
+  setUpEncoderAndDecoder(false, false);
+  sendRequestHeadersAndData();
+
+  // encoder_filters_[1] is the first filter in the chain.
+  EXPECT_CALL(*encoder_filters_[1], encodeHeaders(_, false))
+      .WillOnce(Invoke([&](HeaderMap&, bool) -> FilterHeadersStatus {
+        return FilterHeadersStatus::StopAllIterationAndBuffer;
+      }));
+  HeaderMapPtr response_headers{new TestHeaderMapImpl{{":status", "200"}}};
+  decoder_filters_[0]->callbacks_->encodeHeaders(std::move(response_headers), false);
+
+  // Invoke encodeData while all iteration is stopped and make sure the filters do not have
+  // encodeData called.
+  EXPECT_CALL(*encoder_filters_[0], encodeData(_, _)).Times(0);
+  EXPECT_CALL(*encoder_filters_[1], encodeData(_, _)).Times(0);
+  Buffer::OwnedImpl response_body("response");
+  decoder_filters_[0]->callbacks_->encodeData(response_body, false);
+  decoder_filters_[0]->callbacks_->encodeTrailers(
+      HeaderMapPtr{new TestHeaderMapImpl{{"some", "trailer"}}});
+
+  // Verify that once encoder_filters_[1]'s continueEncoding() is called, encoder_filters_[0]'s
+  // encodeHeaders() is called, and both filters receive data and trailers consequently.
+  EXPECT_CALL(*encoder_filters_[0], encodeHeaders(_, _))
+      .WillOnce(Return(FilterHeadersStatus::Continue));
+  EXPECT_CALL(response_encoder_, encodeHeaders(_, false));
+  EXPECT_CALL(*encoder_filters_[1], encodeData(_, _)).WillOnce(Return(FilterDataStatus::Continue));
+  EXPECT_CALL(*encoder_filters_[0], encodeData(_, _)).WillOnce(Return(FilterDataStatus::Continue));
+  EXPECT_CALL(response_encoder_, encodeData(_, _));
+  EXPECT_CALL(*encoder_filters_[1], encodeTrailers(_))
+      .WillOnce(Return(FilterTrailersStatus::Continue));
+  EXPECT_CALL(*encoder_filters_[0], encodeTrailers(_))
+      .WillOnce(Return(FilterTrailersStatus::Continue));
+  EXPECT_CALL(response_encoder_, encodeTrailers(_));
+  EXPECT_CALL(*encoder_filters_[0], encodeComplete());
+  EXPECT_CALL(*encoder_filters_[1], encodeComplete());
+  expectOnDestroy();
+  encoder_filters_[1]->callbacks_->continueEncoding();
+}
 } // namespace Http
 } // namespace Envoy

test/integration/BUILD

@@ -272,6 +272,9 @@ envoy_cc_test_library(
         "//source/extensions/filters/network/http_connection_manager:config",
         "//test/common/upstream:utility_lib",
         "//test/integration/filters:add_trailers_filter_config_lib",
+        "//test/integration/filters:call_decodedata_once_filter_config_lib",
+        "//test/integration/filters:decode_headers_return_stop_all_filter_config_lib",
+        "//test/integration/filters:encode_headers_return_stop_all_filter_config_lib",
         "//test/integration/filters:headers_only_filter_config_lib",
         "//test/integration/filters:modify_buffer_filter_config_lib",
         "//test/integration/filters:passthrough_filter_config_lib",