stack: Add BoxCloneSyncService

linkerd/app/core/src/control.rs

@@ -71,12 +71,7 @@ impl Config {
         identity: identity::NewClient,
     ) -> svc::ArcNewService<
         (),
-        impl svc::Service<
-                http::Request<tonic::body::BoxBody>,
-                Response = http::Response<RspBody>,
-                Error = ControlError,
-                Future = impl Send,
-            > + Clone,
+        svc::BoxCloneSyncService<http::Request<tonic::body::BoxBody>, http::Response<RspBody>>,
     > {
         let addr = self.addr;
 
@@ -135,6 +130,7 @@ impl Config {
             .push(svc::NewMapErr::layer_from_target::<ControlError, _>())
             .instrument(|c: &ControlAddr| info_span!("controller", addr = %c.addr))
             .push_map_target(move |()| addr.clone())
+            .push_on_service(svc::BoxCloneSyncService::layer())
             .push(svc::ArcNewService::layer())
             .into_inner()
     }

linkerd/app/core/src/svc.rs

@@ -34,15 +34,15 @@ pub type BoxHttp<B = http::BoxBody> =
 pub type ArcNewHttp<T, B = http::BoxBody> = ArcNewService<T, BoxHttp<B>>;
 
 pub type BoxCloneHttp<B = http::BoxBody> =
-    BoxCloneService<http::Request<B>, http::Response<http::BoxBody>, Error>;
+    BoxCloneSyncService<http::Request<B>, http::Response<http::BoxBody>>;
 
 pub type ArcNewCloneHttp<T, B = http::BoxBody> = ArcNewService<T, BoxCloneHttp<B>>;
 
 pub type BoxTcp<I> = BoxService<I, (), Error>;
 
 pub type ArcNewTcp<T, I> = ArcNewService<T, BoxTcp<I>>;
 
-pub type BoxCloneTcp<I> = BoxCloneService<I, (), Error>;
+pub type BoxCloneTcp<I> = BoxCloneSyncService<I, ()>;
 
 pub type ArcNewCloneTcp<T, I> = ArcNewService<T, BoxCloneTcp<I>>;
 
@@ -302,11 +302,25 @@ impl<S> Stack<S> {
         T: 'static,
         B: 'static,
         S: NewService<T, Service = Svc> + Send + Sync + 'static,
-        Svc: Service<http::Request<B>, Response = http::Response<http::BoxBody>, Error = Error>,
-        Svc: Clone + Send + 'static,
+        Svc: Service<http::Request<B>, Response = http::Response<http::BoxBody>>,
+        Svc: Clone + Send + Sync + 'static,
+        Svc::Error: Into<Error>,
+        Svc::Future: Send,
+    {
+        self.push_on_service(BoxCloneHttp::layer())
+            .push(ArcNewService::layer())
+    }
+
+    pub fn arc_new_clone_tcp<T, I, Svc>(self) -> Stack<ArcNewCloneTcp<T, I>>
+    where
+        T: 'static,
+        I: 'static,
+        S: NewService<T, Service = Svc> + Send + Sync + 'static,
+        Svc: Service<I, Response = ()> + Clone + Send + Sync + 'static,
+        Svc::Error: Into<Error>,
         Svc::Future: Send,
     {
-        self.push_on_service(BoxCloneService::layer())
+        self.push_on_service(BoxCloneTcp::layer())
             .push(ArcNewService::layer())
     }
 

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

@@ -74,7 +74,7 @@ impl<C> Inbound<C> {
             + Param<Remote<ClientAddr>>
             + Param<tls::ConditionalServerTls>
             + Param<policy::AllowPolicy>,
-        T: Clone + Send + Unpin + 'static,
+        T: Clone + Send + Sync + Unpin + 'static,
         P: profiles::GetProfile<Error = Error>,
         C: svc::MakeConnection<Http> + Clone + Send + Sync + Unpin + 'static,
         C::Connection: Send + Unpin,

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

@@ -43,6 +43,7 @@ impl<H> Inbound<H> {
         HSvc: svc::Service<http::Request<http::BoxBody>, Response = http::Response<http::BoxBody>>
             + Clone
             + Send
+            + Sync
             + Unpin
             + 'static,
         HSvc::Error: Into<Error>,

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

@@ -79,20 +79,7 @@ impl<N> Outbound<N> {
     /// 'failfast'. While in failfast, buffered requests are failed and the
     /// service becomes unavailable so callers may choose alternate concrete
     /// services.
-    pub fn push_http_concrete<T, NSvc, R>(
-        self,
-        resolve: R,
-    ) -> Outbound<
-        svc::ArcNewService<
-            T,
-            impl svc::Service<
-                    http::Request<http::BoxBody>,
-                    Response = http::Response<http::BoxBody>,
-                    Error = Error,
-                    Future = impl Send,
-                > + Clone,
-        >,
-    >
+    pub fn push_http_concrete<T, NSvc, R>(self, resolve: R) -> Outbound<svc::ArcNewCloneHttp<T>>
     where
         // Concrete target type.
         T: svc::Param<ParentRef>,
@@ -157,7 +144,7 @@ impl<N> Outbound<N> {
                 // TODO(ver) Configure this queue from the target (i.e. from
                 // discovery).
                 .push(svc::NewQueue::layer_via(config.http_request_queue))
-                .push(svc::ArcNewService::layer())
+                .arc_new_clone_http()
         })
     }
 }
@@ -182,20 +169,7 @@ where
         config: &crate::Config,
         rt: &crate::Runtime,
         resolve: R,
-    ) -> impl svc::Layer<
-        N,
-        Service = svc::ArcNewService<
-            Self,
-            impl svc::Service<
-                http::Request<http::BoxBody>,
-                Response = http::Response<http::BoxBody>,
-                Error = BalanceError,
-                Future = impl std::future::Future<
-                    Output = Result<http::Response<http::BoxBody>, BalanceError>,
-                > + Send,
-            >,
-        >,
-    > + Clone
+    ) -> impl svc::Layer<N, Service = svc::ArcNewHttp<Self>> + Clone
     where
         // Endpoint resolution.
         R: Resolve<ConcreteAddr, Error = Error, Endpoint = Metadata> + 'static,
