http: fix filter manager dropping a body frame on continue (#46842)

changelogs/current/bug_fixes/http__filter-manager-lost-body-chunk-on-continue.rst

@@ -0,0 +1,10 @@
+Fixed a request/response body data-loss bug in the HTTP filter manager. When a filter stopped
+iteration on headers (for example a wasm filter with ``allow_on_headers_stop_iteration``, which maps
+to a single-iteration stop rather than ``StopAllIterationAndWatermark``), resumed asynchronously, and
+then on a subsequent body frame moved that frame into the filter-manager buffer via
+``addDecodedData()``/``addEncodedData()`` before returning ``Continue``, the now-empty frame was
+forwarded down the chain and the buffered bytes were silently dropped. This corrupted large streamed
+request bodies (for example one 16 KiB chunk lost when chained with an ``ext_proc`` filter in
+``FULL_DUPLEX_STREAMED`` mode). The just-buffered data is now forwarded instead of the empty frame.
+This behavioral change can be reverted by setting the runtime guard
+``envoy.reloadable_features.filter_manager_forward_added_data_on_continue`` to ``false``.

source/common/http/filter_manager.cc

@@ -201,9 +201,9 @@ void ActiveStreamFilterBase::commonHandleBufferData(Buffer::Instance& provided_d
   }
 }
 
