fix: don't crash the derper (#1110)

* add more debugging to TLS acceptor
* fix(derp) client conn: ensure flushing on shutdown
* `ClientConn` task should close on EOF

---------

Co-authored-by: Kasey <klhuizinga@gmail.com>

Author: dignifiedquire

src/hp/derp/client_conn.rs

@@ -294,11 +294,16 @@ where
         loop {
             trace!("tick");
             tokio::select! {
+                biased;
+
                 _ = done.cancelled() => {
                     trace!("cancelled");
+                    // final flush
+                    self.io.flush().await?;
                     return Ok(());
                 }
                 read_res = read_frame(&mut self.io, MAX_FRAME_SIZE, &mut read_buf) => {
+                    trace!("handle read");
                     self.handle_read(read_res, &mut read_buf).await?;
                 }
                 peer = self.peer_gone.recv() => {
@@ -496,14 +501,7 @@ where
                 }
                 Ok(())
             }
-            Err(err) => {
-                if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
-                    if io_err.kind() == std::io::ErrorKind::UnexpectedEof {
-                        return Ok(());
-                    }
-                }
-                Err(err)
-            }
+            Err(err) => Err(err),
         }
     }
 

src/hp/derp/http/server.rs

@@ -365,21 +365,26 @@ where
                 // Note: This can't possibly be fulfilled until the 101 response
                 // is returned below, so it's better to spawn this future instead
                 // waiting for it to complete to then return a response.
-                tokio::task::spawn(async move {
-                    match hyper::upgrade::on(&mut req).await {
-                        Ok(upgraded) => {
-                            if let Err(e) =
-                                derp_connection_handler(&closure_conn_handler, upgraded).await
-                            {
-                                tracing::warn!(
-                                    "server \"{HTTP_UPGRADE_PROTOCOL}\" io error: {:?}",
-                                    e
-                                )
-                            };
+                tokio::task::spawn(
+                    async move {
+                        match hyper::upgrade::on(&mut req).await {
+                            Ok(upgraded) => {
+                                if let Err(e) =
+                                    derp_connection_handler(&closure_conn_handler, upgraded).await
+                                {
+                                    tracing::warn!(
+                                        "server \"{HTTP_UPGRADE_PROTOCOL}\" io error: {:?}",
+                                        e
+                                    );
+                                } else {
+                                    tracing::info!("server \"{HTTP_UPGRADE_PROTOCOL}\" success");
+                                };
+                            }
+                            Err(e) => tracing::warn!("upgrade error: {:?}", e),
                         }
-                        Err(e) => tracing::warn!("upgrade error: {:?}", e),
                     }
-                });
+                    .instrument(tracing::debug_span!("derp_connection_handler")),
+                );
 
                 // Now return a 101 Response saying we agree to the upgrade to the
                 // HTTP_UPGRADE_PROTOCOL
@@ -514,6 +519,7 @@ impl DerpService {
         match tls_config {
             Some(tls_config) => self.tls_serve_connection(stream, tls_config).await,
             None => {
+                debug!("HTTP: serve connection");
                 self.serve_connection(MaybeTlsStreamServer::Plain(stream))
                     .await
             }
@@ -526,18 +532,25 @@ impl DerpService {
         match acceptor {
             TlsAcceptor::LetsEncrypt(a) => match a.accept(stream).await? {
                 None => {
-                    info!("received TLS-ALPN-01 validation request");
+                    info!("TLS[acme]: received TLS-ALPN-01 validation request");
                 }
                 Some(start_handshake) => {
-                    let tls_stream = start_handshake.into_stream(config).await?;
+                    debug!("TLS[acme]: start handshake");
+                    let tls_stream = start_handshake
+                        .into_stream(config)
+                        .await
+                        .context("TLS[acme] handshake")?;
                     self.serve_connection(MaybeTlsStreamServer::Tls(tls_stream))
-                        .await?;
+                        .await
+                        .context("TLS[acme] serve connection")?;
                 }
             },
             TlsAcceptor::Manual(a) => {
-                let tls_stream = a.accept(stream).await?;
+                debug!("TLS[manual]: accept");
+                let tls_stream = a.accept(stream).await.context("TLS[manual] accept")?;
                 self.serve_connection(MaybeTlsStreamServer::Tls(tls_stream))
-                    .await?;
+                    .await
+                    .context("TLS[manual] serve connection")?;
             }
         }
         Ok(())