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

changelogs/current.yaml

@@ -6,6 +6,13 @@ behavior_changes:
   change: |
     Removed support for (long deprecated) opentracing.  See `issue 27401
     <https://github.com/envoyproxy/envoy/issues/27401>`_ for details.
+- 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: http
   change: |
     Added HTTP1-safe option for :ref:`max_connection_duration

docs/root/configuration/observability/access_log/usage.rst

@@ -428,6 +428,12 @@ The following command operators are supported:
     * ``us``: Microsecond precision.
     * ``ns``: Nanosecond precision.
 
+    NOTE: enabling independent half-close behavior for H/2 and H/3 protocols can produce
+    ``*_TX_END`` values lower than ``*_RX_END`` values, in cases where upstream peer has half-closed
+    its stream before downstream peer. In these cases ``COMMON_DURATION`` value will become negative.
+    Independent half-close behavior is enabled by setting the
+    ``envoy.reloadable_features.allow_multiplexed_upstream_half_close`` runtime value to ``true``.
+
   TCP/UDP
     Not implemented ("-").
 

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;
@@ -640,6 +640,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

@@ -466,19 +466,35 @@ void ActiveStreamDecoderFilter::encode1xxHeaders(ResponseHeaderMapPtr&& headers)
   }
 }
 
+void ActiveStreamDecoderFilter::maybeMarkDecoderFilterTerminal(bool encoded_end_stream) {
+  // If this filter encoded end_stream and the decoder filter chain has not yet been finished
+  // then make this filter terminal. Decoding will not go past this filter.
+  filter_encoded_end_stream_ = encoded_end_stream;
+
+  // If the filter that encoded end_stream has also already observed request (downstream)
+  // end_stream, but the decoder filter chain has not been completed, then decoding is aborted as
+  // there is no need to iterate over remaining decoder filters any more.
+  if (encoded_end_stream && end_stream_ && !parent_.state_.decoder_filter_chain_complete_) {
+    parent_.state_.decoder_filter_chain_aborted_ = true;
+  }
+}
+
 void ActiveStreamDecoderFilter::encodeHeaders(ResponseHeaderMapPtr&& headers, bool end_stream,
                                               absl::string_view details) {
+  maybeMarkDecoderFilterTerminal(end_stream);
   parent_.streamInfo().setResponseCodeDetails(details);
   parent_.filter_manager_callbacks_.setResponseHeaders(std::move(headers));
   parent_.encodeHeaders(nullptr, *parent_.filter_manager_callbacks_.responseHeaders(), end_stream);
 }
 
 void ActiveStreamDecoderFilter::encodeData(Buffer::Instance& data, bool end_stream) {
+  maybeMarkDecoderFilterTerminal(end_stream);
   parent_.encodeData(nullptr, data, end_stream,
                      FilterManager::FilterIterationStartState::CanStartFromCurrent);
 }
 
 void ActiveStreamDecoderFilter::encodeTrailers(ResponseTrailerMapPtr&& trailers) {
+  maybeMarkDecoderFilterTerminal(true);
   parent_.filter_manager_callbacks_.setResponseTrailers(std::move(trailers));
   parent_.encodeTrailers(nullptr, *parent_.filter_manager_callbacks_.responseTrailers());
 }
@@ -526,6 +542,8 @@ void FilterManager::decodeHeaders(ActiveStreamDecoderFilter* filter, RequestHead
   std::list<ActiveStreamDecoderFilterPtr>::iterator entry =
       commonDecodePrefix(filter, FilterIterationStartState::AlwaysStartFromNext);
   std::list<ActiveStreamDecoderFilterPtr>::iterator continue_data_entry = decoder_filters_.end();
+  // Terminal filter is either the last one or filter that encoded end_stream.
+  bool terminal_filter_decoded_end_stream = false;
 
   for (; entry != decoder_filters_.end(); entry++) {
     ASSERT(!(state_.filter_call_state_ & FilterCallState::DecodeHeaders));
@@ -599,13 +617,17 @@ void FilterManager::decodeHeaders(ActiveStreamDecoderFilter* filter, RequestHead
     if (end_stream && buffered_request_data_ && continue_data_entry == decoder_filters_.end()) {
       continue_data_entry = entry;
     }
+    terminal_filter_decoded_end_stream =
+        (end_stream && continue_data_entry == decoder_filters_.end()) &&
+        (std::next(entry) == decoder_filters_.end() || (*entry)->filter_encoded_end_stream_);
   }
 
   maybeContinueDecoding(continue_data_entry);
 
   if (end_stream) {
     disarmRequestTimeout();
   }
+  maybeEndDecode(terminal_filter_decoded_end_stream);
 }
 
 void FilterManager::decodeData(ActiveStreamDecoderFilter* filter, Buffer::Instance& data,
@@ -625,6 +647,7 @@ void FilterManager::decodeData(ActiveStreamDecoderFilter* filter, Buffer::Instan
   // Filter iteration may start at the current filter.
   std::list<ActiveStreamDecoderFilterPtr>::iterator entry =
       commonDecodePrefix(filter, filter_iteration_start_state);
+  bool terminal_filter_decoded_end_stream = false;
 
   for (; entry != decoder_filters_.end(); entry++) {
     // If the filter pointed by entry has stopped for all frame types, return now.
@@ -702,6 +725,10 @@ void FilterManager::decodeData(ActiveStreamDecoderFilter* filter, Buffer::Instan
       trailers_added_entry = entry;
     }
 
+    terminal_filter_decoded_end_stream =
+        end_stream &&
+        (std::next(entry) == decoder_filters_.end() || (*entry)->filter_encoded_end_stream_);
+
     if (!(*entry)->commonHandleAfterDataCallback(status, data, state_.decoder_filters_streaming_) &&
         std::next(entry) != decoder_filters_.end()) {
       // Stop iteration IFF this is not the last filter. If it is the last filter, continue with
@@ -720,6 +747,7 @@ void FilterManager::decodeData(ActiveStreamDecoderFilter* filter, Buffer::Instan
   if (end_stream) {
     disarmRequestTimeout();
   }
+  maybeEndDecode(terminal_filter_decoded_end_stream);
 }
 
 RequestTrailerMap& FilterManager::addDecodedTrailers() {
@@ -764,6 +792,7 @@ void FilterManager::decodeTrailers(ActiveStreamDecoderFilter* filter, RequestTra
   // Filter iteration may start at the current filter.
   std::list<ActiveStreamDecoderFilterPtr>::iterator entry =
       commonDecodePrefix(filter, FilterIterationStartState::CanStartFromCurrent);
+  bool terminal_filter_decoded_end_stream = false;
 
   for (; entry != decoder_filters_.end(); entry++) {
     // If the filter pointed by entry has stopped for all frame type, return now.
@@ -787,12 +816,20 @@ void FilterManager::decodeTrailers(ActiveStreamDecoderFilter* filter, RequestTra
     }
 
     processNewlyAddedMetadata();
+    terminal_filter_decoded_end_stream =
+        std::next(entry) == decoder_filters_.end() || (*entry)->filter_encoded_end_stream_;
 
-    if (!(*entry)->commonHandleAfterTrailersCallback(status)) {
+    if (terminal_filter_decoded_end_stream) {
+      break;
+    }
+
+    if (!(*entry)->commonHandleAfterTrailersCallback(status) &&
+        std::next(entry) != decoder_filters_.end()) {
       return;
     }
   }
   disarmRequestTimeout();
+  maybeEndDecode(terminal_filter_decoded_end_stream);
 }
 
 void FilterManager::decodeMetadata(ActiveStreamDecoderFilter* filter, MetadataMap& metadata_map) {
@@ -922,9 +959,9 @@ void DownstreamFilterManager::sendLocalReply(
 
   // Stop filter chain iteration if local reply was sent while filter decoding or encoding callbacks
   // are running.
-  if (state_.filter_call_state_ & FilterCallState::IsDecodingMask) {
-    state_.decoder_filter_chain_aborted_ = true;
-  } else if (state_.filter_call_state_ & FilterCallState::IsEncodingMask) {
+  // Local reply always stops decoder filter chain.
+  state_.decoder_filter_chain_aborted_ = true;
+  if (state_.filter_call_state_ & FilterCallState::IsEncodingMask) {
     state_.encoder_filter_chain_aborted_ = true;
   }
 
@@ -1448,6 +1485,56 @@ void FilterManager::maybeEndEncode(bool end_stream) {
   if (end_stream) {
     ASSERT(!state_.encoder_filter_chain_complete_);
     state_.encoder_filter_chain_complete_ = true;
+    if (filter_manager_callbacks_.isHalfCloseEnabled()) {
+      // If independent half close is enabled the stream is closed when both decoder and encoder
+      // filter chains has completed or were aborted.
+      checkAndCloseStreamIfFullyClosed();
+    } else {
+      // Otherwise encoding end_stream always closes the stream (and resets it if request was not
+      // complete yet).
+      filter_manager_callbacks_.endStream();
+    }
+  }
+}
+
+void FilterManager::maybeEndDecode(bool terminal_filter_decoded_end_stream) {
+  if (terminal_filter_decoded_end_stream) {
+    ASSERT(!state_.decoder_filter_chain_complete_);
+    state_.decoder_filter_chain_complete_ = true;
+    // If the decoder filter chain was aborted (i.e. due to local reply)
+    // we rely on the encoding of end_stream to close the stream.
+    if (filter_manager_callbacks_.isHalfCloseEnabled() && !stopDecoderFilterChain()) {
+      checkAndCloseStreamIfFullyClosed();
+    }
+  }
+}
+
+void FilterManager::checkAndCloseStreamIfFullyClosed() {
+  // This function is only used when half close semantics are enabled.
+  if (!filter_manager_callbacks_.isHalfCloseEnabled()) {
+    return;
+  }
+  // When the independent half close is enabled the stream is always closed on error responses
+  // from the server.
+  if (filter_manager_callbacks_.responseHeaders().has_value()) {
+    const uint64_t response_status =
+        Http::Utility::getResponseStatus(filter_manager_callbacks_.responseHeaders().ref());
+    bool error_response =
+        !(Http::CodeUtility::is2xx(response_status) || Http::CodeUtility::is1xx(response_status));
+    // Abort the decoder filter if it has not yet been completed.
+    if (error_response && !state_.decoder_filter_chain_complete_) {
+      state_.decoder_filter_chain_aborted_ = true;
+    }
+  }
+
+  // If independent half close is enabled then close the stream when
+  // 1. Both encoder and decoder filter chains has completed.
+  // 2. Encoder filter chain has completed and decoder filter chain was aborted (i.e. local reply).
+  // There is no need to check for aborted encoder filter chain as the filter will either be
+  // completed or stream is reset.
+  if (state_.encoder_filter_chain_complete_ &&
+      (state_.decoder_filter_chain_complete_ || state_.decoder_filter_chain_aborted_)) {
+    ENVOY_STREAM_LOG(trace, "closing stream", *this);
     filter_manager_callbacks_.endStream();
   }
 }

source/common/http/filter_manager.h

@@ -280,9 +280,20 @@ struct ActiveStreamDecoderFilter : public ActiveStreamFilterBase,
 
   void requestDataTooLarge();
   void requestDataDrained();
+  // Check if the filter that encoded end_stream has also decoded end_stream and if true
+  // stop the decoder filter chain. This will end the request after encoder filter chain
+  // is completed.
+  // This allows non-terminal filters (i.e. cache filter) to encode responses when independent
+  // half-close is enabled. Encoding end_stream effectively makes the filter terminal - decoder
+  // filer chain will not go past this filter.
+  void maybeMarkDecoderFilterTerminal(bool encoded_end_stream);
 
   StreamDecoderFilterSharedPtr handle_;
   bool is_grpc_request_{};
+  // Indicates that this filter called an encodeXXX method with end_stream == true.
+  // When independent half close is enabled this filter becomes the terminal filter
+  // in the decoder filter chain.
+  bool filter_encoded_end_stream_{false};
 };
 
 using ActiveStreamDecoderFilterPtr = std::unique_ptr<ActiveStreamDecoderFilter>;
@@ -656,7 +667,11 @@ 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_.decoder_filter_chain_complete_)
+       << DUMP_MEMBER(state_.encoder_filter_chain_complete_)
+       << DUMP_MEMBER(state_.observed_decode_end_stream_)
+       << DUMP_MEMBER(state_.observed_encode_end_stream_) << "\n";
 
     DUMP_DETAILS(filter_manager_callbacks_.requestHeaders());
     DUMP_DETAILS(filter_manager_callbacks_.requestTrailers());
@@ -782,6 +797,15 @@ class FilterManager : public ScopeTrackedObject,
    */
   void maybeEndEncode(bool end_stream);
 
+  /**
+   * If terminal_filter_decoded_end_stream is true, marks decoding as complete. This is a noop if
+   * terminal_filter_decoded_end_stream is false.
+   * @param end_stream whether decoding is complete.
+   */
+  void maybeEndDecode(bool terminal_filter_decoded_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,
@@ -868,16 +892,22 @@ class FilterManager : public ScopeTrackedObject,
 protected:
   struct State {
     State()
-        : encoder_filter_chain_complete_(false), observed_decode_end_stream_(false),
-          observed_encode_end_stream_(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) {}
+        : decoder_filter_chain_complete_(false), encoder_filter_chain_complete_(false),
+          observed_decode_end_stream_(false), observed_encode_end_stream_(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) {}
     uint32_t filter_call_state_{0};
 
+    // Set after decoder filter chain has completed iteration. Prevents further calls to decoder
+    // filters. This flag is used to determine stream completion when the independent half-close is
+    // enabled.
+    bool decoder_filter_chain_complete_ : 1;
+
     // Set after encoder filter chain has completed iteration. Prevents further calls to encoder
-    // filters.
+    // filters. This flag is used to determine stream completion when the independent half-close is
+    // enabled.
     bool encoder_filter_chain_complete_ : 1;
 
     // Set `true` when the filter manager observes end stream on the decoder path (from downstream