-bool ActiveStreamFilterBase::commonHandleAfterDataCallback(FilterDataStatus status,
-                                                           Buffer::Instance& provided_data,
-                                                           bool& buffer_was_streaming) {
+bool ActiveStreamFilterBase::commonHandleAfterDataCallback(
+    FilterDataStatus status, Buffer::Instance& provided_data, bool& buffer_was_streaming,
+    bool provided_data_nonempty_before_callback) {
 
   if (status == FilterDataStatus::Continue) {
     if (iteration_state_ == IterationState::StopSingleIteration) {
@@ -212,6 +212,24 @@ bool ActiveStreamFilterBase::commonHandleAfterDataCallback(FilterDataStatus stat
       return false;
     } else {
       ASSERT(headers_continued_);
+      // A filter that is already iterating may drain the current frame into the filter-manager
+      // buffer via addDecoded/EncodedData() and then return Continue (e.g. a wasm filter resuming
+      // after buffering a body chunk while paused on headers). The signature of that drain is: the
+      // filter called addData during this callback (`filter_added_data_in_data_callback_`) and the
+      // frame went from non-empty to empty because its content was moved into bufferedData(). In
+      // that case forward the buffered frame down the chain so it is not silently lost. The
+      // non-empty-before + empty-after + flag checks keep this narrow: filters that add a
+      // *separate* buffer (e.g. the StopAll test filter, including on a zero-length end_stream
+      // frame) or that empty the frame without calling addData (e.g. a compressor buffering
+      // internally) do not match and are handled by the existing path. See
+      // https://github.com/envoyproxy/envoy/issues/46841.
+      if (Runtime::runtimeFeatureEnabled(
+              "envoy.reloadable_features.filter_manager_forward_added_data_on_continue") &&
+          parent_.state_.filter_added_data_in_data_callback_ &&
+          provided_data_nonempty_before_callback && provided_data.length() == 0 && bufferedData() &&
+          bufferedData().get() != &provided_data && bufferedData()->length() > 0) {
+        provided_data.move(*bufferedData());
+      }
     }
   } else {
     iteration_state_ = IterationState::StopSingleIteration;
@@ -739,6 +757,8 @@ void FilterManager::decodeData(ActiveStreamDecoderFilter* filter, Buffer::Instan
     recordLatestDataFilter(entry, state_.latest_data_decoding_filter_, decoder_filters_);
 
     state_.filter_call_state_ |= FilterCallState::DecodeData;
+    state_.filter_added_data_in_data_callback_ = false;
+    const bool data_nonempty_before_callback = data.length() > 0;
     (*entry)->end_stream_ = end_stream && !filter_manager_callbacks_.requestTrailers();
     FilterDataStatus status = (*entry)->handle_->decodeData(data, (*entry)->end_stream_);
     if ((*entry)->end_stream_) {
@@ -773,7 +793,8 @@ void FilterManager::decodeData(ActiveStreamDecoderFilter* filter, Buffer::Instan
     // below.
     terminal_filter_decoded_end_stream = end_stream && std::next(entry) == decoder_filters_.end();
 
-    if (!(*entry)->commonHandleAfterDataCallback(status, data, state_.decoder_filters_streaming_) &&
+    if (!(*entry)->commonHandleAfterDataCallback(status, data, state_.decoder_filters_streaming_,
+                                                 data_nonempty_before_callback) &&
         std::next(entry) != decoder_filters_.end()) {
       // Stop iteration IFF this is not the last filter. If it is the last filter, continue with
       // processing since we need to handle the case where a terminal filter wants to buffer, but
@@ -810,6 +831,11 @@ void FilterManager::addDecodedData(ActiveStreamDecoderFilter& filter, Buffer::In
       ((state_.filter_call_state_ & FilterCallState::DecodeTrailers) && !filter.canIterate())) {
     // Make sure if this triggers watermarks, the correct action is taken.
     state_.decoder_filters_streaming_ = streaming;
+    // Record that a filter drained data into the buffer during its own decodeData() callback so
+    // commonHandleAfterDataCallback() can forward it if the current frame was emptied. See #46841.
+    if (state_.filter_call_state_ & FilterCallState::DecodeData) {
+      state_.filter_added_data_in_data_callback_ = true;
+    }
     // If no call is happening or we are in the decode headers/data callback, buffer the data.
     // Inline processing happens in the decodeHeaders() callback if necessary.
     filter.commonHandleBufferData(data);
@@ -1422,6 +1448,11 @@ void FilterManager::addEncodedData(ActiveStreamEncoderFilter& filter, Buffer::In
       ((state_.filter_call_state_ & FilterCallState::EncodeTrailers) && !filter.canIterate())) {
     // Make sure if this triggers watermarks, the correct action is taken.
     state_.encoder_filters_streaming_ = streaming;
+    // Record that a filter drained data into the buffer during its own encodeData() callback so
+    // commonHandleAfterDataCallback() can forward it if the current frame was emptied. See #46841.
+    if (state_.filter_call_state_ & FilterCallState::EncodeData) {
+      state_.filter_added_data_in_data_callback_ = true;
+    }
     // If no call is happening or we are in the decode headers/data callback, buffer the data.
     // Inline processing happens in the decodeHeaders() callback if necessary.
     filter.commonHandleBufferData(data);
@@ -1470,6 +1501,8 @@ void FilterManager::encodeData(ActiveStreamEncoderFilter* filter, Buffer::Instan
 
     recordLatestDataFilter(entry, state_.latest_data_encoding_filter_, encoder_filters_);
 
+    state_.filter_added_data_in_data_callback_ = false;
+    const bool data_nonempty_before_callback = data.length() > 0;
     (*entry)->end_stream_ = end_stream && !filter_manager_callbacks_.responseTrailers();
     FilterDataStatus status = (*entry)->handle_->encodeData(data, (*entry)->end_stream_);
     if (state_.encoder_filter_chain_aborted_) {
@@ -1492,7 +1525,8 @@ void FilterManager::encodeData(ActiveStreamEncoderFilter* filter, Buffer::Instan
       trailers_added_entry = entry;
     }
 
-    if (!(*entry)->commonHandleAfterDataCallback(status, data, state_.encoder_filters_streaming_)) {
+    if (!(*entry)->commonHandleAfterDataCallback(status, data, state_.encoder_filters_streaming_,
+                                                 data_nonempty_before_callback)) {
       return;
     }
   }

source/common/http/filter_manager.h

@@ -118,7 +118,8 @@ struct ActiveStreamFilterBase : public virtual StreamFilterCallbacks,
   bool commonHandleAfter1xxHeadersCallback(Filter1xxHeadersStatus status);
   bool commonHandleAfterHeadersCallback(FilterHeadersStatus status, bool& end_stream);
   bool commonHandleAfterDataCallback(FilterDataStatus status, Buffer::Instance& provided_data,
-                                     bool& buffer_was_streaming);
+                                     bool& buffer_was_streaming,
+                                     bool provided_data_nonempty_before_callback);
   bool commonHandleAfterTrailersCallback(FilterTrailersStatus status);
 
   // Buffers provided_data.
@@ -965,6 +966,14 @@ class FilterManager : public ScopeTrackedObject,
     bool decoder_filters_streaming_{true};
     bool destroyed_{false};
 
+    // Set true when a filter calls addDecodedData()/addEncodedData() during its own
+    // decodeData()/encodeData() callback. Reset immediately before each data callback. Combined
+    // with a frame that went from non-empty to empty across the callback, this signals the filter
+    // drained the current frame into the filter-manager buffer, so commonHandleAfterDataCallback()
+    // must forward the buffered data instead of the now-empty frame. See
+    // https://github.com/envoyproxy/envoy/issues/46841.
+    bool filter_added_data_in_data_callback_{false};
+
     // Result of filter chain creation.
     CreateChainResult create_chain_result_;
 

source/common/runtime/runtime_features.cc

@@ -46,6 +46,11 @@ RUNTIME_GUARD(envoy_reloadable_features_explicit_internal_address_config);
 RUNTIME_GUARD(envoy_reloadable_features_ext_proc_fail_close_spurious_resp);
 RUNTIME_GUARD(envoy_reloadable_features_ext_proc_inject_data_with_state_update);
 RUNTIME_GUARD(envoy_reloadable_features_filter_chain_aborted_can_not_continue);
+// When a filter drains the current data frame into the filter-manager buffer via
+// addDecoded/EncodedData() and then returns Continue (e.g. a wasm filter resuming after buffering),
+// forward that buffered data down the chain instead of the now-empty frame, so the frame is not
+// lost. See https://github.com/envoyproxy/envoy/issues/46841
+RUNTIME_GUARD(envoy_reloadable_features_filter_manager_forward_added_data_on_continue);
 RUNTIME_GUARD(envoy_reloadable_features_gcp_authn_use_fixed_url);
 RUNTIME_GUARD(envoy_reloadable_features_getaddrinfo_num_retries);
 RUNTIME_GUARD(envoy_reloadable_features_grpc_side_stream_flow_control);

test/common/http/filter_manager_test.cc

@@ -71,6 +71,8 @@ class FilterManagerTest : public testing::Test {
     EXPECT_TRUE(MessageDifferencer::Equals(*(fs_value->serializeAsProto()), *expected));
   }
 
+  void runAddDecodedDataOnContinueTest(bool forward_data);
+
   std::unique_ptr<DownstreamFilterManager> filter_manager_;
   NiceMock<MockFilterManagerCallbacks> filter_manager_callbacks_;
   NiceMock<Event::MockDispatcher> dispatcher_;
@@ -937,6 +939,113 @@ TEST_F(FilterManagerTest, AllDecodeOperationsBlockedAfterDownstreamReset) {
   filter_manager_->destroyFilters();
 }
 
+// Reproduces the request-body frame loss reported in
+// https://github.com/Kuadrant/wasm-shim/issues/388 (and Envoy #46841): a filter
+// that returns StopIteration on headers (e.g. a wasm filter with
+// allow_on_headers_stop_iteration, mapped to IterationState::StopSingleIteration)
+// buffers a body chunk, continues asynchronously, and then on the next chunk
+// drains the frame into the filter-manager buffer via addDecodedData() while
+// returning Continue. Because the filter's IterationState is already Continue at
+// that point, commonHandleAfterDataCallback() forwards the now-empty frame
+// instead of the just-buffered data, so a downstream FULL_DUPLEX_STREAMED
+// ext_proc filter never receives that chunk. Exactly one mid-stream frame is
+// lost while the message count is unchanged.
+//
+// The `forward_data` parameter toggles the
+// `filter_manager_forward_added_data_on_continue` runtime guard: when disabled
+// this documents the pre-fix data loss; when enabled it verifies the fix.
+void FilterManagerTest::runAddDecodedDataOnContinueTest(bool forward_data) {
+  TestScopedRuntime scoped_runtime;
+  scoped_runtime.mergeValues(
+      {{"envoy.reloadable_features.filter_manager_forward_added_data_on_continue",
+        forward_data ? "true" : "false"}});
+
+  initialize();
+
+  // filter_1 mimics the wasm filter (stops on headers, buffers, continues after
+  // an async callout); filter_2 mimics a FULL_DUPLEX_STREAMED ext_proc filter
+  // that streams (does not re-buffer) every request body byte it receives.
+  std::shared_ptr<MockStreamDecoderFilter> filter_1(new NiceMock<MockStreamDecoderFilter>());
+  std::shared_ptr<MockStreamDecoderFilter> filter_2(new NiceMock<MockStreamDecoderFilter>());
+
+  EXPECT_CALL(filter_factory_, createFilterChain(_))
+      .WillRepeatedly(Invoke([&](FilterChainManager& manager) -> bool {
+        auto factory = createDecoderFilterFactoryCb(filter_1);
+        manager.applyFilterFactoryCb({}, factory);
+        factory = createDecoderFilterFactoryCb(filter_2);
+        manager.applyFilterFactoryCb({}, factory);
+        return true;
+      }));
+
+  RequestHeaderMapPtr headers{
+      new TestRequestHeaderMapImpl{{":authority", "host"}, {":path", "/"}, {":method", "POST"}}};
+  ON_CALL(filter_manager_callbacks_, requestHeaders()).WillByDefault(Return(makeOptRef(*headers)));
+
+  filter_manager_->createDownstreamFilterChain();
+  filter_manager_->requestHeadersInitialized();
+
+  // filter_2 accumulates and drains (streams) every request body byte it sees.
+  std::string filter_2_received;
+  ON_CALL(*filter_2, decodeData(_, _))
+      .WillByDefault(Invoke([&](Buffer::Instance& data, bool) -> FilterDataStatus {
+        filter_2_received += data.toString();
+        data.drain(data.length());
+        return FilterDataStatus::StopIterationNoBuffer;
+      }));
+
+  // filter_1 stops iteration on headers to await an async callout; filter_2 must
+  // not see the headers yet.
+  EXPECT_CALL(*filter_1, decodeHeaders(_, false))
+      .WillOnce(Return(FilterHeadersStatus::StopIteration));
+  filter_manager_->decodeHeaders(*headers, false);
+
+  // Chunk A arrives while filter_1 is paused: buffer it in the filter manager.
+  EXPECT_CALL(*filter_1, decodeData(_, false))
+      .WillOnce(Return(FilterDataStatus::StopIterationAndBuffer));
+  Buffer::OwnedImpl chunk_a("AAAA");
+  filter_manager_->decodeData(chunk_a, false);
+
+  // The async callout completes: filter_1 continues, which delivers the buffered
+  // chunk A (and the headers) down to filter_2.
+  EXPECT_CALL(*filter_2, decodeHeaders(_, false)).WillOnce(Return(FilterHeadersStatus::Continue));
+  filter_1->callbacks_->continueDecoding();
+  EXPECT_EQ("AAAA", filter_2_received);
+
+  // Chunk B arrives right after the continue. filter_1, still in its buffering
+  // mode, moves the frame into the filter-manager buffer via addDecodedData()
+  // and then returns Continue. This frame must still reach filter_2.
+  EXPECT_CALL(*filter_1, decodeData(_, false))
+      .WillOnce(Invoke([&](Buffer::Instance& data, bool) -> FilterDataStatus {
+        filter_1->callbacks_->addDecodedData(data, false);
+        return FilterDataStatus::Continue;
+      }));
+  Buffer::OwnedImpl chunk_b("BBBB");
+  filter_manager_->decodeData(chunk_b, false);
+
+  // Chunk C (end of stream): filter_1 passes it straight through.
+  EXPECT_CALL(*filter_1, decodeData(_, true)).WillOnce(Return(FilterDataStatus::Continue));
+  Buffer::OwnedImpl chunk_c("CCCC");
+  filter_manager_->decodeData(chunk_c, true);
+
+  // With the fix (guard enabled) all three frames reach filter_2. Without it,
+  // chunk B is dropped and filter_2 only sees "AAAACCCC".
+  if (forward_data) {
+    EXPECT_EQ("AAAABBBBCCCC", filter_2_received);
+  } else {
+    EXPECT_EQ("AAAACCCC", filter_2_received);
+  }
+
+  filter_manager_->destroyFilters();
+}
+
+TEST_F(FilterManagerTest, DecodeDataFrameLostAfterContinueWithoutGuard) {
+  runAddDecodedDataOnContinueTest(/*forward_data=*/false);
+}
+
+TEST_F(FilterManagerTest, DecodeDataFrameNotLostAfterContinueWithAddDecodedData) {
+  runAddDecodedDataOnContinueTest(/*forward_data=*/true);
+}
+
 } // namespace
 } // namespace Http
 } // namespace Envoy

test/extensions/filters/http/ext_proc/BUILD

@@ -194,6 +194,7 @@ envoy_extension_cc_test(
         "//source/extensions/retry/host/previous_hosts:config",
         "//test/common/http:common_lib",
         "//test/integration:http_integration_lib",
+        "//test/integration/filters:add_data_and_continue_filter_lib",
         "//test/integration/filters:common_lib",
         "//test/integration/filters:stream_info_to_headers_filter_lib",
         "//test/proto:helloworld_proto_cc_proto",

test/extensions/filters/http/ext_proc/ext_proc_integration_test.cc

@@ -5619,6 +5619,59 @@ TEST_P(ExtProcIntegrationTest, ServerWaitForBodyBeforeSendsHeaderRespDuplexStrea
   verifyDownstreamResponse(*response, 200);
 }
 
+// Regression test for https://github.com/envoyproxy/envoy/issues/46841
+// A filter ahead of ext_proc moves the first request body frame into the
+// filter-manager buffer via addDecodedData() and then returns Continue
+// (the pattern a wasm filter follows when it resumes after buffering a chunk).
+// The just-buffered frame must still be forwarded down the chain, so the
+// ext_proc server in FULL_DUPLEX_STREAMED mode must observe the full
+// request body byte-for-byte. Before the fix the first frame was dropped.
+TEST_P(ExtProcIntegrationTest, AddDataAndContinueBeforeExtProcDuplexStreamed) {
+  const std::string body_sent(64 * 1024, 's');
+
+  auto* processing_mode = proto_config_.mutable_processing_mode();
+  processing_mode->set_request_header_mode(ProcessingMode::SEND);
+  processing_mode->set_request_body_mode(ProcessingMode::FULL_DUPLEX_STREAMED);
+  processing_mode->set_request_trailer_mode(ProcessingMode::SEND);
+  processing_mode->set_response_header_mode(ProcessingMode::SKIP);
+
+  // initializeConfig() prepends the ext_proc filter; prepending our filter
+  // afterwards places it ahead of ext_proc in the decode chain, i.e.
+  // [add-data-and-continue-filter, ext_proc].
+  initializeConfig();
+  config_helper_.prependFilter(R"EOF(
+    name: add-data-and-continue-filter
+    typed_config:
+      "@type": type.googleapis.com/test.integration.filters.AddDataAndContinueFilterConfig
+  )EOF");
+  HttpIntegrationTest::initialize();
+
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+  Http::TestRequestHeaderMapImpl default_headers;
+  HttpTestUtility::addDefaultHeaders(default_headers);
+  auto encoder_decoder = codec_client_->startRequest(default_headers);
+  request_encoder_ = &encoder_decoder.first;
+  IntegrationStreamDecoderPtr response = std::move(encoder_decoder.second);
+  codec_client_->sendData(*request_encoder_, body_sent, true);
+
+  // The ext_proc server receives the headers.
+  ProcessingRequest header_request;
+  serverReceiveHeaderReq(header_request);
+  // The ext_proc server must receive the entire request body, byte-for-byte.
+  uint32_t total_req_body_msg = serverReceiveBodyDuplexStreamed(body_sent, processor_stream_);
+  EXPECT_GT(total_req_body_msg, 0);
+
+  // Send responses back so the request completes cleanly.
+  serverSendHeaderResp();
+  uint32_t total_resp_body_msg = 2 * total_req_body_msg;
+  const std::string body_upstream(total_resp_body_msg, 'r');
+  serverSendBodyRespDuplexStreamed(total_resp_body_msg, processor_stream_);
+
+  handleUpstreamRequest();
+  EXPECT_EQ(upstream_request_->body().toString(), body_upstream);
+  verifyDownstreamResponse(*response, 200);
+}
+
 // Buffer the whole message including header, body and trailer before sending response.
 TEST_P(ExtProcIntegrationTest,
        ServerWaitForBodyAndTrailerBeforeSendsHeaderRespDuplexStreamedSmallBody) {

test/integration/filters/BUILD

@@ -430,6 +430,21 @@ envoy_cc_test_library(
     ],
 )
 
+envoy_cc_test_library(
+    name = "add_data_and_continue_filter_lib",
+    srcs = [
+        "add_data_and_continue_filter.cc",
+    ],
+    deps = [
+        ":test_filters_proto_cc_proto",
+        "//envoy/http:filter_interface",
+        "//envoy/registry",
+        "//envoy/server:filter_config_interface",
+        "//source/extensions/filters/http/common:pass_through_filter_lib",
+        "//test/extensions/filters/http/common:empty_http_filter_config_lib",
+    ],
+)
+
 envoy_cc_test_library(
     name = "on_local_reply_filter_config_lib",
     srcs = [