chore(deps)!: upgrade to tower 0.5

this commit updates our tower dependency from 0.4 to 0.5.

note that this commit does not affect the `tower-service` and
`tower-layer` crates, reëxported by `tower` itself. the `Service<T>`
trait and the closely related `Layer<S>` trait have not been changed.

the `tower` crate's utilities have changed in various ways, some of
particular note for the linkerd2 proxy. see these items, excerpted from
the tower changelog:

- **retry**: **Breaking Change** `retry::Policy::retry` now accepts `&mut Req` and `&mut Res` instead of the previous mutable versions. This
  increases the flexibility of the retry policy. To update, update your method signature to include `mut` for both parameters. ([tower-rs/tower#584])
- **retry**: **Breaking Change** Change Policy to accept &mut self ([tower-rs/tower#681])
- **retry**: **Breaking Change** `Budget` is now a trait. This allows end-users to implement their own budget and bucket implementations. ([tower-rs/tower#703])
- **util**: **Breaking Change** `Either::A` and `Either::B` have been renamed `Either::Left` and `Either::Right`, respectively. ([tower-rs/tower#637])
- **util**: **Breaking Change** `Either` now requires its two services to have the same error type. ([tower-rs/tower#637])
- **util**: **Breaking Change** `Either` no longer implemenmts `Future`. ([tower-rs/tower#637])
- **buffer**: **Breaking Change** `Buffer<S, Request>` is now generic over `Buffer<Request, S::Future>.` ([tower-rs/tower#654])

see:

* <tower-rs/tower#584>
* <tower-rs/tower#681>
* <tower-rs/tower#703>
* <tower-rs/tower#637>
* <tower-rs/tower#654>

the `Either` trait bounds are particularly impactful for us. because
this runs counter to how we treat errors (skewing towards boxed errors,
in general), we temporarily vendor a version of `Either` from the 0.4
release, whose variants have been renamed to match the 0.5 interface.

updating to box the inner `A` and `B` services' errors, so we satiate
the new `A::Error = B::Error` bounds, can be addressed as a follow-on.
that's intentionally left as a separate change, due to the net size of
our patchset between this branch and #3504.

* <tower-rs/tower@v0.4.x...master>
* <https://github.com/tower-rs/tower/blob/master/tower/CHANGELOG.md>

this work is based upon #3504. for more information, see:

* linkerd/linkerd2#8733
* #3504

Signed-off-by: katelyn martin <[email protected]>
X-Ref: tower-rs/tower#815
X-Ref: tower-rs/tower#817
X-Ref: tower-rs/tower#818
X-Ref: tower-rs/tower#819

Author: cratelyn

Cargo.lock

@@ -993,7 +993,7 @@ dependencies = [
  "pin-project",
  "tokio",
  "tokio-test",
- "tower 0.4.13",
+ "tower 0.5.2",
 ]
 
 [[package]]
@@ -1393,7 +1393,7 @@ dependencies = [
  "tokio",
  "tokio-stream",
  "tonic",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -1416,7 +1416,7 @@ dependencies = [
  "serde_json",
  "thiserror 2.0.12",
  "tokio",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -1485,7 +1485,7 @@ dependencies = [
  "tokio",
  "tokio-stream",
  "tonic",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -1506,7 +1506,7 @@ dependencies = [
  "tokio",
  "tokio-test",
  "tonic",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tower-test",
  "tracing",
 ]
@@ -1545,7 +1545,7 @@ dependencies = [
  "tokio",
  "tokio-test",
  "tonic",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -1581,7 +1581,7 @@ dependencies = [
  "tokio-rustls",
  "tokio-stream",
  "tonic",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
  "tracing-subscriber",
 ]
@@ -1629,7 +1629,7 @@ dependencies = [
  "tokio-rustls",
  "tokio-test",
  "tonic",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tower-test",
  "tracing",
 ]
@@ -1658,7 +1658,7 @@ dependencies = [
  "tokio-stream",
  "tokio-test",
  "tonic",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
  "tracing-subscriber",
 ]
@@ -1835,7 +1835,7 @@ dependencies = [
  "http",
  "linkerd-stack",
  "pin-project",
- "tower 0.4.13",
+ "tower 0.5.2",
 ]
 
 [[package]]
@@ -1854,7 +1854,7 @@ dependencies = [
  "parking_lot",
  "pin-project",
  "tokio",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -1864,7 +1864,7 @@ version = "0.1.0"
 dependencies = [
  "http",
  "linkerd-stack",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -1895,7 +1895,7 @@ dependencies = [
  "http-body",
  "linkerd-stack",
  "pin-project",
- "tower 0.4.13",
+ "tower 0.5.2",
 ]
 
 [[package]]
@@ -1919,7 +1919,7 @@ dependencies = [
  "pin-project",
  "thiserror 2.0.12",
  "tokio",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -1973,7 +1973,7 @@ dependencies = [
  "pin-project",
  "thiserror 2.0.12",
  "tokio",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
  "try-lock",
 ]
@@ -2009,7 +2009,7 @@ dependencies = [
  "linkerd-tracing",
  "parking_lot",
  "tokio",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -2212,7 +2212,7 @@ dependencies = [
  "rand 0.9.0",
  "tokio",
  "tokio-test",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tower-test",
  "tracing",
 ]
@@ -2236,7 +2236,7 @@ dependencies = [
  "pin-project",
  "prost 0.13.5",
  "tonic",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -2254,7 +2254,7 @@ dependencies = [
  "linkerd-stack",
  "rand 0.9.0",
  "tokio",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -2319,7 +2319,7 @@ version = "0.1.0"
 dependencies = [
  "futures",
  "linkerd-error",
- "tower 0.4.13",
+ "tower 0.5.2",
 ]
 
 [[package]]
@@ -2334,7 +2334,7 @@ dependencies = [
  "linkerd-stack",
  "tokio",
  "tokio-stream",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -2377,7 +2377,7 @@ dependencies = [
  "thiserror 2.0.12",
  "tokio",
  "tokio-test",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tower-test",
  "tracing",
  "try-lock",
@@ -2412,7 +2412,7 @@ dependencies = [
  "linkerd-proxy-core",
  "pin-project",
  "thiserror 2.0.12",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -2451,7 +2451,7 @@ dependencies = [
  "tokio",
  "tokio-test",
  "tonic",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
  "x509-parser",
 ]
@@ -2483,7 +2483,7 @@ dependencies = [
  "thiserror 2.0.12",
  "tokio",
  "tonic",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -2499,7 +2499,7 @@ dependencies = [
  "pin-project",
  "rand 0.9.0",
  "tokio",
- "tower 0.4.13",
+ "tower 0.5.2",
 ]
 
 [[package]]
@@ -2530,7 +2530,7 @@ dependencies = [
  "tokio",
  "tokio-stream",
  "tokio-test",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tower-test",
  "tracing",
 ]
@@ -2542,7 +2542,7 @@ dependencies = [
  "futures",
  "linkerd-error",
  "linkerd-stack",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -2584,7 +2584,7 @@ dependencies = [
  "tokio",
  "tokio-stream",
  "tonic",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -2609,7 +2609,7 @@ dependencies = [
  "tokio",
  "tokio-test",
  "tokio-util",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tower-test",
  "tracing",
 ]
@@ -2622,7 +2622,7 @@ dependencies = [
  "parking_lot",
  "tokio",
  "tokio-test",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tower-test",
 ]
 
@@ -2633,7 +2633,7 @@ dependencies = [
  "futures",
  "linkerd-error",
  "linkerd-stack",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -2654,7 +2654,7 @@ dependencies = [
  "pin-project",
  "thiserror 2.0.12",
  "tokio",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
  "untrusted",
 ]
@@ -2720,7 +2720,7 @@ dependencies = [
  "linkerd-stack",
  "rand 0.8.5",
  "thiserror 1.0.69",
- "tower 0.4.13",
+ "tower 0.5.2",
  "tracing",
 ]
 
@@ -4198,10 +4198,14 @@ checksum = "d039ad9159c98b70ecfd540b2573b97f7f52c3e8d9f8ad57a24b916a536975f9"
 dependencies = [
  "futures-core",
  "futures-util",
+ "indexmap 2.8.0",
  "pin-project-lite",
  "sync_wrapper",
+ "tokio",
+ "tokio-util",
  "tower-layer",
  "tower-service",
+ "tracing",
 ]
 
 [[package]]

Cargo.toml

@@ -122,7 +122,7 @@ tokio-rustls = { version = "0.26", default-features = false, features = [
 ] }
 tonic = { version = "0.12", default-features = false }
 tonic-build = { version = "0.12", default-features = false }
-tower = { version = "0.4", default-features = false }
+tower = { version = "0.5", default-features = false }
 tower-service = { version = "0.3" }
 tower-test = { version = "0.4" }
 

linkerd/app/gateway/src/server.rs

@@ -64,10 +64,10 @@ impl Gateway {
                             SessionProtocol::Http1 => http::Variant::Http1,
                             SessionProtocol::Http2 => http::Variant::H2,
                         };
-                        return Ok(svc::Either::A(Http { parent, version }));
+                        return Ok(svc::Either::Left(Http { parent, version }));
                     }
 
-                    Ok(svc::Either::B(Opaq(parent)))
+                    Ok(svc::Either::Right(Opaq(parent)))
                 },
                 opaq,
             )

linkerd/app/inbound/src/accept.rs

@@ -53,12 +53,12 @@ impl<N> Inbound<N> {
                     move |t: T| -> Result<_, Error> {
                         let addr: OrigDstAddr = t.param();
                         if addr.port() == proxy_port {
-                            return Ok(svc::Either::B(t));
+                            return Ok(svc::Either::Right(t));
                         }
 
                         let policy = policies.get_policy(addr);
                         tracing::debug!(policy = ?&*policy.borrow(), "Accepted");
-                        Ok(svc::Either::A(Accept {
+                        Ok(svc::Either::Left(Accept {
                             client_addr: t.param(),
                             orig_dst_addr: addr,
                             policy,

linkerd/app/inbound/src/detect.rs

@@ -120,8 +120,10 @@ impl Inbound<svc::ArcNewTcp<Http, io::BoxedIo>> {
                 .push_switch(
                     |(detected, Detect { tls, .. })| -> Result<_, Infallible> {
                         match detected {
-                            http::Detection::Http(http) => Ok(svc::Either::A(Http { http, tls })),
-                            http::Detection::NotHttp => Ok(svc::Either::B(tls)),
+                            http::Detection::Http(http) => {
+                                Ok(svc::Either::Left(Http { http, tls }))
+                            }
+                            http::Detection::NotHttp => Ok(svc::Either::Right(tls)),
                             // When HTTP detection fails, forward the connection to the application as
                             // an opaque TCP stream.
                             http::Detection::ReadTimeout(timeout) => {
@@ -140,7 +142,7 @@ impl Inbound<svc::ArcNewTcp<Http, io::BoxedIo>> {
                                             ?timeout,
                                             "Handling connection as HTTP/1 due to policy"
                                         );
-                                        Ok(svc::Either::A(Http {
+                                        Ok(svc::Either::Left(Http {
                                             http: http::Variant::Http1,
                                             tls,
                                         }))
@@ -156,7 +158,7 @@ impl Inbound<svc::ArcNewTcp<Http, io::BoxedIo>> {
                                             ?timeout,
                                             "Handling connection as opaque due to policy"
                                         );
-                                        Ok(svc::Either::B(tls))
+                                        Ok(svc::Either::Right(tls))
                                     }
                                 }
                             }
@@ -183,15 +185,15 @@ impl Inbound<svc::ArcNewTcp<Http, io::BoxedIo>> {
                     move |tls: Tls| -> Result<_, Infallible> {
                         let http = match tls.policy.protocol() {
                             Protocol::Detect { timeout, .. } => {
-                                return Ok(svc::Either::B(Detect { timeout, tls }));
+                                return Ok(svc::Either::Right(Detect { timeout, tls }));
                             }
                             // Meshed HTTP/1 services may actually be transported over HTTP/2 connections
                             // between proxies, so we have to do detection.
                             //
                             // TODO(ver) outbound clients should hint this with ALPN so we don't
                             // have to detect this situation.
                             Protocol::Http1 { .. } if tls.status.is_some() => {
-                                return Ok(svc::Either::B(Detect {
+                                return Ok(svc::Either::Right(Detect {
                                     timeout: detect_timeout,
                                     tls,
                                 }));
@@ -202,7 +204,7 @@ impl Inbound<svc::ArcNewTcp<Http, io::BoxedIo>> {
                             Protocol::Http2 { .. } | Protocol::Grpc { .. } => http::Variant::H2,
                             _ => unreachable!("opaque protocols must not hit the HTTP stack"),
                         };
-                        Ok(svc::Either::A(Http { http, tls }))
+                        Ok(svc::Either::Left(Http { http, tls }))
                     },
                     detect.into_inner(),
                 )
@@ -256,10 +258,10 @@ impl<I> Inbound<svc::ArcNewTcp<Tls, TlsIo<I>>> {
                         // whether app TLS was employed, but we use this as a signal that we should
                         // not perform additional protocol detection.
                         if matches!(protocol, Protocol::Tls { .. }) {
-                            return Ok(svc::Either::B(tls));
+                            return Ok(svc::Either::Right(tls));
                         }
 
-                        Ok(svc::Either::A(tls))
+                        Ok(svc::Either::Left(tls))
                     },
                     forward
                         .clone()
@@ -283,14 +285,14 @@ impl<I> Inbound<svc::ArcNewTcp<Tls, TlsIo<I>>> {
                         if matches!(policy.protocol(), Protocol::Opaque { .. }) {
                             const TLS_PORT_SKIPPED: tls::ConditionalServerTls =
                                 tls::ConditionalServerTls::None(tls::NoServerTls::PortSkipped);
-                            return Ok(svc::Either::B(Tls {
+                            return Ok(svc::Either::Right(Tls {
                                 client_addr: t.param(),
                                 orig_dst_addr: t.param(),
                                 status: TLS_PORT_SKIPPED,
                                 policy,
                             }));
                         }
-                        Ok(svc::Either::A(t))
+                        Ok(svc::Either::Left(t))
                     },
                     forward
                         .push_on_service(svc::MapTargetLayer::new(io::BoxedIo::new))

linkerd/app/inbound/src/direct.rs

@@ -157,14 +157,14 @@ impl<N> Inbound<N> {
                                     port,
                                     name: None,
                                     protocol,
-                                } => Ok(svc::Either::A({
+                                } => Ok(svc::Either::Left({
                                     // When the transport header targets an alternate port (but does
                                     // not identify an alternate target name), we check the new
                                     // target's policy (rather than the inbound proxy's address).
                                     let addr = (client.local_addr.ip(), port).into();
                                     let policy = policies.get_policy(OrigDstAddr(addr));
                                     match protocol {
-                                        None => svc::Either::A(LocalTcp {
+                                        None => svc::Either::Left(LocalTcp {
                                             server_addr: Remote(ServerAddr(addr)),
                                             client_addr: client.client_addr,
                                             client_id: client.client_id,
@@ -174,7 +174,7 @@ impl<N> Inbound<N> {
                                             // When TransportHeader includes the protocol, but does not
                                             // include an alternate name we go through the Inbound HTTP
                                             // stack.
-                                            svc::Either::B(LocalHttp {
+                                            svc::Either::Right(LocalHttp {
                                                 addr: Remote(ServerAddr(addr)),
                                                 policy,
                                                 protocol,
@@ -188,7 +188,7 @@ impl<N> Inbound<N> {
                                     port,
                                     name: Some(name),
                                     protocol,
-                                } => Ok(svc::Either::B({
+                                } => Ok(svc::Either::Right({
                                     // When the transport header provides an alternate target, the
                                     // connection is a gateway connection. We check the _gateway
                                     // address's_ policy (rather than the target address).

linkerd/app/inbound/src/http/router.rs

@@ -163,14 +163,14 @@ impl<C> Inbound<C> {
                     |(rx, logical): (Option<profiles::Receiver>, Logical)| -> Result<_, Infallible> {
                         if let Some(rx) = rx {
                             if let Some(addr) = rx.logical_addr() {
-                                return Ok(svc::Either::A(Profile {
+                                return Ok(svc::Either::Left(Profile {
                                     addr,
                                     logical,
                                     profiles: rx,
                                 }));
                             }
                         }
-                        Ok(svc::Either::B(logical))
+                        Ok(svc::Either::Right(logical))
                     },
                     http.clone().into_inner(),
                 )
@@ -189,17 +189,17 @@ impl<C> Inbound<C> {
                         // discovery (so that we skip the profile stack above).
                         let addr = match logical.logical.clone() {
                             Some(addr) => addr,
-                            None => return Ok(svc::Either::B((None, logical))),
+                            None => return Ok(svc::Either::Right((None, logical))),
                         };
                         if !allow_profile.matches(addr.name()) {
                             tracing::debug!(
                                 %addr,
                                 suffixes = %allow_profile,
                                 "Skipping discovery, address not in configured DNS suffixes",
                             );
-                            return Ok(svc::Either::B((None, logical)));
+                            return Ok(svc::Either::Right((None, logical)));
                         }
-                        Ok(svc::Either::A(logical))
+                        Ok(svc::Either::Left(logical))
                     },
                     router
                         .check_new_service::<(Option<profiles::Receiver>, Logical), http::Request<_>>()

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

@@ -120,28 +120,30 @@ impl<N> Outbound<N> {
                     move |parent: T| -> Result<_, Infallible> {
                         Ok(match parent.param() {
                             Dispatch::Balance(addr, ewma) => {
-                                svc::Either::A(svc::Either::A(balance::Balance {
+                                svc::Either::Left(svc::Either::Left(balance::Balance {
                                     addr,
                                     ewma,
                                     parent,
                                     queue,
                                 }))
                             }
-                            Dispatch::Forward(addr, metadata) => svc::Either::A(svc::Either::B({
-                                let is_local = inbound_ips.contains(&addr.ip());
-                                let http2 = http2.override_from(metadata.http2_client_params());
-                                Endpoint {
-                                    is_local,
-                                    addr,
-                                    metadata,
-                                    parent,
-                                    queue,
-                                    close_server_connection_on_remote_proxy_error: true,
-                                    http1,
-                                    http2,
-                                }
-                            })),
-                            Dispatch::Fail { message } => svc::Either::B(message),
+                            Dispatch::Forward(addr, metadata) => {
+                                svc::Either::Left(svc::Either::Right({
+                                    let is_local = inbound_ips.contains(&addr.ip());
+                                    let http2 = http2.override_from(metadata.http2_client_params());
+                                    Endpoint {
+                                        is_local,
+                                        addr,
+                                        metadata,
+                                        parent,
+                                        queue,
+                                        close_server_connection_on_remote_proxy_error: true,
+                                        http1,
+                                        http2,
+                                    }
+                                }))
+                            }
+                            Dispatch::Fail { message } => svc::Either::Right(message),
                         })
                     },
                     svc::stack(fail).check_new_clone().into_inner(),

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

@@ -148,7 +148,7 @@ where
                                 );
                                 let parent_ref = ParentRef::from(ep.clone());
                                 let backend_ref = BackendRef::from(ep);
-                                svc::Either::A(Concrete {
+                                svc::Either::Left(Concrete {
                                     parent_ref,
                                     backend_ref,
                                     target: concrete::Dispatch::Forward(remote, meta),
@@ -157,8 +157,10 @@ where
                                     failure_accrual: Default::default(),
                                 })
                             }
-                            Self::Profile(profile) => svc::Either::B(svc::Either::A(profile)),
-                            Self::Policy(policy) => svc::Either::B(svc::Either::B(policy)),
+                            Self::Profile(profile) => {
+                                svc::Either::Right(svc::Either::Left(profile))
+                            }
+                            Self::Policy(policy) => svc::Either::Right(svc::Either::Right(policy)),
                         })
                     },
                     // Switch profile and policy routing.

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

@@ -72,8 +72,8 @@ where
             http.push_switch(
                 |pp: Policy<T>| {
                     Ok::<_, Infallible>(match pp {
-                        Self::Http(http) => svc::Either::A(http),
-                        Self::Grpc(grpc) => svc::Either::B(grpc),
+                        Self::Http(http) => svc::Either::Left(http),
+                        Self::Grpc(grpc) => svc::Either::Right(grpc),
                     })
                 },
                 grpc.into_inner(),

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

@@ -1,11 +1,14 @@
-use futures::{future, FutureExt};
+use futures::{
+    future::{self, Either},
+    FutureExt,
+};
 use linkerd_app_core::{
     classify,
     http_metrics::retries::Handle,
     metrics::{self, ProfileRouteLabels},
     profiles::{self, http::Route},
     proxy::http::{Body, ClientHandle, EraseResponse},
-    svc::{layer, Either, Param},
+    svc::{layer, Param},
     Error, Result,
 };
 use linkerd_http_classify::{Classify, ClassifyEos, ClassifyResponse};
@@ -33,7 +36,7 @@ pub struct NewRetryPolicy {
 #[derive(Clone, Debug)]
 pub struct RetryPolicy {
     metrics: Handle,
-    budget: Arc<retry::Budget>,
+    budget: Arc<retry::TpsBudget>,
     response_classes: profiles::http::ResponseClasses,
 }
 
@@ -75,13 +78,15 @@ where
     ReqB::Error: Into<Error>,
     RspB: Body + Unpin,
 {
-    type Future = future::Ready<Self>;
+    type Future = future::Ready<()>;
 
     fn retry(
-        &self,
-        req: &http::Request<ReplayBody<ReqB>>,
-        result: Result<&http::Response<PeekTrailersBody<RspB>>, &Error>,
+        &mut self,
+        req: &mut http::Request<ReplayBody<ReqB>>,
+        result: &mut Result<http::Response<PeekTrailersBody<RspB>>, Error>,
     ) -> Option<Self::Future> {
+        use retry::Budget as _;
+
         let retryable = match result {
             Err(_) => false,
             Ok(rsp) => {
@@ -113,17 +118,17 @@ where
             return None;
         }
 
-        let withdrew = self.budget.withdraw().is_ok();
+        let withdrew = self.budget.withdraw();
         self.metrics.incr_retryable(withdrew);
         if !withdrew {
             return None;
         }
 
-        Some(future::ready(self.clone()))
+        Some(future::ready(()))
     }
 
     fn clone_request(
-        &self,
+        &mut self,
         req: &http::Request<ReplayBody<ReqB>>,
     ) -> Option<http::Request<ReplayBody<ReqB>>> {
         // Since the body is already wrapped in a ReplayBody, it must not be obviously too large to
@@ -177,13 +182,13 @@ where
                     size = body.size_hint().lower(),
                     "Body is too large to buffer"
                 );
-                return Either::B(http::Request::from_parts(head, body));
+                return Either::Right(http::Request::from_parts(head, body));
             }
         };
 
         // The body may still be too large to be buffered if the body's length was not known.
         // `ReplayBody` handles this gracefully.
-        Either::A((self, http::Request::from_parts(head, replay_body)))
+        Either::Left((self, http::Request::from_parts(head, replay_body)))
     }
 
     /// If the response is HTTP/2, return a future that checks for a `TRAILERS`

linkerd/app/outbound/src/ingress.rs

@@ -326,14 +326,14 @@ impl<N> Outbound<N> {
                     |(detected, parent): (http::Detection, T)| -> Result<_, Infallible> {
                         match detected {
                             http::Detection::Http(version) => {
-                                return Ok(svc::Either::A(Http { version, parent }));
+                                return Ok(svc::Either::Left(Http { version, parent }));
                             }
                             http::Detection::ReadTimeout(timeout) => {
                                 tracing::info!("Continuing after timeout: {timeout:?}");
                             }
                             _ => {}
                         }
-                        Ok(svc::Either::B(parent))
+                        Ok(svc::Either::Right(parent))
                     },
                     fallback,
                 )

linkerd/app/outbound/src/opaq/concrete.rs

@@ -222,7 +222,7 @@ impl<C> Outbound<C> {
                     move |parent: T| -> Result<_, Infallible> {
                         Ok(match parent.param() {
                             Dispatch::Balance(addr, ewma) => {
-                                svc::Either::A(svc::Either::A(Balance {
+                                svc::Either::Left(svc::Either::Left(Balance {
                                     addr,
                                     ewma,
                                     queue,
@@ -231,14 +231,14 @@ impl<C> Outbound<C> {
                             }
 
                             Dispatch::Forward(addr, meta) => {
-                                svc::Either::A(svc::Either::B(Endpoint {
+                                svc::Either::Left(svc::Either::Right(Endpoint {
                                     addr,
                                     is_local: false,
                                     metadata: meta,
                                     parent,
                                 }))
                             }
-                            Dispatch::Fail { message } => svc::Either::B(message),
+                            Dispatch::Fail { message } => svc::Either::Right(message),
                         })
                     },
                     svc::stack(fail).check_new_clone().into_inner(),

linkerd/app/outbound/src/protocol.rs

@@ -77,14 +77,14 @@ impl<N> Outbound<N> {
                 |(detected, parent): (http::Detection, T)| -> Result<_, Infallible> {
                     match detected {
                         http::Detection::Http(version) => {
-                            return Ok(svc::Either::A(Http { version, parent }));
+                            return Ok(svc::Either::Left(Http { version, parent }));
                         }
                         http::Detection::ReadTimeout(timeout) => {
                             tracing::info!("Continuing after timeout: {timeout:?}");
                         }
                         _ => {}
                     }
-                    Ok(svc::Either::B(parent))
+                    Ok(svc::Either::Right(parent))
                 },
                 opaq.clone().into_inner(),
             )
@@ -118,21 +118,21 @@ impl<N> Outbound<N> {
                 .push_switch(
                     |parent: T| -> Result<_, Infallible> {
                         match parent.param() {
-                            Protocol::Http1 => Ok(svc::Either::A(svc::Either::A(Http {
+                            Protocol::Http1 => Ok(svc::Either::Left(svc::Either::Left(Http {
                                 version: http::Variant::Http1,
                                 parent,
                             }))),
-                            Protocol::Http2 => Ok(svc::Either::A(svc::Either::A(Http {
+                            Protocol::Http2 => Ok(svc::Either::Left(svc::Either::Left(Http {
                                 version: http::Variant::H2,
                                 parent,
                             }))),
-                            Protocol::Opaque => {
-                                Ok(svc::Either::A(svc::Either::B(svc::Either::A(parent))))
-                            }
-                            Protocol::Tls => {
-                                Ok(svc::Either::A(svc::Either::B(svc::Either::B(parent))))
-                            }
-                            Protocol::Detect => Ok(svc::Either::B(parent)),
+                            Protocol::Opaque => Ok(svc::Either::Left(svc::Either::Right(
+                                svc::Either::Left(parent),
+                            ))),
+                            Protocol::Tls => Ok(svc::Either::Left(svc::Either::Right(
+                                svc::Either::Right(parent),
+                            ))),
+                            Protocol::Detect => Ok(svc::Either::Right(parent)),
                         }
                     },
                     detect.into_inner(),

linkerd/app/outbound/src/tls/concrete.rs

@@ -195,7 +195,7 @@ impl<C> Outbound<C> {
                     move |parent: T| -> Result<_, Infallible> {
                         Ok(match parent.param() {
                             Dispatch::Balance(concrete, ewma) => {
-                                svc::Either::A(svc::Either::A(Balance {
+                                svc::Either::Left(svc::Either::Left(Balance {
                                     concrete,
                                     ewma,
                                     queue,
@@ -204,14 +204,14 @@ impl<C> Outbound<C> {
                             }
 
                             Dispatch::Forward(addr, meta) => {
-                                svc::Either::A(svc::Either::B(Endpoint {
+                                svc::Either::Left(svc::Either::Right(Endpoint {
                                     addr,
                                     is_local: false,
                                     metadata: meta,
                                     parent,
                                 }))
                             }
-                            Dispatch::Fail { message } => svc::Either::B(message),
+                            Dispatch::Fail { message } => svc::Either::Right(message),
                         })
                     },
                     svc::stack(fail).check_new_clone().into_inner(),

linkerd/retry/src/lib.rs

@@ -1,21 +1,25 @@
 #![deny(rust_2018_idioms, clippy::disallowed_methods, clippy::disallowed_types)]
 #![forbid(unsafe_code)]
 
-use futures::future;
+use futures::future::{self, Either};
 use linkerd_error::{Error, Result};
 use linkerd_stack::{
     layer::{self, Layer},
     proxy::Proxy,
     util::AndThen,
-    Either, NewService, Service,
+    NewService, Service,
 };
 use std::{
     future::Future,
     task::{Context, Poll},
 };
-pub use tower::retry::{budget::Budget, Policy};
 use tracing::trace;
 
+pub use tower::retry::{
+    budget::{Budget, TpsBudget},
+    Policy,
+};
+
 /// A strategy for obtaining per-target retry polices.
 pub trait NewPolicy<T> {
     type Policy;
@@ -26,7 +30,7 @@ pub trait NewPolicy<T> {
 /// An extension to [`tower::retry::Policy`] that adds a method to prepare a
 /// request to be retried, possibly changing its type.
 pub trait PrepareRetry<Req, Rsp>:
-    tower::retry::Policy<Self::RetryRequest, Self::RetryResponse, Error>
+    Sized + tower::retry::Policy<Self::RetryRequest, Self::RetryResponse, Error>
 {
     /// A request type that can be retried.
     ///
@@ -48,8 +52,8 @@ pub trait PrepareRetry<Req, Rsp>:
 
     /// Prepare an initial request for a potential retry.
     ///
-    /// If the request is retryable, this should return `Either::A`. Otherwise,
-    /// if this returns `Either::B`, the request will not be retried if it
+    /// If the request is retryable, this should return `Either::Left`. Otherwise,
+    /// if this returns `Either::Right`, the request will not be retried if it
     /// fails.
     ///
     /// If retrying requires a specific request type other than the input type
@@ -152,8 +156,8 @@ where
     fn call(&mut self, req: Req) -> Self::Future {
         let (policy, req) = match self.policy.clone() {
             Some(p) => match p.prepare_request(req) {
-                Either::A(req) => req,
-                Either::B(req) => {
+                Either::Left(req) => req,
+                Either::Right(req) => {
                     return future::Either::Left(self.proxy.proxy(&mut self.inner, req))
                 }
             },

linkerd/service-profiles/src/http.rs

@@ -7,7 +7,7 @@ use std::{
     sync::Arc,
     time::Duration,
 };
-use tower::retry::budget::Budget;
+use tower::retry::budget::TpsBudget;
 
 pub use self::proxy::NewProxyRouter;
 
@@ -56,7 +56,7 @@ pub enum ResponseMatch {
 
 #[derive(Clone, Debug)]
 pub struct Retries {
-    budget: Arc<Budget>,
+    budget: Arc<TpsBudget>,
 }
 
 #[derive(Clone, Default)]
@@ -107,7 +107,7 @@ impl Route {
         self.timeout
     }
 
-    pub fn set_retries(&mut self, budget: Arc<Budget>) {
+    pub fn set_retries(&mut self, budget: Arc<TpsBudget>) {
         self.retries = Some(Retries { budget });
     }
 
@@ -201,7 +201,7 @@ impl ResponseMatch {
 // === impl Retries ===
 
 impl Retries {
-    pub fn budget(&self) -> &Arc<Budget> {
+    pub fn budget(&self) -> &Arc<TpsBudget> {
         &self.budget
     }
 }

linkerd/service-profiles/src/proto.rs

@@ -5,7 +5,7 @@ use linkerd_dns_name::Name;
 use linkerd_proxy_api_resolve::pb as resolve;
 use regex::Regex;
 use std::{str::FromStr, sync::Arc, time::Duration};
-use tower::retry::budget::Budget;
+use tower::retry::budget::TpsBudget;
 use tracing::warn;
 
 pub(super) fn convert_profile(proto: api::DestinationProfile, port: u16) -> Profile {
@@ -36,7 +36,7 @@ pub(super) fn convert_profile(proto: api::DestinationProfile, port: u16) -> Prof
 
 fn convert_route(
     orig: api::Route,
-    retry_budget: Option<&Arc<Budget>>,
+    retry_budget: Option<&Arc<TpsBudget>>,
 ) -> Option<(http::RequestMatch, http::Route)> {
     let req_match = orig.condition.and_then(convert_req_match)?;
     let rsp_classes = orig
@@ -65,7 +65,7 @@ fn convert_dst_override(orig: api::WeightedDst) -> Option<Target> {
     })
 }
 
-fn set_route_retry(route: &mut http::Route, retry_budget: Option<&Arc<Budget>>) {
+fn set_route_retry(route: &mut http::Route, retry_budget: Option<&Arc<TpsBudget>>) {
     let budget = match retry_budget {
         Some(budget) => budget.clone(),
         None => {
@@ -170,7 +170,7 @@ fn convert_rsp_match(orig: api::ResponseMatch) -> Option<http::ResponseMatch> {
     Some(m)
 }
 
-fn convert_retry_budget(orig: api::RetryBudget) -> Option<Arc<Budget>> {
+fn convert_retry_budget(orig: api::RetryBudget) -> Option<Arc<TpsBudget>> {
     let min_retries = if orig.min_retries_per_second <= i32::MAX as u32 {
         orig.min_retries_per_second
     } else {
@@ -217,7 +217,7 @@ fn convert_retry_budget(orig: api::RetryBudget) -> Option<Arc<Budget>> {
         }
     };
 
-    Some(Arc::new(Budget::new(ttl, min_retries, retry_ratio)))
+    Some(Arc::new(TpsBudget::new(ttl, min_retries, retry_ratio)))
 }
 
 #[cfg(test)]

linkerd/stack/src/either.rs

@@ -1,5 +1,8 @@
 use crate::{layer, NewService};
-pub use tower::util::Either;
+
+// pub use tower::util::Either;
+pub use self::vendor::Either;
+mod vendor;
 
 #[derive(Clone, Debug)]
 pub struct NewEither<L, R> {
@@ -30,8 +33,8 @@ where
 
     fn new_service(&self, target: Either<T, U>) -> Self::Service {
         match target {
-            Either::A(t) => Either::A(self.left.new_service(t)),
-            Either::B(t) => Either::B(self.right.new_service(t)),
+            Either::Left(t) => Either::Left(self.left.new_service(t)),
+            Either::Right(t) => Either::Right(self.right.new_service(t)),
         }
     }
 }
@@ -47,8 +50,8 @@ where
 
     fn new_service(&self, target: T) -> Self::Service {
         match self {
-            Either::A(n) => Either::A(n.new_service(target)),
-            Either::B(n) => Either::B(n.new_service(target)),
+            Either::Left(n) => Either::Left(n.new_service(target)),
+            Either::Right(n) => Either::Right(n.new_service(target)),
         }
     }
 }

linkerd/stack/src/either/vendor.rs

@@ -0,0 +1,101 @@
+//! Contains [`Either`] and related types and functions.
+//!
+//! See [`Either`] documentation for more details.
+//!
+//! TODO(kate): this is a lightly modified variant of `tower`'s `Either` service.
+//!
+//! This is pulled in-tree to punt on addressing breaking changes to the trait bounds of
+//! `Either<A, B>`'s `Service` implementation, related to how it no longer boxes the errors
+//! returned by its inner services. see #3744.
+//!
+//! This is vendored from <https://github.com/tower-rs/tower/commit/8b84b98d93a2493422a0ecddb6251f292a904cff>.
+//!
+//! The variants `A` and `B` have been renamed to `Left` and `Right` to match the names of the v0.5
+//! interface.
+
+use futures::ready;
+use linkerd_error::Error;
+use pin_project::pin_project;
+use std::{
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
+use tower::Layer;
+use tower::Service;
+
+/// Combine two different service types into a single type.
+///
+/// Both services must be of the same request, response, and error types.
+/// [`Either`] is useful for handling conditional branching in service middleware
+/// to different inner service types.
+#[pin_project(project = EitherProj)]
+#[derive(Clone, Debug)]
+pub enum Either<A, B> {
+    /// One type of backing [`Service`].
+    Left(#[pin] A),
+    /// The other type of backing [`Service`].
+    Right(#[pin] B),
+}
+
+impl<A, B, Request> Service<Request> for Either<A, B>
+where
+    A: Service<Request>,
+    A::Error: Into<Error>,
+    B: Service<Request, Response = A::Response>,
+    B::Error: Into<Error>,
+{
+    type Response = A::Response;
+    type Error = Error;
+    type Future = Either<A::Future, B::Future>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        use self::Either::*;
+
+        match self {
+            Left(service) => Poll::Ready(Ok(ready!(service.poll_ready(cx)).map_err(Into::into)?)),
+            Right(service) => Poll::Ready(Ok(ready!(service.poll_ready(cx)).map_err(Into::into)?)),
+        }
+    }
+
+    fn call(&mut self, request: Request) -> Self::Future {
+        use self::Either::*;
+
+        match self {
+            Left(service) => Left(service.call(request)),
+            Right(service) => Right(service.call(request)),
+        }
+    }
+}
+
+impl<A, B, T, AE, BE> Future for Either<A, B>
+where
+    A: Future<Output = Result<T, AE>>,
+    AE: Into<Error>,
+    B: Future<Output = Result<T, BE>>,
+    BE: Into<Error>,
+{
+    type Output = Result<T, Error>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        match self.project() {
+            EitherProj::Left(fut) => Poll::Ready(Ok(ready!(fut.poll(cx)).map_err(Into::into)?)),
+            EitherProj::Right(fut) => Poll::Ready(Ok(ready!(fut.poll(cx)).map_err(Into::into)?)),
+        }
+    }
+}
+
+impl<S, A, B> Layer<S> for Either<A, B>
+where
+    A: Layer<S>,
+    B: Layer<S>,
+{
+    type Service = Either<A::Service, B::Service>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        match self {
+            Either::Left(layer) => Either::Left(layer.layer(inner)),
+            Either::Right(layer) => Either::Right(layer.layer(inner)),
+        }
+    }
+}

linkerd/stack/src/queue.rs

@@ -34,9 +34,11 @@ pub struct NewQueueWithoutTimeout<X, Req, N> {
     _req: PhantomData<fn(Req)>,
 }
 
-pub type Queue<Req, Rsp, E = Error> = gate::Gate<Buffer<BoxService<Req, Rsp, E>, Req>>;
+pub type Queue<Req, Rsp, E = Error> =
+    gate::Gate<Buffer<Req, <BoxService<Req, Rsp, E> as tower::Service<Req>>::Future>>;
 
-pub type QueueWithoutTimeout<Req, Rsp, E = Error> = Buffer<BoxService<Req, Rsp, E>, Req>;
+pub type QueueWithoutTimeout<Req, Rsp, E = Error> =
+    Buffer<Req, <BoxService<Req, Rsp, E> as tower::Service<Req>>::Future>;
 
 // === impl NewQueue ===
 

linkerd/stack/src/unwrap_or.rs

@@ -26,8 +26,8 @@ impl<T, U> Predicate<(Option<T>, U)> for UnwrapOr<U> {
 
     fn check(&mut self, (t, u): (Option<T>, U)) -> Result<Either<(T, U), U>, Error> {
         match t {
-            Some(t) => Ok(Either::A((t, u))),
-            None => Ok(Either::B(u)),
+            Some(t) => Ok(Either::Left((t, u))),
+            None => Ok(Either::Right(u)),
         }
     }
 }
@@ -36,7 +36,8 @@ impl<T, U: Default> Predicate<Option<T>> for UnwrapOr<U> {
     type Request = Either<T, U>;
 
     fn check(&mut self, t: Option<T>) -> Result<Either<T, U>, Error> {
-        Ok(t.map(Either::A).unwrap_or_else(|| Either::B(U::default())))
+        Ok(t.map(Either::Left)
+            .unwrap_or_else(|| Either::Right(U::default())))
     }
 }