retry: enforce a cap on max replay buffer size

Cargo.lock

@@ -703,6 +703,7 @@ dependencies = [
  "linkerd-proxy-tcp",
  "linkerd-proxy-transport",
  "linkerd-reconnect",
+ "linkerd-retry",
  "linkerd-service-profiles",
  "linkerd-stack",
  "linkerd-stack-metrics",
@@ -1019,13 +1020,11 @@ dependencies = [
  "http-body",
  "hyper",
  "linkerd-error",
- "linkerd-http-box",
- "linkerd-stack",
  "linkerd-tracing",
  "parking_lot",
  "pin-project",
+ "thiserror",
  "tokio",
- "tower",
  "tracing",
 ]
 
@@ -1298,6 +1297,17 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "linkerd-retry"
+version = "0.1.0"
+dependencies = [
+ "linkerd-error",
+ "linkerd-stack",
+ "pin-project",
+ "tower",
+ "tracing",
+]
+
 [[package]]
 name = "linkerd-service-profiles"
 version = "0.1.0"

Cargo.toml

@@ -46,6 +46,7 @@ members = [
     "linkerd/proxy/tcp",
     "linkerd/proxy/transport",
     "linkerd/reconnect",
+    "linkerd/retry",
     "linkerd/service-profiles",
     "linkerd/signal",
     "linkerd/stack",

linkerd/app/core/Cargo.toml

@@ -52,6 +52,7 @@ linkerd-proxy-tap = { path = "../../proxy/tap" }
 linkerd-proxy-tcp = { path = "../../proxy/tcp" }
 linkerd-proxy-transport = { path = "../../proxy/transport" }
 linkerd-reconnect = { path = "../../reconnect" }
+linkerd-retry = { path = "../../retry" }
 linkerd-timeout = { path = "../../timeout" }
 linkerd-tracing = { path = "../../tracing" }
 linkerd-service-profiles = { path = "../../service-profiles" }

linkerd/app/core/src/retry.rs

@@ -4,12 +4,14 @@ use super::http_metrics::retries::Handle;
 use super::metrics::HttpRouteRetry;
 use crate::profiles;
 use futures::future;
+use linkerd_error::Error;
 use linkerd_http_classify::{Classify, ClassifyEos, ClassifyResponse};
-use linkerd_stack::Param;
+use linkerd_stack::{Either, Param};
 use std::sync::Arc;
-use tower::retry::budget::Budget;
+use tower::retry::{budget::Budget, Policy};
 
-pub use linkerd_http_retry::*;
+use linkerd_http_retry::ReplayBody;
+use linkerd_retry::*;
 
 pub fn layer(metrics: HttpRouteRetry) -> NewRetryLayer<NewRetry> {
     NewRetryLayer::new(NewRetry::new(metrics))
@@ -55,6 +57,37 @@ impl NewPolicy<Route> for NewRetry {
 
 // === impl Retry ===
 
+impl Retry {
+    fn can_retry<A: http_body::Body>(&self, req: &http::Request<A>) -> bool {
+        let content_length = |req: &http::Request<_>| {
+            req.headers()
+                .get(http::header::CONTENT_LENGTH)
+                .and_then(|value| value.to_str().ok()?.parse::<usize>().ok())
+        };
+
+        // Requests without bodies can always be retried, as we will not need to
+        // buffer the body. If the request *does* have a body, retry it if and
+        // only if the request contains a `content-length` header and the
+        // content length is >= 64 kb.
+        let has_body = !req.body().is_end_stream();
+        if has_body && content_length(&req).unwrap_or(usize::MAX) > MAX_BUFFERED_BYTES {
+            tracing::trace!(
+                req.has_body = has_body,
+                req.content_length = ?content_length(&req),
+                "not retryable",
+            );
+            return false;
+        }
+
+        tracing::trace!(
+            req.has_body = has_body,
+            req.content_length = ?content_length(&req),
+            "retryable",
+        );
+        true
+    }
+}
+
 impl<A, B, E> Policy<http::Request<A>, http::Response<B>, E> for Retry
 where
     A: http_body::Body + Clone,
@@ -109,33 +142,20 @@ where
     }
 }
 
-impl<A: http_body::Body> CanRetry<A> for Retry {
-    fn can_retry(&self, req: &http::Request<A>) -> bool {
-        let content_length = |req: &http::Request<_>| {
-            req.headers()
-                .get(http::header::CONTENT_LENGTH)
-                .and_then(|value| value.to_str().ok()?.parse::<usize>().ok())
-        };
+impl<A, B, E> RetryPolicy<http::Request<A>, http::Response<B>, E> for Retry
+where
+    A: http_body::Body + Unpin,
+    A::Error: Into<Error>,
+{
+    type RetryRequest = http::Request<ReplayBody<A>>;
 
-        // Requests without bodies can always be retried, as we will not need to
-        // buffer the body. If the request *does* have a body, retry it if and
-        // only if the request contains a `content-length` header and the
-        // content length is >= 64 kb.
-        let has_body = !req.body().is_end_stream();
-        if has_body && content_length(&req).unwrap_or(usize::MAX) > MAX_BUFFERED_BYTES {
-            tracing::trace!(
-                req.has_body = has_body,
-                req.content_length = ?content_length(&req),
-                "not retryable",
-            );
-            return false;
+    fn prepare_request(
+        &self,
+        req: http::Request<A>,
+    ) -> Either<Self::RetryRequest, http::Request<A>> {
+        if self.can_retry(&req) {
+            return Either::A(req.map(|body| ReplayBody::new(body, MAX_BUFFERED_BYTES)));
         }
-
-        tracing::trace!(
-            req.has_body = has_body,
-            req.content_length = ?content_length(&req),
-            "retryable",
-        );
-        true
+        Either::B(req)
     }
 }

linkerd/app/inbound/fuzz/Cargo.lock

@@ -610,6 +610,7 @@ dependencies = [
  "linkerd-proxy-tcp",
  "linkerd-proxy-transport",
  "linkerd-reconnect",
+ "linkerd-retry",
  "linkerd-service-profiles",
  "linkerd-stack",
  "linkerd-stack-metrics",
@@ -863,11 +864,9 @@ dependencies = [
  "http",
  "http-body",
  "linkerd-error",
- "linkerd-http-box",
- "linkerd-stack",
  "parking_lot",
  "pin-project",
- "tower",
+ "thiserror",
  "tracing",
 ]
 
@@ -1127,6 +1126,17 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "linkerd-retry"
+version = "0.1.0"
+dependencies = [
+ "linkerd-error",
+ "linkerd-stack",
+ "pin-project",
+ "tower",
+ "tracing",
+]
+
 [[package]]
 name = "linkerd-service-profiles"
 version = "0.1.0"

linkerd/app/outbound/src/http/logical.rs

@@ -120,6 +120,12 @@ impl<E> Outbound<E> {
                             .http_route_actual
                             .to_layer::<classify::Response, _>(),
                     )
+                    // Depending on whether or not the request can be retried,
+                    // it may have one of two `Body` types. This layer unifies
+                    // any `Body` type into `BoxBody` so that the rest of the
+                    // stack doesn't have to implement `Service` for requests
+                    // with both body types.
+                    .push_on_response(http::EraseRequest::layer())
                     // Sets an optional retry policy.
                     .push(retry::layer(rt.metrics.http_route_retry.clone()))
                     // Sets an optional request timeout.

linkerd/http-box/src/erase_request.rs

@@ -0,0 +1,77 @@
+//! A middleware that boxes HTTP request bodies.
+use crate::BoxBody;
+use linkerd_error::Error;
+use linkerd_stack::{layer, Proxy};
+use std::task::{Context, Poll};
+
+/// A middleware that boxes HTTP request bodies.
+///
+/// This is *very* similar to the [`BoxRequest`] middleware. However, that
+/// middleware is generic over a specific body type that is erased. A given
+/// instance of `EraseRequest` can only erase the type of one particular `Body`
+/// type, while this middleware will erase
+/// bodies of *any* type.
+///
+/// An astute reader may ask, why not simply replace `BoxRequest` with this
+/// middleware, if it is a more  flexible superset of the same behavior? The
+/// answer is that in many cases, the use of this more flexible middleware
+/// renders request body types uninferrable. If all `BoxRequest`s in the stack
+/// are replaced with `EraseRequest`, suddenly a great deal of
+/// `check_new_service` and `check_service` checks will require explicit
+/// annotations for the pre-erasure body type. This is not great.
+///
+/// Instead, this type is implemented separately and should be used only when a
+/// stack must be able to implement `Service<http::Request<B>>` for *multiple
+/// distinct values of `B`*.
+#[derive(Debug)]
+pub struct EraseRequest<S>(S);
+
+impl<S> EraseRequest<S> {
+    pub fn layer() -> impl layer::Layer<S, Service = Self> + Clone + Copy {
+        layer::mk(EraseRequest)
+    }
+}
+
+impl<S: Clone> Clone for EraseRequest<S> {
+    fn clone(&self) -> Self {
+        EraseRequest(self.0.clone())
+    }
+}
+
+impl<S, B> tower::Service<http::Request<B>> for EraseRequest<S>
+where
+    B: http_body::Body + Send + 'static,
+    B::Data: Send + 'static,
+    B::Error: Into<Error>,
+    S: tower::Service<http::Request<BoxBody>>,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = S::Future;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.0.poll_ready(cx)
+    }
+
+    fn call(&mut self, req: http::Request<B>) -> Self::Future {
+        self.0.call(req.map(BoxBody::new))
+    }
+}
+
+impl<S, B, P> Proxy<http::Request<B>, S> for EraseRequest<P>
+where
+    B: http_body::Body + Send + 'static,
+    B::Data: Send + 'static,
+    B::Error: Into<Error>,
+    S: tower::Service<P::Request>,
+    P: Proxy<http::Request<BoxBody>, S>,
+{
+    type Request = P::Request;
+    type Response = P::Response;
+    type Error = P::Error;
+    type Future = P::Future;
+
+    fn proxy(&self, inner: &mut S, req: http::Request<B>) -> Self::Future {
+        self.0.proxy(inner, req.map(BoxBody::new))
+    }
+}

linkerd/http-box/src/lib.rs

@@ -3,11 +3,13 @@
 #![allow(clippy::inconsistent_struct_constructor)]
 
 mod body;
+mod erase_request;
 mod request;
 mod response;
 
 pub use self::{
     body::{BoxBody, Data},
+    erase_request::EraseRequest,
     request::BoxRequest,
     response::BoxResponse,
 };

linkerd/http-retry/Cargo.toml

@@ -12,12 +12,10 @@ futures = { version = "0.3", default-features = false }
 http-body = "0.4"
 http = "0.2"
 linkerd-error = { path = "../error" }
-linkerd-http-box = { path = "../http-box" }
-linkerd-stack = { path = "../stack" }
 pin-project = "1"
 parking_lot = "0.11"
-tower = { version = "0.4.7", default-features = false, features = ["retry", "util"] }
 tracing = "0.1.23"
+thiserror = "1"
 
 [dev-dependencies]
 hyper = "0.14"