fix(supervisor): prevent file descriptor leaks in SSE streaming and IPC

src/ipc/server.rs

@@ -112,8 +112,10 @@ impl IpcServer {
                                     while let Some(req) = incoming_chan.recv().await {
                                         if let Err(err) = tx.send((req, outgoing_chan.clone())).await {
                                             debug!("Failed to send message: {err:?}");
+                                            break;
                                         }
                                     }
+                                    trace!("IPC connection handler task terminated cleanly");
                                 });
                             }
                             Err(err) => {
@@ -140,9 +142,7 @@ impl IpcServer {
             bail!("IPC message contains null byte");
         }
         msg.push(0);
-        if let Err(err) = send.write_all(&msg).await {
-            trace!("Failed to send message: {err:?}");
-        }
+        send.write_all(&msg).await.into_diagnostic()?;
         Ok(())
     }
 
@@ -207,8 +207,10 @@ impl IpcServer {
 
                 if let Err(err) = tx.send(msg).await {
                     warn!("Failed to emit message: {err:?}");
+                    break;
                 }
             }
+            trace!("IPC read task terminated cleanly");
         });
         rx
     }
@@ -228,9 +230,35 @@ impl IpcServer {
                     }
                 };
                 if let Err(err) = Self::send(&mut send, msg).await {
-                    warn!("Failed to send message: {err:?}");
+                    // Broken-pipe / reset is expected when a client disconnects normally
+                    // Traverse the error source chain to find the original io::Error
+                    // since miette wraps it in a DiagnosticError
+                    use std::error::Error as StdError;
+                    let is_disconnect = {
+                        let mut cur: Option<&dyn StdError> = Some(err.as_ref() as &dyn StdError);
+                        let mut found = false;
+                        while let Some(e) = cur {
+                            if let Some(io) = e.downcast_ref::<std::io::Error>() {
+                                found = matches!(
+                                    io.kind(),
+                                    std::io::ErrorKind::BrokenPipe
+                                        | std::io::ErrorKind::ConnectionReset
+                                );
+                                break;
+                            }
+                            cur = e.source();
+                        }
+                        found
+                    };
+                    if is_disconnect {
+                        debug!("IPC client disconnected: {err:?}");
+                    } else {
+                        warn!("Failed to send message: {err:?}");
+                    }
+                    break;
                 }
             }
+            trace!("IPC send task terminated cleanly");
         });
         tx
     }

src/web/routes/logs.rs

@@ -15,6 +15,7 @@ use crate::pitchfork_toml::PitchforkToml;
 use crate::state_file::StateFile;
 use crate::web::bp;
 use crate::web::helpers::{html_escape, url_encode};
+use std::io::{Read, Seek, SeekFrom};
 
 fn base_html(title: &str, content: &str) -> String {
     let bp = bp();
@@ -341,35 +342,165 @@ pub async fn stream_sse(
             }
         };
         let log_path = daemon_id.log_path();
