Allow multiplexed upstream servers to half close the stream before the downstream

changelogs/current.yaml

@@ -26,6 +26,13 @@ behavior_changes:
     :ref:`TlvsMetadata type <envoy_v3_api_msg_data.core.v3.TlvsMetadata>`.
     This change can be temporarily disabled by setting the runtime flag
     ``envoy.reloadable_features.use_typed_metadata_in_proxy_protocol_listener`` to ``false``.
+- area: http
+  change: |
+    Allow HTTP/2 (and HTTP/3) upstream servers to half close the stream before the downstream. This enables bidirectional
+    gRPC streams where server completes streaming before the client. Behavior of HTTP/1 or TCP proxy upstream servers is
+    unchanged and the stream is reset if the upstream server completes response before the downstream. The stream is also
+    reset if the upstream server responds with an error status before the downstream. This behavior is disabled by default
+    and can be enabled by setting the ``envoy.reloadable_features.allow_multiplexed_upstream_half_close`` runtime key to true.
 
 minor_behavior_changes:
 # *Changes that may cause incompatibilities for some users, but should not for most*

envoy/http/stream_reset_handler.h

@@ -36,7 +36,9 @@ enum class StreamResetReason {
   // Received payload did not conform to HTTP protocol.
   ProtocolError,
   // If the stream was locally reset by the Overload Manager.
-  OverloadManager
+  OverloadManager,
+  // If stream was locally reset due to HTTP/1 upstream half closing before downstream.
+  Http1PrematureUpstreamHalfClose,
 };
 
 /**

source/common/http/async_client_impl.cc

@@ -145,13 +145,16 @@ void AsyncStreamImpl::encodeHeaders(ResponseHeaderMapPtr&& headers, bool end_str
   encoded_response_headers_ = true;
   stream_callbacks_.onHeaders(std::move(headers), end_stream);
   closeRemote(end_stream);
-  // At present, the router cleans up stream state as soon as the remote is closed, making a
-  // half-open local stream unsupported and dangerous. Ensure we close locally to trigger completion
-  // and keep things consistent. Another option would be to issue a stream reset here if local isn't
-  // yet closed, triggering cleanup along a more standardized path. However, this would require
-  // additional logic to handle the response completion and subsequent reset, and run the risk of
-  // being interpreted as a failure, when in fact no error has necessarily occurred. Gracefully
-  // closing seems most in-line with behavior elsewhere in Envoy for now.
+  // At present, the AsyncStream is always fully closed when the server half closes the stream.
+  // This is the case even when allow_multiplexed_upstream_half_close runtime flag is set, as there
+  // are currently no known use cases where early server half close needs to be supported.
+  //
+  // Always ensure we close locally to trigger completion. Another option would be to issue a stream
+  // reset here if local isn't yet closed, triggering cleanup along a more standardized path.
+  // However, this would require additional logic to handle the response completion and subsequent
+  // reset, and run the risk of being interpreted as a failure, when in fact no error has
+  // necessarily occurred. Gracefully closing seems most in-line with behavior elsewhere in Envoy
+  // for now.
   closeLocal(end_stream);
 }
 

source/common/http/conn_manager_impl.h

@@ -283,7 +283,9 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     OptRef<const Tracing::Config> tracingConfig() const override;
     const ScopeTrackedObject& scope() override;
     OptRef<DownstreamStreamFilterCallbacks> downstreamCallbacks() override { return *this; }
-    bool isHalfCloseEnabled() override { return false; }
+    bool isHalfCloseEnabled() override {
+      return filter_manager_.allowUpstreamHalfClose() ? true : false;
+    }
 
     // DownstreamStreamFilterCallbacks
     void setRoute(Router::RouteConstSharedPtr route) override;

source/common/http/conn_pool_base.cc

@@ -177,6 +177,7 @@ void MultiplexedActiveClientBase::onStreamReset(Http::StreamResetReason reason)
   case StreamResetReason::RemoteRefusedStreamReset:
   case StreamResetReason::Overflow:
   case StreamResetReason::ConnectError:
+  case StreamResetReason::Http1PrematureUpstreamHalfClose:
     break;
   }
 }

source/common/http/filter_manager.cc

@@ -903,16 +903,28 @@ FilterManager::commonEncodePrefix(ActiveStreamEncoderFilter* filter, bool end_st
                                   FilterIterationStartState filter_iteration_start_state) {
   // Only do base state setting on the initial call. Subsequent calls for filtering do not touch
   // the base state.
-  ENVOY_STREAM_LOG(trace, "commonEncodePrefix end_stream: {}, isHalfCloseEnabled: {}", *this,
-                   end_stream, filter_manager_callbacks_.isHalfCloseEnabled());
+  ENVOY_STREAM_LOG(trace,
+                   "commonEncodePrefix end_stream: {}, isHalfCloseEnabled: {}, force_close: {}",
+                   *this, end_stream, filter_manager_callbacks_.isHalfCloseEnabled(),
+                   static_cast<bool>(state_.force_close_stream_));
   if (filter == nullptr) {
     // half close is enabled in case tcp proxying is done with http1 encoder. In this case, we
     // should not set the local_complete_ flag to true when end_stream is true.
     // setting local_complete_ to true will cause any data sent in the upstream direction to be
     // dropped.
-    if (end_stream && !filter_manager_callbacks_.isHalfCloseEnabled()) {
-      ASSERT(!state_.local_complete_);
-      state_.local_complete_ = true;
+    if (allow_upstream_half_close_) {
+      if (end_stream) {
+        state_.encoder_end_stream_ = true;
+        if (!filter_manager_callbacks_.isHalfCloseEnabled() || state_.force_close_stream_) {
+          ASSERT(!state_.local_complete_);
+          state_.local_complete_ = true;
+        }
+      }
+    } else {
+      if (end_stream && !filter_manager_callbacks_.isHalfCloseEnabled()) {
+        ASSERT(!state_.local_complete_);
+        state_.local_complete_ = true;
+      }
     }
     return encoder_filters_.begin();
   }
@@ -960,6 +972,8 @@ void DownstreamFilterManager::sendLocalReply(
   ASSERT(!state_.under_on_local_reply_);
   const bool is_head_request = state_.is_head_request_;
   const bool is_grpc_request = state_.is_grpc_request_;
+  // Local reply closes the stream even if downstream is not half closed.
+  state_.force_close_stream_ = true;
 
   // Stop filter chain iteration if local reply was sent while filter decoding or encoding callbacks
   // are running.
@@ -1284,6 +1298,14 @@ void FilterManager::encodeHeaders(ActiveStreamEncoderFilter* filter, ResponseHea
 
   const bool modified_end_stream = (end_stream && continue_data_entry == encoder_filters_.end());
   state_.non_100_response_headers_encoded_ = true;
+  if (allow_upstream_half_close_) {
+    const uint64_t response_status = Http::Utility::getResponseStatus(headers);
+    if (!(Http::CodeUtility::is2xx(response_status) || Http::CodeUtility::is1xx(response_status))) {
+      // Even if upstream half close is enabled the stream is closed on error responses from the
+      // server.
+      state_.force_close_stream_ = true;
+    }
+  }
   filter_manager_callbacks_.encodeHeaders(headers, modified_end_stream);
   if (state_.saw_downstream_reset_) {
     return;
@@ -1489,6 +1511,27 @@ void FilterManager::maybeEndEncode(bool end_stream) {
   if (end_stream) {
     ASSERT(!state_.remote_encode_complete_);
     state_.remote_encode_complete_ = true;
+    if (allow_upstream_half_close_) {
+      maybeCloseStream();
+    } else {
+      state_.stream_closed_ = true;
+      filter_manager_callbacks_.endStream();
+    }
+  }
+}
+
+void FilterManager::maybeCloseStream() {
+  ASSERT(allow_upstream_half_close_);
+  if (state_.stream_closed_) {
+    return;
+  }
+
+  // If upstream half close is enabled then close the stream either when force close
+  // is set (i.e local reply) or when both server and client half closed.
+  if (state_.remote_encode_complete_ &&
+      (state_.remote_decode_complete_ || state_.force_close_stream_)) {
+    state_.stream_closed_ = true;
+    ENVOY_STREAM_LOG(trace, "closing stream", *this);
     filter_manager_callbacks_.endStream();
   }
 }
@@ -1671,7 +1714,14 @@ Buffer::InstancePtr ActiveStreamEncoderFilter::createBuffer() {
 Buffer::InstancePtr& ActiveStreamEncoderFilter::bufferedData() {
   return parent_.buffered_response_data_;
 }
-bool ActiveStreamEncoderFilter::complete() { return parent_.state_.local_complete_; }
+bool ActiveStreamEncoderFilter::complete() {
+  // This is used for determining end_stream flag when iterating encoder filter chain.
+  // When the upstream half close is enabled the local complete may not be set when
+  // end stream is observed on the encode path in case upstream half closes before the
+  // downstream.
+  return parent_.allow_upstream_half_close_ ? parent_.state_.encoder_end_stream_
+                                            : parent_.state_.local_complete_;
+}
 bool ActiveStreamEncoderFilter::has1xxHeaders() {
   return parent_.state_.has_1xx_headers_ && !continued_1xx_headers_;
 }

source/common/http/filter_manager.h

@@ -648,7 +648,9 @@ class FilterManager : public ScopeTrackedObject,
         proxy_100_continue_(proxy_100_continue), buffer_limit_(buffer_limit),
         filter_chain_factory_(filter_chain_factory),
         no_downgrade_to_canonical_name_(Runtime::runtimeFeatureEnabled(
-            "envoy.reloadable_features.no_downgrade_to_canonical_name")) {}
+            "envoy.reloadable_features.no_downgrade_to_canonical_name")),
+        allow_upstream_half_close_(Runtime::runtimeFeatureEnabled(
+            "envoy.reloadable_features.allow_multiplexed_upstream_half_close")) {}
   ~FilterManager() override {
     ASSERT(state_.destroyed_);
     ASSERT(state_.filter_call_state_ == 0);
@@ -748,6 +750,9 @@ class FilterManager : public ScopeTrackedObject,
   void decodeHeaders(RequestHeaderMap& headers, bool end_stream) {
     state_.remote_decode_complete_ = end_stream;
     decodeHeaders(nullptr, headers, end_stream);
+    if (allow_upstream_half_close_) {
+      maybeCloseStream();
+    }
   }
 
   /**
@@ -758,6 +763,9 @@ class FilterManager : public ScopeTrackedObject,
   void decodeData(Buffer::Instance& data, bool end_stream) {
     state_.remote_decode_complete_ = end_stream;
     decodeData(nullptr, data, end_stream, FilterIterationStartState::CanStartFromCurrent);
+    if (allow_upstream_half_close_) {
+      maybeCloseStream();
+    }
   }
 
   /**
@@ -767,6 +775,9 @@ class FilterManager : public ScopeTrackedObject,
   void decodeTrailers(RequestTrailerMap& trailers) {
     state_.remote_decode_complete_ = true;
     decodeTrailers(nullptr, trailers);
+    if (allow_upstream_half_close_) {
+      maybeCloseStream();
+    }
   }
 
   /**
@@ -783,6 +794,8 @@ class FilterManager : public ScopeTrackedObject,
    */
   void maybeEndEncode(bool end_stream);
 
