add REFUSED_STREAM retry policy (#74)

This also requires wiring up REFUSED_STREAM handling in the HTTP/2 codec.

docs/configuration/http_filters/router_filter.rst

@@ -68,7 +68,7 @@ using a ',' delimited list. The supported policies are:
 
 5xx
   Envoy will attempt a retry if the upstream server responds with any 5xx response code, or does not
-  respond at all (disconnect/reset/etc.). (Includes *connect-failure*)
+  respond at all (disconnect/reset/etc.). (Includes *connect-failure* and *refused-stream*)
 
   * **NOTE:** Envoy will not retry when a request times out (resulting in a 504 error code). This is
     by design. The request timeout is an outer time limit for a request, including any retries that
@@ -92,6 +92,10 @@ retriable-4xx
     needs to read then attempt another write. If a retry happens in this type of case it will always
     fail with another 409.
 
+refused-stream
+  Envoy will attempt a retry if the upstream server resets the stream with a REFUSED_STREAM error
+  code. This reset type indicates that a request is safe to retry. (Included in *5xx*)
+
 The number of retries can be controlled via the
 :ref:`config_http_filters_router_x-envoy-max-retries` header or via the :ref:`route
 configuration <config_http_conn_man_route_table_route_retry>`.

include/envoy/http/codec.h

@@ -77,8 +77,12 @@ class StreamDecoder {
 enum class StreamResetReason {
   // If a local codec level reset was sent on the stream.
   LocalReset,
+  // If a local codec level refused stream reset was sent on the stream (allowing for retry).
+  LocalRefusedStreamReset,
   // If a remote codec level reset was received on the stream.
   RemoteReset,
+  // If a remote codec level refused stream reset was received on the stream (allowing for retry).
+  RemoteRefusedStreamReset,
   // If the stream was locally reset by a connection pool due to an initial connection failure.
   ConnectionFailure,
   // If the stream was locally reset due to connection termination.

include/envoy/router/router.h

@@ -31,6 +31,7 @@ class RetryPolicy {
   static const uint32_t RETRY_ON_5XX             = 0x1;
   static const uint32_t RETRY_ON_CONNECT_FAILURE = 0x2;
   static const uint32_t RETRY_ON_RETRIABLE_4XX   = 0x4;
+  static const uint32_t RETRY_ON_REFUSED_STREAM  = 0x8;
   // clang-format on
 
   virtual ~RetryPolicy() {}

source/common/http/headers.h

@@ -63,6 +63,7 @@ class Headers {
   struct {
     const std::string _5xx{"5xx"};
     const std::string ConnectFailure{"connect-failure"};
+    const std::string RefusedStream{"refused-stream"};
     const std::string Retriable4xx{"retriable-4xx"};
   } EnvoyRetryOnValues;
 

source/common/http/http2/codec_impl.cc

@@ -194,8 +194,10 @@ void ConnectionImpl::StreamImpl::encodeData(const Buffer::Instance& data, bool e
 void ConnectionImpl::StreamImpl::resetStream(StreamResetReason reason) {
   // Higher layers expect calling resetStream() to immediately raise reset callbacks.
   runResetCallbacks(reason);
-  int rc =
-      nghttp2_submit_rst_stream(parent_.session_, NGHTTP2_FLAG_NONE, stream_id_, NGHTTP2_NO_ERROR);
+  int rc = nghttp2_submit_rst_stream(parent_.session_, NGHTTP2_FLAG_NONE, stream_id_,
+                                     reason == StreamResetReason::LocalRefusedStreamReset
+                                         ? NGHTTP2_REFUSED_STREAM
+                                         : NGHTTP2_NO_ERROR);
   ASSERT(rc == 0);
   UNREFERENCED_PARAMETER(rc);
 
@@ -344,7 +346,9 @@ int ConnectionImpl::onStreamClose(int32_t stream_id, uint32_t error_code) {
   if (stream) {
     conn_log_debug("stream closed: {}", connection_, error_code);
     if (!stream->remote_end_stream_ || !stream->local_end_stream_) {
-      stream->runResetCallbacks(StreamResetReason::RemoteReset);
+      stream->runResetCallbacks(error_code == NGHTTP2_REFUSED_STREAM
+                                    ? StreamResetReason::RemoteRefusedStreamReset
+                                    : StreamResetReason::RemoteReset);
     }
 
     connection_.dispatcher().deferredDelete(std::move(stream->removeFromList(active_streams_)));

source/common/router/retry_state_impl.cc

@@ -65,6 +65,8 @@ uint32_t RetryStateImpl::parseRetryOn(const std::string& config) {
       ret |= RetryPolicy::RETRY_ON_CONNECT_FAILURE;
     } else if (retry_on == Http::Headers::get().EnvoyRetryOnValues.Retriable4xx) {
       ret |= RetryPolicy::RETRY_ON_RETRIABLE_4XX;
+    } else if (retry_on == Http::Headers::get().EnvoyRetryOnValues.RefusedStream) {
+      ret |= RetryPolicy::RETRY_ON_REFUSED_STREAM;
     }
   }
 
@@ -127,6 +129,11 @@ bool RetryStateImpl::wouldRetry(const Http::HeaderMap* response_headers,
     }
   }
 
+  if ((retry_on_ & RetryPolicy::RETRY_ON_REFUSED_STREAM) && reset_reason.valid() &&
+      reset_reason.value() == Http::StreamResetReason::RemoteRefusedStreamReset) {
+    return true;
+  }
+
   if ((retry_on_ & RetryPolicy::RETRY_ON_CONNECT_FAILURE) && reset_reason.valid() &&
       reset_reason.value() == Http::StreamResetReason::ConnectionFailure) {
     return true;

source/common/router/router.cc

@@ -358,10 +358,12 @@ Filter::streamResetReasonToFailureReason(Http::StreamResetReason reset_reason) {
   case Http::StreamResetReason::ConnectionTermination:
     return Http::AccessLog::FailureReason::UpstreamConnectionTermination;
   case Http::StreamResetReason::LocalReset:
+  case Http::StreamResetReason::LocalRefusedStreamReset:
     return Http::AccessLog::FailureReason::LocalReset;
   case Http::StreamResetReason::Overflow:
     return Http::AccessLog::FailureReason::UpstreamOverflow;
   case Http::StreamResetReason::RemoteReset:
+  case Http::StreamResetReason::RemoteRefusedStreamReset:
     return Http::AccessLog::FailureReason::UpstreamRemoteReset;
   }
 

test/common/http/http2/codec_impl_test.cc

@@ -60,6 +60,29 @@ class Http2CodecImplTest : public testing::TestWithParam<uint64_t> {
   ConnectionWrapper server_wrapper_;
 };
 
+TEST_P(Http2CodecImplTest, RefusedStreamReset) {
+  MockStreamDecoder response_decoder;
+  StreamEncoder& request_encoder = client_.newStream(response_decoder);
+
+  MockStreamDecoder request_decoder;
+  StreamEncoder* response_encoder;
+  EXPECT_CALL(server_callbacks_, newStream(_))
+      .WillOnce(Invoke([&](StreamEncoder& encoder) -> StreamDecoder& {
+        response_encoder = &encoder;
+        return request_decoder;
+      }));
+
+  HeaderMapImpl request_headers;
+  HttpTestUtility::addDefaultHeaders(request_headers);
+  EXPECT_CALL(request_decoder, decodeHeaders_(_, false));
+  request_encoder.encodeHeaders(request_headers, false);
+
+  MockStreamCallbacks callbacks;
+  request_encoder.getStream().addCallbacks(callbacks);
+  EXPECT_CALL(callbacks, onResetStream(StreamResetReason::RemoteRefusedStreamReset));
+  response_encoder->getStream().resetStream(StreamResetReason::LocalRefusedStreamReset);
+}
+
 TEST_P(Http2CodecImplTest, InvalidFrame) {
   MockStreamDecoder response_decoder;
   StreamEncoder& request_encoder = client_.newStream(response_decoder);

test/common/router/retry_state_impl_test.cc

@@ -41,6 +41,8 @@ class RouterRetryStateImplTest : public testing::Test {
 
   const Optional<Http::StreamResetReason> no_reset_;
   const Optional<Http::StreamResetReason> remote_reset_{Http::StreamResetReason::RemoteReset};
+  const Optional<Http::StreamResetReason> remote_refused_stream_reset_{
+      Http::StreamResetReason::RemoteRefusedStreamReset};
   const Optional<Http::StreamResetReason> connect_failure_{
       Http::StreamResetReason::ConnectionFailure};
 };
@@ -52,6 +54,19 @@ TEST_F(RouterRetryStateImplTest, PolicyNoneRemoteReset) {
   EXPECT_FALSE(state_->shouldRetry(nullptr, remote_reset_, callback_));
 }
 
+TEST_F(RouterRetryStateImplTest, PolicyRefusedStream) {
+  Http::HeaderMapImpl request_headers{{"x-envoy-retry-on", "refused-stream"}};
+  setup(request_headers);
+  EXPECT_TRUE(state_->enabled());
+
+  expectTimerCreateAndEnable();
+  EXPECT_TRUE(state_->shouldRetry(nullptr, remote_refused_stream_reset_, callback_));
+  EXPECT_CALL(callback_ready_, ready());
+  retry_timer_->callback_();
+
+  EXPECT_FALSE(state_->shouldRetry(nullptr, remote_refused_stream_reset_, callback_));
+}
+
 TEST_F(RouterRetryStateImplTest, Policy5xxRemoteReset) {
   Http::HeaderMapImpl request_headers{{"x-envoy-retry-on", "5xx"}};
   setup(request_headers);