@@ -271,7 +245,8 @@ where
                         port = %meta.port().map(u16::from).unwrap_or(0),
                     )
                 })
-                .push(svc::ArcNewService::layer())
+                .push_on_service(svc::MapErr::layer_boxed())
+                .arc_new_http()
                 .into_inner()
         })
     }

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

@@ -81,19 +81,7 @@ impl<N> Outbound<N> {
     /// support per-request routing over a set of concrete inner services.
     /// Only available inner services are used for routing. When there are no
     /// available backends, requests are failed with a [`svc::stack::LoadShedError`].
-    pub fn push_http_logical<T, NSvc>(
-        self,
-    ) -> Outbound<
-        svc::ArcNewService<
-            T,
-            impl svc::Service<
-                    http::Request<http::BoxBody>,
-                    Response = http::Response<http::BoxBody>,
-                    Error = Error,
-                    Future = impl Send,
-                > + Clone,
-        >,
-    >
+    pub fn push_http_logical<T, NSvc>(self) -> Outbound<svc::ArcNewCloneHttp<T>>
     where
         // Logical target.
         T: svc::Param<watch::Receiver<Routes>>,
@@ -111,16 +99,13 @@ impl<N> Outbound<N> {
         self.map_stack(|_config, rt, concrete| {
             // For each `T` target, watch its `Profile`, rebuilding a
             // router stack.
-            let watch = concrete
+            concrete
                 // Share the concrete stack with each router stack.
                 .lift_new()
                 .push_on_service(RouterParams::layer(rt.metrics.clone()))
                 // Rebuild the inner router stack every time the watch changes.
-                .push(svc::NewSpawnWatch::<Routes, _>::layer_into::<RouterParams<T>>());
-
-            watch
-                .push_on_service(svc::MapErr::layer_boxed())
-                .push(svc::ArcNewService::layer())
+                .push(svc::NewSpawnWatch::<Routes, _>::layer_into::<RouterParams<T>>())
+                .arc_new_clone_http()
         })
     }
 }
@@ -133,18 +118,7 @@ where
 {
     fn layer<N, S>(
         metrics: OutboundMetrics,
-    ) -> impl svc::Layer<
-        N,
-        Service = svc::ArcNewService<
-            RouterParams<T>,
-            impl svc::Service<
-                    http::Request<http::BoxBody>,
-                    Response = http::Response<http::BoxBody>,
-                    Error = Error,
-                    Future = impl Send,
-                > + Clone,
-        >,
-    > + Clone
+    ) -> impl svc::Layer<N, Service = svc::ArcNewCloneHttp<RouterParams<T>>> + Clone
     where
         N: svc::NewService<Concrete<T>, Service = S>,
         N: Clone + Send + Sync + 'static,
