upstream: avoid reset after end_stream in TCP HTTP upstream (#14106)

Signed-off-by: Snow Pettersen <snowp@lyft.com>

include/envoy/router/router.h

@@ -14,6 +14,7 @@
 #include "envoy/config/core/v3/base.pb.h"
 #include "envoy/config/route/v3/route_components.pb.h"
 #include "envoy/config/typed_metadata.h"
+#include "envoy/event/deferred_deletable.h"
 #include "envoy/http/codec.h"
 #include "envoy/http/codes.h"
 #include "envoy/http/conn_pool.h"
@@ -1254,9 +1255,8 @@ class GenericConnectionPoolCallbacks {
  *
  * It is similar logically to RequestEncoder, only without the getStream interface.
  */
-class GenericUpstream {
+class GenericUpstream : public Event::DeferredDeletable {
 public:
-  virtual ~GenericUpstream() = default;
   /**
    * Encode a data frame.
    * @param data supplies the data to encode. The data may be moved by the encoder.

source/common/http/status.cc

@@ -25,6 +25,8 @@ absl::string_view statusCodeToString(StatusCode code) {
     return "CodecClientError";
   case StatusCode::InboundFramesWithEmptyPayload:
     return "InboundFramesWithEmptyPayloadError";
+  case StatusCode::StreamAlreadyReset:
+    return "StreamAlreadyReset";
   }
   NOT_REACHED_GCOVR_EXCL_LINE;
 }
@@ -113,6 +115,13 @@ Status inboundFramesWithEmptyPayloadError() {
   return status;
 }
 
+Status streamAlreadyReset() {
+  absl::Status status(absl::StatusCode::kInternal,
+                      "Attempted to proxy headers after stream has been reset.");
+  storePayload(status, EnvoyStatusPayload(StatusCode::StreamAlreadyReset));
+  return status;
+}
+
 // Methods for checking and extracting error information
 StatusCode getStatusCode(const Status& status) {
   return status.ok() ? StatusCode::Ok : getPayload(status).status_code_;

source/common/http/status.h

@@ -74,6 +74,11 @@ enum class StatusCode : int {
    * Indicates that peer sent too many consecutive DATA frames with empty payload.
    */
   InboundFramesWithEmptyPayload = 5,
+
+  /**
+   * Indicates that we attempted to proxy upstream headers to a stream that has already been reset.
+   */
+  StreamAlreadyReset = 6,
 };
 
 using Status = absl::Status;
@@ -94,6 +99,7 @@ Status bufferFloodError(absl::string_view message);
 Status prematureResponseError(absl::string_view message, Http::Code http_code);
 Status codecClientError(absl::string_view message);
 Status inboundFramesWithEmptyPayloadError();
+Status streamAlreadyReset();
 
 /**
  * Returns Envoy::StatusCode of the given status object.
@@ -109,6 +115,7 @@ ABSL_MUST_USE_RESULT bool isBufferFloodError(const Status& status);
 ABSL_MUST_USE_RESULT bool isPrematureResponseError(const Status& status);
 ABSL_MUST_USE_RESULT bool isCodecClientError(const Status& status);
 ABSL_MUST_USE_RESULT bool isInboundFramesWithEmptyPayloadError(const Status& status);
+ABSL_MUST_USE_RESULT bool isStreamAlreadyReset(const Status& status);
 
 /**
  * Returns Http::Code value of the PrematureResponseError status.

source/common/router/upstream_request.cc

@@ -481,8 +481,8 @@ void UpstreamRequest::clearRequestEncoder() {
   // Before clearing the encoder, unsubscribe from callbacks.
   if (upstream_) {
     parent_.callbacks()->removeDownstreamWatermarkCallbacks(downstream_watermark_manager_);
+    parent_.callbacks()->dispatcher().deferredDelete(std::move(upstream_));
   }
-  upstream_.reset();
 }
 
 void UpstreamRequest::DownstreamWatermarkManager::onAboveWriteBufferHighWatermark() {

source/extensions/upstreams/http/tcp/upstream_request.cc

@@ -11,6 +11,7 @@
 #include "common/http/header_map_impl.h"
 #include "common/http/headers.h"
 #include "common/http/message_impl.h"
+#include "common/http/status.h"
 #include "common/network/transport_socket_options_impl.h"
 #include "common/router/router.h"
 
@@ -45,6 +46,10 @@ void TcpUpstream::encodeData(Buffer::Instance& data, bool end_stream) {
 
 Envoy::Http::Status TcpUpstream::encodeHeaders(const Envoy::Http::RequestHeaderMap&,
                                                bool end_stream) {
+  if (!upstream_request_) {
+    return Envoy::Http::streamAlreadyReset();
+  }
+
   // Headers should only happen once, so use this opportunity to add the proxy
   // proto header, if configured.
   ASSERT(upstream_request_->routeEntry().connectConfig().has_value());
@@ -86,7 +91,16 @@ void TcpUpstream::resetStream() {
 }
 
 void TcpUpstream::onUpstreamData(Buffer::Instance& data, bool end_stream) {
+  if (!upstream_request_) {
+    return;
+  }
+
   upstream_request_->decodeData(data, end_stream);
+  // This ensures that if we get a reset after end_stream we won't propagate two
+  // "end streams" to the upstream_request_.
+  if (end_stream) {
+    upstream_request_ = nullptr;
+  }
 }
 
 void TcpUpstream::onEvent(Network::ConnectionEvent event) {