chore: better error logging for "failed to join reader and writer tasks" #3910 (#3913)

Signed-off-by: PeaBrane <[email protected]>

Author: PeaBrane

lib/runtime/src/pipeline/network/tcp/client.rs

@@ -84,6 +84,7 @@ impl TcpClient {
         }
 
         let stream = TcpClient::connect(&info.address).await?;
+        let peer_port = stream.peer_addr().ok().map(|addr| addr.port());
         let (read_half, write_half) = tokio::io::split(stream);
 
         let framed_reader = FramedRead::new(read_half, TwoPartCodec::default());
@@ -100,7 +101,7 @@ impl TcpClient {
 
         // transport specific handshake message
         let handshake = CallHomeHandshake {
-            subject: info.subject,
+            subject: info.subject.clone(),
             stream_type: StreamType::Response,
         };
 
@@ -127,6 +128,7 @@ impl TcpClient {
 
         let writer_task = tokio::spawn(handle_writer(framed_writer, bytes_rx, alive_rx, context));
 
+        let subject = info.subject.clone();
         tokio::spawn(async move {
             // await both tasks
             let (reader, writer) = tokio::join!(reader_task, writer_task);
@@ -166,9 +168,29 @@ impl TcpClient {
 
                     Ok(())
                 }
-                _ => {
-                    tracing::error!("failed to join reader and writer tasks");
-                    anyhow::bail!("failed to join reader and writer tasks");
+                (Err(reader_err), Ok(_)) => {
+                    tracing::error!(
+                        "reader task failed to join (peer_port: {peer_port:?}, subject: {subject}): {reader_err:?}"
+                    );
+                    anyhow::bail!(
+                        "reader task failed to join (peer_port: {peer_port:?}, subject: {subject}): {reader_err:?}"
+                    );
+                }
+                (Ok(_), Err(writer_err)) => {
+                    tracing::error!(
+                        "writer task failed to join (peer_port: {peer_port:?}, subject: {subject}): {writer_err:?}"
+                    );
+                    anyhow::bail!(
+                        "writer task failed to join (peer_port: {peer_port:?}, subject: {subject}): {writer_err:?}"
+                    );
+                }
+                (Err(reader_err), Err(writer_err)) => {
+                    tracing::error!(
+                        "both reader and writer tasks failed to join (peer_port: {peer_port:?}, subject: {subject}) - reader: {reader_err:?}, writer: {writer_err:?}"
+                    );
+                    anyhow::bail!(
+                        "both reader and writer tasks failed to join (peer_port: {peer_port:?}, subject: {subject}) - reader: {reader_err:?}, writer: {writer_err:?}"
+                    );
                 }
             }
         });
@@ -227,10 +249,10 @@ async fn handle_reader(
                            }
                         }
                     }
-                    Some(Err(_)) => {
+                    Some(Err(e)) => {
                         // TODO(#171) - address fatal errors
                         // in this case the binary representation of the message is invalid
-                        panic!("fatal error - failed to decode message from stream; invalid line protocol");
+                        panic!("fatal error - failed to decode message from stream; invalid line protocol: {e:?}");
                     }
                     None => {
                         tracing::debug!("tcp stream closed by server");