Close log subscriber connection if send fails (instead of erroring)

binaries/coordinator/src/lib.rs

@@ -581,8 +581,11 @@ async fn start_inner(
             Event::Log(message) => {
                 if let Some(dataflow) = running_dataflows.get_mut(&message.dataflow_id) {
                     for subscriber in &mut dataflow.log_subscribers {
-                        subscriber.send_message(&message).await?;
+                        if subscriber.send_message(&message).await.is_err() {
+                            subscriber.close();
+                        }
                     }
+                    dataflow.log_subscribers.retain(|s| !s.is_closed());
                 }
             }
         }

binaries/coordinator/src/log_subscriber.rs

@@ -1,15 +1,19 @@
 use dora_core::coordinator_messages::LogMessage;
+use eyre::{Context, ContextCompat};
 
 use crate::tcp_utils::tcp_send;
 
 pub struct LogSubscriber {
     pub level: log::LevelFilter,
-    connection: tokio::net::TcpStream,
+    connection: Option<tokio::net::TcpStream>,
 }
 
 impl LogSubscriber {
     pub fn new(level: log::LevelFilter, connection: tokio::net::TcpStream) -> Self {
-        Self { level, connection }
+        Self {
+            level,
+            connection: Some(connection),
+        }
     }
 
     pub async fn send_message(&mut self, message: &LogMessage) -> eyre::Result<()> {
@@ -17,7 +21,18 @@ impl LogSubscriber {
             return Ok(());
         }
         let message = serde_json::to_vec(&message)?;
-        tcp_send(&mut self.connection, &message).await?;
+        let connection = self.connection.as_mut().context("connection is closed")?;
+        tcp_send(connection, &message)
+            .await
+            .context("failed to send message")?;
         Ok(())
     }
+
+    pub fn is_closed(&self) -> bool {
+        self.connection.is_none()
+    }
+
+    pub fn close(&mut self) {
+        self.connection = None;
+    }
 }