@@ -191,8 +165,7 @@ where
                         .into_inner(),
                 )
                 .push(svc::NewMapErr::layer_from_target::<LogicalError, _>())
-                .push_on_service(svc::MapErr::layer_boxed())
-                .push(svc::ArcNewService::layer())
+                .arc_new_clone_http()
                 .into_inner()
         })
     }

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

@@ -51,18 +51,7 @@ where
     /// services.
     pub(super) fn layer<N, S>(
         route_backend_metrics: RouteBackendMetrics,
-    ) -> impl svc::Layer<
-        N,
-        Service = svc::ArcNewService<
-            Self,
-            impl svc::Service<
-                    http::Request<http::BoxBody>,
-                    Response = http::Response<http::BoxBody>,
-                    Error = Error,
-                    Future = impl Send,
-                > + Clone,
-        >,
-    > + Clone
+    ) -> impl svc::Layer<N, Service = svc::ArcNewCloneHttp<Self>> + Clone
     where
         // Inner stack.
         N: svc::NewService<Concrete<T>, Service = S>,
@@ -89,7 +78,7 @@ where
                 },
                 grpc.into_inner(),
             )
-            .push(svc::ArcNewService::layer())
+            .arc_new_clone_http()
             .into_inner()
         })
     }

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

@@ -84,18 +84,7 @@ where
     /// backends are expected to be cached/shared by the inner stack.
     pub(crate) fn layer<N, S>(
         backend_metrics: backend::RouteBackendMetrics,
-    ) -> impl svc::Layer<
-        N,
-        Service = svc::ArcNewService<
-            Self,
-            impl svc::Service<
-                    http::Request<http::BoxBody>,
-                    Response = http::Response<http::BoxBody>,
-                    Error = Error,
-                    Future = impl Send,
-                > + Clone,
-        >,
-    > + Clone
+    ) -> impl svc::Layer<N, Service = svc::ArcNewCloneHttp<Self>> + Clone
     where
         // Inner stack.
         N: svc::NewService<Concrete<T>, Service = S>,
@@ -126,14 +115,12 @@ where
                 .push(classify::NewClassify::layer())
                 .push(svc::NewMapErr::layer_with(|rt: &Self| {
                     let route = rt.params.route_ref.clone();
-                    move |source| {
-                        Error::from(RouteError {
-                            route: route.clone(),
-                            source,
-                        })
+                    move |source| RouteError {
+                        route: route.clone(),
+                        source,
                     }
                 }))
-                .push(svc::ArcNewService::layer())
+                .arc_new_clone_http()
                 .into_inner()
         })
     }

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

@@ -85,18 +85,7 @@ where
     /// filters.
     pub(crate) fn layer<N, S>(
         metrics: RouteBackendMetrics,
-    ) -> impl svc::Layer<
-        N,
-        Service = svc::ArcNewService<
-            Self,
-            impl svc::Service<
-                    http::Request<http::BoxBody>,
-                    Response = http::Response<http::BoxBody>,
-                    Error = Error,
-                    Future = impl Send,
-                > + Clone,
-        >,
-    > + Clone
+    ) -> impl svc::Layer<N, Service = svc::ArcNewCloneHttp<Self>> + Clone
     where
         // Inner stack.
         N: svc::NewService<Concrete<T>, Service = S>,
@@ -131,7 +120,7 @@ where
                         })
                     }
                 }))
-                .push(svc::ArcNewService::layer())
+                .arc_new_clone_http()
                 .into_inner()
         })
     }

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

@@ -71,18 +71,7 @@ where
     /// set of inner services so that.
     pub(super) fn layer<N, S>(
         route_backend_metrics: RouteBackendMetrics,
-    ) -> impl svc::Layer<
-        N,
-        Service = svc::ArcNewService<
-            Self,
-            impl svc::Service<
-                    http::Request<http::BoxBody>,
-                    Response = http::Response<http::BoxBody>,
-                    Error = Error,
-                    Future = impl Send,
-                > + Clone,
-        >,
-    > + Clone
+    ) -> impl svc::Layer<N, Service = svc::ArcNewCloneHttp<Self>> + Clone
     where
         // Inner stack.
         N: svc::NewService<Concrete<T>, Service = S>,
@@ -105,7 +94,7 @@ where
                 // `SelectRoute` impl.
                 .push_on_service(route::MatchedRoute::layer(route_backend_metrics.clone()))
                 .push(svc::NewOneshotRoute::<Self, (), _>::layer_cached())
-                .push(svc::ArcNewService::layer())
+                .arc_new_clone_http()
                 .into_inner()
         })
     }