fuzz: don't panic when the proxy closes the conn (#986)

Currently, the inbound HTTP fuzz tests are failing. This is because they
reuse the test-support code for making an HTTP request to a proxy stack
and running the futures necessary to drive that request in the
background. This code currently unwraps both the `JoinHandle`s of the
spawned tasks (which would be `Err` if the task panicked) _and_ the
returned `Result` from those `JoinHandle`s (which is an `Err` if the
`Service::call` future returned an error, or if the client returned an
error). If the future completes with an error, then the proxy simply
tears down the connection.

In the integration tests, we currently `expect` _both_ of these
`Results` --- since the inputs are valid, we want to assert the proxy
doesn't return an error incorrectly. However, the fuzz tests can and
will generate malformed HTTP requests, and in this case, the proxy will
reject those requests by returning an error and closing the connection.
This is *not* incorrect behavior. Instead, we want to ensure that the
proxy doesn't panic in the face of potentially malformed requests.

This branch changes the test-support HTTP code to return the `Result` of
serving a request, and unwrap it in the integration tests rather than in
the support code. The fuzz logic is updated to simply log errors
returned here, since returning an error is expected behavior when we
receive invalid inputs.

If we wanted to be _really_ fancy, a potential follow-up would be to
extend the fuzz logic to determine whether or not a fuzz spec should
result in an error, and assert that errors are only returned for invalid
requests...but, doing this without using any of the code that's being
exercised in the fuzz test (e.g. all of `hyper`'s request parsing etc)
would be a _lot_ of work...

Depends on #985

Signed-off-by: Eliza Weisman <eliza@buoyant.io>

Author: hawkw

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

@@ -323,6 +323,7 @@ pub mod fuzz_logic {
                             .body(Body::default())
                         {
                             let rsp = http_util::http_request(&mut client, req).await;
+                            tracing::info!(?rsp);
                             let _body = http_util::body_to_string(rsp.into_body()).await;
                         }
                     }
@@ -331,7 +332,13 @@ pub mod fuzz_logic {
         }
 
         drop(client);
-        bg.await;
+        // It's okay if the background task returns an error, as this would
+        // indcate that the proxy closed the connection --- which it will do on
+        // invalid inputs. We want to ensure that the proxy doesn't crash in the
+        // face of these inputs, and the background task will panic in this
+        // case.
+        let res = bg.await;
+        tracing::info!(?res, "background tasks completed")
     }
 
     fn hello_fuzz_server(

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

@@ -87,7 +87,7 @@ async fn unmeshed_http1_hello_world() {
     assert_eq!(body, "Hello world!");
 
     drop(client);
-    bg.await;
+    bg.await.expect("background task failed");
 }
 
 #[tokio::test(flavor = "current_thread")]
@@ -136,7 +136,7 @@ async fn downgrade_origin_form() {
     assert_eq!(body, "Hello world!");
 
     drop(client);
-    bg.await;
+    bg.await.expect("background task failed");
 }
 
 #[tokio::test(flavor = "current_thread")]
@@ -184,7 +184,7 @@ async fn downgrade_absolute_form() {
     assert_eq!(body, "Hello world!");
 
     drop(client);
-    bg.await;
+    bg.await.expect("background task failed");
 }
 
 #[tracing::instrument]

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

@@ -278,7 +278,7 @@ async fn meshed_hello_world() {
     assert_eq!(body, "Hello world!");
 
     drop(client);
-    bg.await;
+    bg.await.expect("background task failed");
 }
 
 #[tokio::test(flavor = "current_thread")]
@@ -338,7 +338,7 @@ async fn stacks_idle_out() {
     assert_eq!(body, "Hello world!");
 
     drop(client);
-    bg.await;
+    bg.await.expect("background task failed");
 
     assert_eq!(handle.tracked_services(), 1);
     // wait for long enough to ensure that it _definitely_ idles out...
@@ -438,8 +438,14 @@ async fn active_stacks_dont_idle_out() {
     tokio::time::sleep(idle_timeout * 2).await;
     assert_eq!(handle.tracked_services(), 0);
 
-    client_bg.await.unwrap();
-    proxy_bg.await.unwrap();
+    client_bg
+        .await
+        .unwrap()
+        .expect("client background task failed");
+    proxy_bg
+        .await
+        .unwrap()
+        .expect("proxy background task failed");
 }
 
 async fn unmeshed_hello_world(
@@ -470,7 +476,7 @@ async fn unmeshed_hello_world(
     assert_eq!(body, "Hello world!");
 
     drop(client);
-    bg.await;
+    bg.await.expect("background task failed");
 }
 
 #[tracing::instrument]

linkerd/app/test/src/http_util.rs

@@ -36,7 +36,7 @@ impl Default for Server {
     }
 }
 
-pub async fn run_proxy<S>(mut server: S) -> (io::DuplexStream, JoinHandle<()>)
+pub async fn run_proxy<S>(mut server: S) -> (io::DuplexStream, JoinHandle<Result<(), Error>>)
 where
     S: tower::Service<io::DuplexStream> + Send + Sync + 'static,
     S::Error: Into<Error>,
@@ -56,7 +56,7 @@ where
         drop(server);
         tracing::debug!("dropped server");
         tracing::info!(?res, "proxy serve task complete");
-        res.expect("proxy failed");
+        res.map(|_| ())
     }
     .instrument(tracing::info_span!("proxy"));
     (client_io, tokio::spawn(proxy))
@@ -65,15 +65,15 @@ where
 pub async fn connect_client(
     client_settings: &mut ClientBuilder,
     io: io::DuplexStream,
-) -> (SendRequest<Body>, JoinHandle<()>) {
+) -> (SendRequest<Body>, JoinHandle<Result<(), Error>>) {
     let (client, conn) = client_settings
         .handshake(io)
         .await
         .expect("Client must connect");
     let client_bg = conn
         .map(|res| {
             tracing::info!(?res, "Client background complete");
-            res.expect("client bg task failed");
+            res.map_err(Into::into)
         })
         .instrument(tracing::info_span!("client_bg"));
     (client, tokio::spawn(client_bg))
@@ -82,7 +82,7 @@ pub async fn connect_client(
 pub async fn connect_and_accept<S>(
     client_settings: &mut ClientBuilder,
     server: S,
-) -> (SendRequest<Body>, impl Future<Output = ()>)
+) -> (SendRequest<Body>, impl Future<Output = Result<(), Error>>)
 where
     S: tower::Service<io::DuplexStream> + Send + Sync + 'static,
     S::Error: Into<Error>,
@@ -93,11 +93,9 @@ where
     let (client_io, proxy) = run_proxy(server).await;
     let (client, client_bg) = connect_client(client_settings, client_io).await;
     let bg = async move {
-        let res = tokio::try_join! {
-            proxy,
-            client_bg,
-        };
-        res.unwrap();
+        proxy.await.expect("proxy background task panicked")?;
+        client_bg.await.expect("client background task panicked")?;
+        Ok(())
     };
     (client, bg)
 }