decompressor: fixing a bug for requests with added trailers

docs/root/version_history/current.rst

@@ -76,6 +76,7 @@ Bug Fixes
 * aws request signer: fix the AWS Request Signer extension to correctly normalize the path and query string to be signed according to AWS' guidelines, so that the hash on the server side matches. See `AWS SigV4 documentaion <https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html>`_.
 * cluster: delete pools when they're idle to fix unbounded memory use when using PROXY protocol upstream with tcp_proxy. This behavior can be temporarily reverted by setting the ``envoy.reloadable_features.conn_pool_delete_when_idle`` runtime guard to false.
 * cluster: finish cluster warming even if hosts are removed before health check initialization. This only affected clusters with :ref:`ignore_health_on_host_removal <envoy_v3_api_field_config.cluster.v3.Cluster.ignore_health_on_host_removal>`.
+* compressor: fix a bug where if trailers were added and a subsequent filter paused the filter chain, the request could be stalled. This behavior can be reverted by setting ``envoy.reloadable_features.fix_added_trailers`` to false.
 * dynamic forward proxy: fixing a validation bug where san and sni checks were not applied setting :ref:`http_protocol_options <envoy_v3_api_msg_extensions.upstreams.http.v3.HttpProtocolOptions>` via :ref:`typed_extension_protocol_options <envoy_v3_api_field_config.cluster.v3.Cluster.typed_extension_protocol_options>`.
 * ext_authz: fix the ext_authz filter to correctly merge multiple same headers using the ',' as separator in the check request to the external authorization service.
 * ext_authz: the network ext_authz filter now correctly sets dynamic metdata returned by the authorization service for non-OK responses. This behavior now matches the http ext_authz filter.

source/common/http/filter_manager.cc

@@ -604,6 +604,9 @@ void FilterManager::decodeData(ActiveStreamDecoderFilter* filter, Buffer::Instan
   ScopeTrackerScopeState scope(&*this, dispatcher_);
   filter_manager_callbacks_.resetIdleTimer();
 
+  const bool fix_added_trailers =
+      Runtime::runtimeFeatureEnabled("envoy.reloadable_features.fix_added_trailers");
+
   // 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_) {
@@ -690,6 +693,9 @@ void FilterManager::decodeData(ActiveStreamDecoderFilter* filter, Buffer::Instan
 
     if (!trailers_exists_at_start && filter_manager_callbacks_.requestTrailers() &&
         trailers_added_entry == decoder_filters_.end()) {
+      if (fix_added_trailers) {
+        end_stream = false;
+      }
       trailers_added_entry = entry;
     }
 
@@ -698,7 +704,11 @@ void FilterManager::decodeData(ActiveStreamDecoderFilter* filter, Buffer::Instan
       // 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
       // a previous filter has added trailers.
-      return;
+      if (fix_added_trailers) {
+        break;
+      } else {
+        return;
+      }
     }
   }
 

source/common/runtime/runtime_features.cc

@@ -64,6 +64,7 @@ constexpr const char* runtime_features[] = {
     "envoy.reloadable_features.disable_tls_inspector_injection",
     "envoy.reloadable_features.dont_add_content_length_for_bodiless_requests",
     "envoy.reloadable_features.enable_compression_without_content_length_header",
+    "envoy.reloadable_features.fix_added_trailers",
     "envoy.reloadable_features.grpc_bridge_stats_disabled",
     "envoy.reloadable_features.grpc_web_fix_non_proto_encoded_response_handling",
     "envoy.reloadable_features.grpc_json_transcoder_adhere_to_buffer_limits",

test/extensions/filters/http/decompressor/BUILD

@@ -40,6 +40,7 @@ envoy_extension_cc_test(
         "//source/extensions/compression/gzip/decompressor:config",
         "//source/extensions/filters/http/decompressor:config",
         "//test/integration:http_integration_lib",
+        "//test/integration/filters:encoder_decoder_buffer_filter_lib",
         "//test/mocks/server:factory_context_mocks",
         "//test/test_common:simulated_time_system_lib",
         "//test/test_common:utility_lib",

test/extensions/filters/http/decompressor/decompressor_filter_integration_test.cc

@@ -286,4 +286,68 @@ TEST_P(DecompressorIntegrationTest, BidirectionalDecompressionError) {
       "http.config_test.decompressor.testlib.gzip.decompressor_library.zlib_data_error", 3);
 }
 
+// Buffer the request after it's been decompressed.
+TEST_P(DecompressorIntegrationTest, DecompressAndBuffer) {
+  // filters are prepended, so add them in reverse order
+
+  config_helper_.addFilter("{ name: encoder-decoder-buffer-filter, typed_config: { \"@type\": "
+                           "type.googleapis.com/google.protobuf.Empty } }");
+
+  config_helper_.addFilter(R"EOF(
+  name: envoy.filters.http.decompressor
+  typed_config:
+    "@type": type.googleapis.com/envoy.extensions.filters.http.decompressor.v3.Decompressor
+    decompressor_library:
+      name: gzip_default
+      typed_config:
+        "@type": "type.googleapis.com/envoy.extensions.compression.gzip.decompressor.v3.Gzip"
+        window_bits: 15
+        chunk_size: 8192
+    request_direction_config:
+      common_config:
+        enabled:
+          default_value: true
+          runtime_key: request_decompressor_enabled
+    response_direction_config:
+      common_config:
+        enabled:
+          default_value: false
+          runtime_key: response_decompressor_enabled
+  )EOF");
+
+  initialize();
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+  auto encoder_decoder =
+      codec_client_->startRequest(Http::TestRequestHeaderMapImpl{{":method", "POST"},
+                                                                 {":scheme", "http"},
+                                                                 {":path", "/test/long/url"},
+                                                                 {"content-encoding", "gzip"},
+                                                                 {":authority", "host"}});
+
+  auto request_encoder = &encoder_decoder.first;
+  auto response = std::move(encoder_decoder.second);
+
+  // Compressed JSON.
+  constexpr uint8_t buffer[] = {0x1f, 0x8b, 0x08, 0x00, 0x9c, 0xb3, 0x38, 0x61, 0x00, 0x03, 0xab,
+                                0x56, 0x50, 0xca, 0xad, 0x4c, 0x29, 0xcd, 0xcd, 0xad, 0x54, 0x52,
+                                0xb0, 0x52, 0x50, 0xca, 0x2a, 0xce, 0xcf, 0x53, 0x52, 0xa8, 0xe5,
+                                0x02, 0x00, 0xa6, 0x6a, 0x24, 0x99, 0x17, 0x00, 0x00, 0x00};
+  Buffer::OwnedImpl data(buffer, 43);
+  codec_client_->sendData(*request_encoder, data, true);
+
+  waitForNextUpstreamRequest();
+
+  upstream_request_->encodeHeaders(Http::TestResponseHeaderMapImpl{{":status", "200"}}, false);
+  upstream_request_->encodeData(10, true);
+
+  ASSERT_TRUE(response->waitForEndStream());
+  EXPECT_TRUE(response->complete());
+
+  Stats::Store& stats = test_server_->server().stats();
+  Stats::CounterSharedPtr counter = TestUtility::findCounter(
+      stats, "http.config_test.decompressor.gzip_default.gzip.request.decompressed");
+  ASSERT_NE(nullptr, counter);
+  EXPECT_EQ(1L, counter->value());
+}
+
 } // namespace Envoy