+  void maybeCloseStream();
+
   virtual void sendLocalReply(Code code, absl::string_view body,
                               const std::function<void(ResponseHeaderMap& headers)>& modify_headers,
                               const absl::optional<Grpc::Status::GrpcStatus> grpc_status,
@@ -861,19 +874,24 @@ class FilterManager : public ScopeTrackedObject,
   bool sawDownstreamReset() { return state_.saw_downstream_reset_; }
 
   virtual bool shouldLoadShed() { return false; };
+  bool allowUpstreamHalfClose() const { return allow_upstream_half_close_; }
 
 protected:
   struct State {
     State()
-        : remote_decode_complete_(false), remote_encode_complete_(false), local_complete_(false),
-          has_1xx_headers_(false), created_filter_chain_(false), is_head_request_(false),
-          is_grpc_request_(false), non_100_response_headers_encoded_(false),
-          under_on_local_reply_(false), decoder_filter_chain_aborted_(false),
-          encoder_filter_chain_aborted_(false), saw_downstream_reset_(false) {}
+        : remote_decode_complete_(false), remote_encode_complete_(false),
+          encoder_end_stream_(false), local_complete_(false), has_1xx_headers_(false),
+          created_filter_chain_(false), is_head_request_(false), is_grpc_request_(false),
+          non_100_response_headers_encoded_(false), under_on_local_reply_(false),
+          decoder_filter_chain_aborted_(false), encoder_filter_chain_aborted_(false),
+          saw_downstream_reset_(false), stream_closed_(false), force_close_stream_(false) {}
     uint32_t filter_call_state_{0};
 
-    bool remote_decode_complete_ : 1;
-    bool remote_encode_complete_ : 1;
+    bool remote_decode_complete_ : 1; // Set when decoder filter chain iteration has completed.
+    bool remote_encode_complete_ : 1; // Set when encoder filter chain iteration has completed.
+    bool encoder_end_stream_ : 1; // This is set the first time the end_stream is observed during
+                                  // encoder filter chain iteration and used to set the end_stream
+                                  // flag when resuming encoder filter chain iteration.
     bool local_complete_ : 1; // This indicates that local is complete prior to filter processing.
                               // A filter can still stop the stream from being complete as seen
                               // by the codec.
@@ -893,6 +911,11 @@ class FilterManager : public ScopeTrackedObject,
     bool decoder_filter_chain_aborted_ : 1;
     bool encoder_filter_chain_aborted_ : 1;
     bool saw_downstream_reset_ : 1;
+    bool stream_closed_ : 1; // Set when both remote_decode_complete_ and remote_encode_complete_ is
+                             // true observed for the first time and prevents ending the stream
+                             // multiple times. Only set when allow_upstream_half_close is enabled.
+    bool force_close_stream_ : 1; // Set to indicate that stream should be closed due to either
+                                  // local reply or error response from the server.
 
     // The following 3 members are booleans rather than part of the space-saving bitfield as they
     // are passed as arguments to functions expecting bools. Extend State using the bitfield
@@ -1086,6 +1109,7 @@ class FilterManager : public ScopeTrackedObject,
   State state_;
 
   const bool no_downgrade_to_canonical_name_{};
+  const bool allow_upstream_half_close_{};
 };
 
 // The DownstreamFilterManager has explicit handling to send local replies.

source/common/http/http1/codec_impl.cc

@@ -1436,7 +1436,9 @@ ClientConnectionImpl::ClientConnectionImpl(Network::Connection& connection, Code
           [&]() -> void { this->onBelowLowWatermark(); },
           [&]() -> void { this->onAboveHighWatermark(); },
           []() -> void { /* TODO(adisuissa): handle overflow watermark */ })),
