refactor(app/integration): remove artificial Sync bounds (#3700)

see linkerd/linkerd2#8733 for more
information.

we are in the process of upgrading to hyper 1.x.

in the process of doing so, we will wish to use our friendly `BoxBody`
type, which provides a convenient and reusable interface to abstract
over different artitrary `B`-typed request and response bodies.

unfortunately, by virtue of its definition, it is not a `Sync` type:

```rust
 pub struct BoxBody {
     inner: Pin<Box<dyn Body<Data = Data, Error = Error> + Send + 'static>>,
 }

 #[pin_project]
 pub struct Data {
     #[pin]
     inner: Box<dyn bytes::Buf + Send + 'static>,
 }
```

these are erased `Box<dyn ..>` objects that only ensure `Send`-ness.

rather than changing that, because that is the proper definition of the
type, we should update code in our test client and test server to stop
requesting arbitrary `Sync` bounds.

this commit removes `Sync` bounds from various places that in fact only
need be `Send + 'static`.

this will help facilitate making use of `BoxBody` in #3504.

Signed-off-by: katelyn martin <[email protected]>

Author: cratelyn

linkerd/app/integration/src/client.rs

@@ -131,8 +131,7 @@ impl Client {
     pub fn request(
         &self,
         builder: http::request::Builder,
-    ) -> impl Future<Output = Result<Response<hyper::Body>, ClientError>> + Send + Sync + 'static
-    {
+    ) -> impl Future<Output = Result<Response<hyper::Body>, ClientError>> + Send + 'static {
         self.send_req(builder.body(Bytes::new().into()).unwrap())
     }
 
@@ -156,8 +155,7 @@ impl Client {
     pub(crate) fn send_req(
         &self,
         mut req: Request<hyper::Body>,
-    ) -> impl Future<Output = Result<Response<hyper::Body>, ClientError>> + Send + Sync + 'static
-    {
+    ) -> impl Future<Output = Result<Response<hyper::Body>, ClientError>> + Send + 'static {
         if req.uri().scheme().is_none() {
             if self.tls.is_some() {
                 *req.uri_mut() = format!("https://{}{}", self.authority, req.uri().path())
@@ -172,7 +170,7 @@ impl Client {
         tracing::debug!(headers = ?req.headers(), "request");
         let (tx, rx) = oneshot::channel();
         let _ = self.tx.send((req.map(Into::into), tx));
-        async { rx.await.expect("request cancelled") }.in_current_span()
+        async move { rx.await.expect("request cancelled") }.in_current_span()
     }
 
     pub async fn wait_for_closed(self) {
@@ -221,7 +219,7 @@ enum Run {
     Http2,
 }
 
-pub type Running = Pin<Box<dyn Future<Output = ()> + Send + Sync + 'static>>;
+pub type Running = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
 
 fn run(
     addr: SocketAddr,

linkerd/app/integration/src/lib.rs

@@ -256,7 +256,7 @@ impl fmt::Display for HumanDuration {
 
 pub async fn cancelable<E: Send + 'static>(
     drain: drain::Watch,
-    f: impl Future<Output = Result<(), E>> + Send + 'static,
+    f: impl Future<Output = Result<(), E>>,
 ) -> Result<(), E> {
     tokio::select! {
         res = f => res,

linkerd/app/integration/src/server.rs

@@ -59,7 +59,7 @@ pub struct Listening {
 }
 
 type RspFuture =
-    Pin<Box<dyn Future<Output = Result<Response<hyper::Body>, Error>> + Send + Sync + 'static>>;
+    Pin<Box<dyn Future<Output = Result<Response<hyper::Body>, Error>> + Send + 'static>>;
 
 impl Listening {
     pub fn connections(&self) -> usize {
@@ -128,7 +128,7 @@ impl Server {
     pub fn route_async<F, U>(mut self, path: &str, cb: F) -> Self
     where
         F: Fn(Request<hyper::Body>) -> U + Send + Sync + 'static,
-        U: TryFuture<Ok = Response<hyper::Body>> + Send + Sync + 'static,
+        U: TryFuture<Ok = Response<hyper::Body>> + Send + 'static,
         U::Error: Into<Error> + Send + 'static,
     {
         let func = move |req| Box::pin(cb(req).map_err(Into::into)) as RspFuture;
@@ -219,9 +219,17 @@ impl Server {
                         tracing::trace!(?result, "serve done");
                         result
                     };
-                    tokio::spawn(
-                        cancelable(drain.clone(), f).instrument(span.clone().or_current()),
-                    );
+                    // let fut = Box::pin(cancelable(drain.clone(), f).instrument(span.clone().or_current()))
+                    let drain = drain.clone();
+                    tokio::spawn(async move {
+                        tokio::select! {
+                            res = f => res,
+                            _ = drain.signaled() => {
+                                tracing::debug!("canceled!");
+                                Ok(())
+                            }
+                        }
+                    });
                 }
             }
             .instrument(

linkerd/app/integration/src/tests/profiles.rs

@@ -293,15 +293,16 @@ mod cross_version {
             .method("POST")
             .body(body)
             .unwrap();
-        let res = tokio::spawn(async move { client.request_body(req).await });
+        let fut = client.send_req(req);
+        let res = tokio::spawn(fut);
         tx.send_data(Bytes::from_static(b"hello"))
             .await
             .expect("the whole body should be read");
         tx.send_data(Bytes::from_static(b"world"))
             .await
             .expect("the whole body should be read");
         drop(tx);
-        let res = res.await.unwrap();
+        let res = res.await.unwrap().unwrap();
         assert_eq!(res.status(), 200);
     }
 
@@ -392,7 +393,8 @@ mod cross_version {
             .method("POST")
             .body(body)
             .unwrap();
-        let res = tokio::spawn(async move { client.request_body(req).await });
+        let fut = client.send_req(req);
+        let res = tokio::spawn(fut);
         // send a 32k chunk
         tx.send_data(Bytes::from(&[1u8; 32 * 1024][..]))
             .await
@@ -406,7 +408,7 @@ mod cross_version {
             .await
             .expect("the whole body should be read");
         drop(tx);
-        let res = res.await.unwrap();
+        let res = res.await.unwrap().unwrap();
 
         assert_eq!(res.status(), 533);
     }