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,48 +4,50 @@ 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_http_retry::ReplayBody;
+use linkerd_retry as retry;
+use linkerd_stack::{layer, Either, Param};
 use std::sync::Arc;
-use tower::retry::budget::Budget;
 
-pub use linkerd_http_retry::*;
-
-pub fn layer(metrics: HttpRouteRetry) -> NewRetryLayer<NewRetry> {
-    NewRetryLayer::new(NewRetry::new(metrics))
+pub fn layer<N>(
+    metrics: HttpRouteRetry,
+) -> impl layer::Layer<N, Service = retry::NewRetry<NewRetryPolicy, N>> + Clone {
+    retry::NewRetry::<_, N>::layer(NewRetryPolicy::new(metrics))
 }
 
 #[derive(Clone, Debug)]
-pub struct NewRetry {
+pub struct NewRetryPolicy {
     metrics: HttpRouteRetry,
 }
 
 #[derive(Clone, Debug)]
-pub struct Retry {
+pub struct RetryPolicy {
     metrics: Handle,
-    budget: Arc<Budget>,
+    budget: Arc<retry::Budget>,
     response_classes: profiles::http::ResponseClasses,
 }
 
 /// Allow buffering requests up to 64 kb
 const MAX_BUFFERED_BYTES: usize = 64 * 1024;
 
-// === impl NewRetry ===
+// === impl NewRetryPolicy ===
 
-impl NewRetry {
+impl NewRetryPolicy {
     pub fn new(metrics: HttpRouteRetry) -> Self {
         Self { metrics }
     }
 }
 
-impl NewPolicy<Route> for NewRetry {
-    type Policy = Retry;
+impl retry::NewPolicy<Route> for NewRetryPolicy {
+    type Policy = RetryPolicy;
 
     fn new_policy(&self, route: &Route) -> Option<Self::Policy> {
         let retries = route.route.retries().cloned()?;
 
         let metrics = self.metrics.get_handle(route.param());
-        Some(Retry {
+        Some(RetryPolicy {
             metrics,
             budget: retries.budget().clone(),
             response_classes: route.route.response_classes().clone(),
@@ -55,7 +57,38 @@ impl NewPolicy<Route> for NewRetry {
 
 // === impl Retry ===
 
-impl<A, B, E> Policy<http::Request<A>, http::Response<B>, E> for Retry
+impl RetryPolicy {
+    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> retry::Policy<http::Request<A>, http::Response<B>, E> for RetryPolicy
 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> retry::PrepareRequest<http::Request<A>, http::Response<B>, E> for RetryPolicy
+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::BoxRequest::erased())
                     // Sets an optional retry policy.
                     .push(retry::layer(rt.metrics.http_route_retry.clone()))
                     // Sets an optional request timeout.
@@ -134,7 +140,7 @@ impl<E> Outbound<E> {
             ))
             // Strips headers that may be set by this proxy and add an outbound
             // canonical-dst-header. The response body is boxed unify the profile
-            // stack's response type. withthat of to endpoint stack.
+            // stack's response type with that of to endpoint stack.
             .push(http::NewHeaderFromTarget::<CanonicalDstHeader, _>::layer())
             .push_on_response(svc::layers().push(http::BoxResponse::layer()))
             .instrument(|l: &Logical| debug_span!("logical", dst = %l.logical_addr))

linkerd/http-box/src/erase_request.rs

@@ -0,0 +1,84 @@
+//! A middleware that boxes HTTP request bodies.
+
+use crate::BoxBody;
+use linkerd_error::Error;
+use linkerd_stack::{layer, Proxy};
+use std::task::{Context, Poll};
+
+/// Boxes request bodies, erasing the original type.
+///
+/// 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 new(inner: S) -> Self {
+        Self(inner)
+    }
+
+    pub fn layer() -> impl layer::Layer<S, Service = Self> + Clone + Copy {
+        layer::mk(Self::new)
+    }
+}
+
+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;
+
+    #[inline]
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.0.poll_ready(cx)
+    }
+
+    #[inline]
+    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;
+
+    #[inline]
+    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,
 };