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

changelogs/current.yaml

@@ -32,6 +32,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.
 - area: golang
   change: |
     Move ``Continue``, ``SendLocalReply`` and ``RecoverPanic` from ``FilterCallbackHandler`` to ``DecoderFilterCallbacks`` and

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

@@ -149,13 +149,14 @@ 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.
+  //
+  // 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.cc

@@ -126,8 +126,10 @@ ConnectionManagerImpl::ConnectionManagerImpl(ConnectionManagerConfigSharedPtr co
                                      /*node_id=*/local_info_.node().id(),
                                      /*server_name=*/config_->serverName(),
                                      /*proxy_status_config=*/config_->proxyStatusConfig())),
-      max_requests_during_dispatch_(runtime_.snapshot().getInteger(
-          ConnectionManagerImpl::MaxRequestsPerIoCycle, UINT32_MAX)) {
+      max_requests_during_dispatch_(
+          runtime_.snapshot().getInteger(ConnectionManagerImpl::MaxRequestsPerIoCycle, UINT32_MAX)),
+      allow_upstream_half_close_(Runtime::runtimeFeatureEnabled(
+          "envoy.reloadable_features.allow_multiplexed_upstream_half_close")) {
   ENVOY_LOG_ONCE_IF(
       trace, accept_new_http_stream_ == nullptr,
       "LoadShedPoint envoy.load_shed_points.http_connection_manager_decode_headers is not "

source/common/http/conn_manager_impl.h

@@ -283,7 +283,7 @@ 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 connection_manager_.allow_upstream_half_close_; }
 
     // DownstreamStreamFilterCallbacks
     void setRoute(Router::RouteConstSharedPtr route) override;
@@ -638,6 +638,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
   uint32_t requests_during_dispatch_count_{0};
   const uint32_t max_requests_during_dispatch_{UINT32_MAX};
   Event::SchedulableCallbackPtr deferred_request_processing_callback_;
+  const bool allow_upstream_half_close_{};
 };
 
 } // namespace Http

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

@@ -901,16 +901,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_.should_stop_decoding_));
   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 (filter_manager_callbacks_.isHalfCloseEnabled()) {
+      if (end_stream) {
+        state_.encoder_end_stream_ = true;
+        if (state_.should_stop_decoding_) {
+          ASSERT(!state_.local_complete_);
+          state_.local_complete_ = true;
+        }
+      }
+    } else {
+      if (end_stream) {
+        ASSERT(!state_.local_complete_);
+        state_.local_complete_ = true;
+      }
     }
     return encoder_filters_.begin();
   }
@@ -958,6 +970,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 stops decoding of downstream request.
+  stopDecoding();
 
   // Stop filter chain iteration if local reply was sent while filter decoding or encoding callbacks
   // are running.
@@ -1282,6 +1296,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 (filter_manager_callbacks_.isHalfCloseEnabled()) {
+    const uint64_t response_status = Http::Utility::getResponseStatus(headers);
+    if (!(Http::CodeUtility::is2xx(response_status) || Http::CodeUtility::is1xx(response_status))) {
+      // When the upstream half close is enabled the stream decoding is stopped on error responses
+      // from the server.
+      stopDecoding();
+    }
+  }
   filter_manager_callbacks_.encodeHeaders(headers, modified_end_stream);
   if (state_.saw_downstream_reset_) {
     return;
@@ -1487,6 +1509,31 @@ void FilterManager::maybeEndEncode(bool end_stream) {
   if (end_stream) {
     ASSERT(!state_.remote_encode_complete_);
     state_.remote_encode_complete_ = true;
+    if (filter_manager_callbacks_.isHalfCloseEnabled()) {
+      checkAndCloseStreamIfFullyClosed();
+    } else {
+      state_.stream_closed_ = true;
+      filter_manager_callbacks_.endStream();
+    }
+  }
+}
+
+void FilterManager::checkAndCloseStreamIfFullyClosed() {
+  // This function is only used when half close semantics are enabled.
+  if (!filter_manager_callbacks_.isHalfCloseEnabled()) {
+    return;
+  }
+
+  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_.should_stop_decoding_)) {
+    state_.stream_closed_ = true;
+    ENVOY_STREAM_LOG(trace, "closing stream", *this);
     filter_manager_callbacks_.endStream();
   }
 }
@@ -1669,7 +1716,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_.filter_manager_callbacks_.isHalfCloseEnabled() ? 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

@@ -657,7 +657,10 @@ class FilterManager : public ScopeTrackedObject,
   // ScopeTrackedObject
   void dumpState(std::ostream& os, int indent_level = 0) const override {
     const char* spaces = spacesForLevel(indent_level);
-    os << spaces << "FilterManager " << this << DUMP_MEMBER(state_.has_1xx_headers_) << "\n";
+    os << spaces << "FilterManager " << this << DUMP_MEMBER(state_.has_1xx_headers_)
+       << DUMP_MEMBER(state_.remote_decode_complete_) << DUMP_MEMBER(state_.remote_encode_complete_)
+       << DUMP_MEMBER(state_.encoder_end_stream_) << DUMP_MEMBER(state_.stream_closed_)
+       << DUMP_MEMBER(state_.should_stop_decoding_) << "\n";
 
     DUMP_DETAILS(filter_manager_callbacks_.requestHeaders());
     DUMP_DETAILS(filter_manager_callbacks_.requestTrailers());
@@ -748,6 +751,7 @@ class FilterManager : public ScopeTrackedObject,
   void decodeHeaders(RequestHeaderMap& headers, bool end_stream) {
     state_.remote_decode_complete_ = end_stream;
     decodeHeaders(nullptr, headers, end_stream);
+    checkAndCloseStreamIfFullyClosed();
   }
 
   /**
@@ -758,6 +762,7 @@ 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);
+    checkAndCloseStreamIfFullyClosed();
   }
 
   /**
@@ -767,6 +772,7 @@ class FilterManager : public ScopeTrackedObject,
   void decodeTrailers(RequestTrailerMap& trailers) {
     state_.remote_decode_complete_ = true;
     decodeTrailers(nullptr, trailers);
+    checkAndCloseStreamIfFullyClosed();
   }
 
   /**
@@ -783,6 +789,8 @@ class FilterManager : public ScopeTrackedObject,
    */
   void maybeEndEncode(bool end_stream);
 
+  void checkAndCloseStreamIfFullyClosed();
+
   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,
@@ -862,21 +870,36 @@ class FilterManager : public ScopeTrackedObject,
 
   virtual bool shouldLoadShed() { return false; };
 
+  void stopDecoding() { state_.should_stop_decoding_ = true; }
+
 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), should_stop_decoding_(false) {}
     uint32_t filter_call_state_{0};
 
-    bool remote_decode_complete_ : 1;
-    bool remote_encode_complete_ : 1;
-    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.
+    bool remote_decode_complete_ : 1; // Set when decoder filter chain iteration has completed. All
+                                      // decoder filters have completed and end_stream was observed.
+    bool remote_encode_complete_ : 1; // Set when encoder filter chain iteration has completed. All
+                                      // decoder filters have completed and end_stream was observed.
+    // TODO(yanavlasov): always use encoder_end_stream_ for tracking end_stream state during encoder
+    // filter iteration.
+    bool encoder_end_stream_ : 1; // This is set the first time the end_stream is observed during
+                                  // encoder filter chain iteration and used only to set the
+                                  // end_stream flag when resuming encoder filter chain iteration.
+                                  // It is only used when the
+                                  // `allow_multiplexed_upstream_half_close` runtime flag is true.
+    bool local_complete_ : 1; // This indicates that request is complete prior to filter processing.
+                              // This flag terminates decoder filter chain iteration.
+                              // It is also used to track the `end_stream` state during encoder
+                              // filter chain iteration when `allow_multiplexed_upstream_half_close`
+                              // runtime flag is false. A filter can still stop the stream from
+                              // being complete as seen by the codec.
     // By default, we will assume there are no 1xx. If encode1xxHeaders
     // is ever called, this is set to true so commonContinue resumes processing the 1xx.
     bool has_1xx_headers_ : 1;
@@ -893,6 +916,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 should_stop_decoding_ : 1; // Set to indicate that decoding on the stream should be stopped
+                                    // 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

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 "";