-      passing_through_proxy_(passing_through_proxy) {
+      passing_through_proxy_(passing_through_proxy),
+      force_reset_on_premature_upstream_half_close_(Runtime::runtimeFeatureEnabled(
+          "envoy.reloadable_features.allow_multiplexed_upstream_half_close")) {
   owned_output_buffer_->setWatermarks(connection.bufferLimit());
   // Inform parent
   output_buffer_ = owned_output_buffer_.get();
@@ -1587,6 +1589,15 @@ CallbackResult ClientConnectionImpl::onMessageCompleteBase() {
       response.decoder_->decodeData(buffer, true);
     }
 
+    if (force_reset_on_premature_upstream_half_close_) {
+      // H/1 connections are always reset if upstream is done before downstream.
+      // When the allow_multiplexed_upstream_half_close is enabled the router filter does not
+      // reset streams where upstream half closed before downstream. In this case the H/1 codec
+      // has to reset the stream.
+      ENVOY_CONN_LOG(trace, "Resetting stream due to premature H/1 upstream close.", connection_);
+      response.encoder_.runResetCallbacks(StreamResetReason::Http1PrematureUpstreamHalfClose);
+    }
+
     // Reset to ensure no information from one requests persists to the next.
     pending_response_.reset();
     headers_or_trailers_.emplace<ResponseHeaderMapPtr>(nullptr);

source/common/http/http1/codec_impl.h

@@ -686,6 +686,8 @@ class ClientConnectionImpl : public ClientConnection, public ConnectionImpl {
   // True if the upstream connection is pointed at an HTTP/1.1 proxy, and
   // plaintext HTTP should be sent with fully qualified URLs.
   bool passing_through_proxy_ = false;
+
+  const bool force_reset_on_premature_upstream_half_close_{};
 };
 
 } // namespace Http1

source/common/http/utility.cc

@@ -1060,6 +1060,8 @@ const std::string Utility::resetReasonToString(const Http::StreamResetReason res
     return "protocol error";
   case Http::StreamResetReason::OverloadManager:
     return "overload manager reset";
+  case Http::StreamResetReason::Http1PrematureUpstreamHalfClose:
+    return "HTTP/1 premature upstream half close";
   }
 
   return "";

source/common/router/router.cc

@@ -751,9 +751,9 @@ Http::FilterHeadersStatus Filter::decodeHeaders(Http::RequestHeaderMap& headers,
   // will never transition from false to true.
   bool can_use_http3 =
       !transport_socket_options_ || !transport_socket_options_->http11ProxyInfo().has_value();
-  UpstreamRequestPtr upstream_request =
-      std::make_unique<UpstreamRequest>(*this, std::move(generic_conn_pool), can_send_early_data,
-                                        can_use_http3, false /*enable_half_close*/);
+  UpstreamRequestPtr upstream_request = std::make_unique<UpstreamRequest>(
+      *this, std::move(generic_conn_pool), can_send_early_data, can_use_http3,
+      allow_multiplexed_upstream_half_close_ /*enable_half_close*/);
   LinkedList::moveIntoList(std::move(upstream_request), upstream_requests_);
   upstream_requests_.front()->acceptHeadersFromRouter(end_stream);
   if (streaming_shadows_) {
@@ -1441,6 +1441,7 @@ Filter::streamResetReasonToResponseFlag(Http::StreamResetReason reset_reason) {
     return StreamInfo::CoreResponseFlag::UpstreamConnectionTermination;
   case Http::StreamResetReason::LocalReset:
   case Http::StreamResetReason::LocalRefusedStreamReset:
+  case Http::StreamResetReason::Http1PrematureUpstreamHalfClose:
     return StreamInfo::CoreResponseFlag::LocalReset;
   case Http::StreamResetReason::Overflow:
     return StreamInfo::CoreResponseFlag::UpstreamOverflow;
@@ -1722,6 +1723,10 @@ void Filter::onUpstreamMetadata(Http::MetadataMapPtr&& metadata_map) {
 
 void Filter::onUpstreamComplete(UpstreamRequest& upstream_request) {
   if (!downstream_end_stream_) {
+    if (allow_multiplexed_upstream_half_close_) {
+      // Continue request if downstream is not done yet.
+      return;
+    }
     upstream_request.resetStream();
   }
   Event::Dispatcher& dispatcher = callbacks_->dispatcher();
@@ -1994,9 +1999,9 @@ void Filter::doRetry(bool can_send_early_data, bool can_use_http3, TimeoutRetry
     cleanup();
     return;
   }
-  UpstreamRequestPtr upstream_request =
-      std::make_unique<UpstreamRequest>(*this, std::move(generic_conn_pool), can_send_early_data,
-                                        can_use_http3, false /*enable_tcp_tunneling*/);
+  UpstreamRequestPtr upstream_request = std::make_unique<UpstreamRequest>(
+      *this, std::move(generic_conn_pool), can_send_early_data, can_use_http3,
+      allow_multiplexed_upstream_half_close_ /*enable_half_close*/);
 
   if (include_attempt_count_in_request_) {
     downstream_headers_->setEnvoyAttemptCount(attempt_count_);

source/common/router/router.h

@@ -304,7 +304,9 @@ class Filter : Logger::Loggable<Logger::Id::router>,
       : config_(config), stats_(stats), grpc_request_(false), exclude_http_code_stats_(false),
         downstream_response_started_(false), downstream_end_stream_(false), is_retry_(false),
         request_buffer_overflowed_(false), streaming_shadows_(Runtime::runtimeFeatureEnabled(
-                                               "envoy.reloadable_features.streaming_shadow")) {}
+                                               "envoy.reloadable_features.streaming_shadow")),
+        allow_multiplexed_upstream_half_close_(Runtime::runtimeFeatureEnabled(
+            "envoy.reloadable_features.allow_multiplexed_upstream_half_close")) {}
 
   ~Filter() override;
 
@@ -605,6 +607,7 @@ class Filter : Logger::Loggable<Logger::Id::router>,
   bool include_timeout_retry_header_in_request_ : 1;
   bool request_buffer_overflowed_ : 1;
   const bool streaming_shadows_ : 1;
+  const bool allow_multiplexed_upstream_half_close_ : 1;
 };
 
 class ProdFilter : public Filter {