-        let mut last_size = std::fs::metadata(&log_path).map(|m| m.len()).unwrap_or(0);
+        // Initialize last_size to 0 to avoid blocking metadata call at connection setup.
+        // The next loop iteration will detect and stream any existing file content.
+        let mut last_size = 0u64;
+        let mut file_handle: Option<std::fs::File> = None;
+
+        // Internal result type for file operations within spawn_blocking
+        enum FileOpResult {
+            Data(Vec<u8>),
+            Truncated,
+            FileRotated,
+            SeekFailed,
+            ReadFailed,
+            Eof,
+        }
+
+        // Track last seen inode to detect rotation when opening fresh handles
+        let mut last_path_ino: Option<u64> = None;
 
         loop {
             tokio::time::sleep(Duration::from_millis(500)).await;
 
-            if let Ok(metadata) = std::fs::metadata(&log_path) {
-                let current_size = metadata.len();
-
-                if current_size > last_size {
-                    // Read new content as bytes to handle invalid UTF-8
-                    if let Ok(file) = std::fs::File::open(&log_path) {
-                        use std::io::{Read, Seek, SeekFrom};
-                        let mut file = file;
-                        if file.seek(SeekFrom::Start(last_size)).is_ok() {
-                            let mut buffer = Vec::new();
-                            if file.read_to_end(&mut buffer).is_ok() && !buffer.is_empty() {
-                                // Use lossy conversion to handle invalid UTF-8 gracefully
-                                let new_content = String::from_utf8_lossy(&buffer);
-                                let escaped = html_escape(&new_content);
-                                yield Ok(Event::default().event("message").data(escaped));
+            // Use spawn_blocking for all blocking file operations
+            let file_op_result = {
+                let path = log_path.clone();
+                let fh = file_handle.take();
+                let mut ls = last_size;
+                let prev_ino = last_path_ino;
+                tokio::task::spawn_blocking(move || {
+                    let mut file = match fh {
+                        Some(f) => f,
+                        None => match std::fs::File::open(&path) {
+                            Ok(f) => f,
+                            Err(_) => return (None, ls, None, prev_ino),
+                        },
+                    };
+
+                    // Check if file was rotated while we had no handle (fresh open case)
+                    // and cache metadata to avoid redundant fstat calls
+                    #[cfg(unix)]
+                    let (fresh_open_rotated, cached_metadata, fresh_ino) = {
+                        use std::os::unix::fs::MetadataExt;
+                        if let Ok(meta) = file.metadata() {
+                            let ino = meta.ino();
+                            if let Some(prev_ino_val) = prev_ino {
+                                if ino != prev_ino_val {
+                                    // File was rotated since we last had a handle
+                                    (true, Some(meta), Some(ino))
+                                } else {
+                                    (false, Some(meta), Some(ino))
+                                }
+                            } else {
+                                // No previous inode, capture current one for future checks
+                                (false, Some(meta), Some(ino))
+                            }
+                        } else {
+                            (false, None, None)
+                        }
+                    };
+                    #[cfg(unix)]
+                    if fresh_open_rotated {
+                        return (None, 0, Some(FileOpResult::FileRotated), fresh_ino);
+                    }
+
+                    // Fallback for non-unix platforms
+                    #[cfg(not(unix))]
+                    let (cached_metadata, fresh_ino): (Option<std::fs::Metadata>, Option<u64>) = (None, None);
+
+                    let metadata = match cached_metadata {
+                        Some(m) => m,
+                        None => match file.metadata() {
+                            Ok(m) => m,
+                            Err(_) => return (Some(file), ls, None, fresh_ino),
+                        },
+                    };
+                    let current_size = metadata.len();
+
+                    // Check if file was recreated (inode changed) on Unix systems
+                    #[cfg(unix)]
+                    {
+                        use std::os::unix::fs::MetadataExt;
+                        let path_ino = std::fs::metadata(&path).map(|m| m.ino()).ok();
+                        let file_ino = metadata.ino();
+                        if let Some(path_ino) = path_ino && path_ino != file_ino {
+                            // File was recreated; drop handle and reset
+                            return (None, 0, Some(FileOpResult::FileRotated), Some(path_ino));
+                        }
+                    }
+
+                    // Use the current file's inode for future rotation checks
+                    #[cfg(unix)]
+                    let current_ino = fresh_ino.or_else(|| {
+                        use std::os::unix::fs::MetadataExt;
+                        file.metadata().ok().map(|m| m.ino())
+                    });
+                    #[cfg(not(unix))]
+                    let current_ino: Option<u64> = None;
+
+                    if current_size > ls {
+                        // Read new content as bytes to handle invalid UTF-8
+                        if file.seek(SeekFrom::Start(ls)).is_err() {
+                            return (None, ls, Some(FileOpResult::SeekFailed), current_ino);
+                        }
+
+                        const MAX_READ_SIZE: u64 = 1024 * 1024;
+                        let mut total_read = 0u64;
+                        let bytes_to_read = (current_size - ls).min(MAX_READ_SIZE) as usize;
+                        let mut buffer = vec![0u8; bytes_to_read];
+                        while total_read < bytes_to_read as u64 {
+                            match file.read(&mut buffer[total_read as usize..]) {
+                                Ok(0) => break,
+                                Ok(n) => total_read += n as u64,
+                                Err(_) => return (None, ls, Some(FileOpResult::ReadFailed), current_ino),
                             }
-                            // Always update last_size to avoid stalling on invalid content
-                            last_size = current_size;
                         }
+
+                        if total_read > 0 {
+                            buffer.truncate(total_read as usize);
+                            ls += total_read;
+                            return (Some(file), ls, Some(FileOpResult::Data(buffer)), current_ino);
+                        }
+                        // If total_read == 0, we hit EOF (only way to exit loop with 0 bytes)
+                        // Advance past bytes that are present in the file but unreadable
+                        ls = current_size;
+                        return (Some(file), ls, Some(FileOpResult::Eof), current_ino);
+                    } else if current_size < ls {
+                        // File was truncated (cleared)
+                        return (Some(file), 0, Some(FileOpResult::Truncated), current_ino);
                     }
-                } else if current_size < last_size {
-                    // File was truncated (cleared), send clear event and reset
-                    yield Ok(Event::default().event("clear").data(""));
-                    last_size = current_size;
+
+                    (Some(file), ls, None, current_ino)
+                }).await
+            };
+
+            match file_op_result {
+                Ok((fh, ls, result, ino)) => {
+                    file_handle = fh;
+                    last_size = ls;
+                    last_path_ino = ino;
+
+                    match result {
+                        Some(FileOpResult::Data(buffer)) => {
+                            let new_content = String::from_utf8_lossy(&buffer);
+                            let escaped = html_escape(&new_content);
+                            yield Ok(Event::default().event("message").data(escaped));
+                        }
+                        Some(FileOpResult::Truncated) => {
+                            yield Ok(Event::default().event("clear").data(""));
+                        }
+                        Some(FileOpResult::FileRotated) => {
+                            // Signal the client to clear stale content from the previous file
+                            yield Ok(Event::default().event("clear").data(""));
+                        }
+                        _ => {}
+                    }
+                }
+                Err(_) => {
+                    file_handle = None;
+                    // last_size remains valid from before this iteration;
+                    // do not reset it or we will re-stream the entire file.
                 }
             }
         }