filers/ondemand: protect body behaviour change

runtime flag to protect the behaviour change

Signed-off-by: William Dauchy <william.dauchy@datadoghq.com>

Author: wdauchy

changelogs/current.yaml

@@ -325,6 +325,11 @@ new_features:
   change: |
     Added runtime guard ``envoy.reloadable_features.report_load_when_rq_active_is_non_zero``.
     When enabled, LRS continues to send locality_stats reoprt to config server when there is no request_issued in the poll cycle.
+- area: on_demand
+  change: |
+    Added runtime guard ``envoy.reloadable_features.on_demand_track_end_stream``.
+    When enabled, the on_demand filter tracks downstream end_stream state to support stream recreation with fully read request bodies.
+    Previously, the filter rejected all requests with bodies by checking for the presence of a decoding buffer, even when the body was complete.
 - area: router
   change: |
     Added :ref:`request_mirror_policies <envoy_v3_api_field_extensions.upstreams.http.v3.HttpProtocolOptions.request_mirror_policies>`

source/common/runtime/runtime_features.cc

@@ -60,6 +60,7 @@ RUNTIME_GUARD(envoy_reloadable_features_no_extension_lookup_by_name);
 RUNTIME_GUARD(envoy_reloadable_features_oauth2_cleanup_cookies);
 RUNTIME_GUARD(envoy_reloadable_features_oauth2_encrypt_tokens);
 RUNTIME_GUARD(envoy_reloadable_features_odcds_over_ads_fix);
+RUNTIME_GUARD(envoy_reloadable_features_on_demand_track_end_stream);
 RUNTIME_GUARD(envoy_reloadable_features_original_dst_rely_on_idle_timeout);
 RUNTIME_GUARD(envoy_reloadable_features_prefix_map_matcher_resume_after_subtree_miss);
 RUNTIME_GUARD(envoy_reloadable_features_quic_defer_logging_to_ack_listener);

source/extensions/filters/http/on_demand/on_demand_update.cc

@@ -245,9 +245,17 @@ void OnDemandRouteUpdate::onRouteConfigUpdateCompletion(bool route_exists) {
     return;
   }
 
-  if (route_exists &&           // route can be resolved after an on-demand
-                                // VHDS update
-      downstream_end_stream_ && // Redirects require fully read body.
+  bool can_recreate_stream = false;
+  if (Runtime::runtimeFeatureEnabled("envoy.reloadable_features.on_demand_track_end_stream")) {
+    // New behavior: track end_stream state to support stream recreation with fully read bodies.
+    can_recreate_stream = downstream_end_stream_;
+  } else {
+    // Old behavior: reject all requests with bodies.
+    can_recreate_stream = !callbacks_->decodingBuffer();
+  }
+  if (route_exists &&        // route can be resolved after an on-demand
+                             // VHDS update
+      can_recreate_stream && // Redirects require fully read body.
       callbacks_->recreateStream(/*headers=*/nullptr)) {
     return;
   }
@@ -261,8 +269,16 @@ void OnDemandRouteUpdate::onClusterDiscoveryCompletion(
     Upstream::ClusterDiscoveryStatus cluster_status) {
   filter_iteration_state_ = Http::FilterHeadersStatus::Continue;
   cluster_discovery_handle_.reset();
+  bool can_recreate_stream = false;
+  if (Runtime::runtimeFeatureEnabled("envoy.reloadable_features.on_demand_track_end_stream")) {
+    // New behavior: track end_stream state to support stream recreation with fully read bodies.
+    can_recreate_stream = downstream_end_stream_;
+  } else {
+    // Old behavior: reject all requests with bodies.
+    can_recreate_stream = !callbacks_->decodingBuffer();
+  }
   if (cluster_status == Upstream::ClusterDiscoveryStatus::Available &&
-      downstream_end_stream_) { // Redirects require fully read body.
+      can_recreate_stream) { // Redirects require fully read body.
     const Http::ResponseHeaderMap* headers = nullptr;
     if (callbacks_->recreateStream(headers)) {
       callbacks_->downstreamCallbacks()->clearRouteCache();