fix: let repeated sysg logs follow sessions reconnect cleanly

ra0x3 · ra0x3 · commit 639ce0674fd5 · 2026-04-05T13:17:39.000-07:00
diff --git a/src/bin/main.rs b/src/bin/main.rs
@@ -16,7 +16,7 @@ use crossterm::{
     event::{self, Event, KeyCode, KeyEvent, KeyModifiers},
     terminal,
 };
-use libc::{SIGKILL, SIGTERM, getpgrp, getppid, killpg};
+use libc::{SIGKILL, SIGTERM, getppid};
 use nix::{
     sys::signal,
     unistd::{Pid, Uid},
@@ -1775,6 +1775,106 @@ fn force_kill(pid: libc::pid_t) {
     let _ = wait_for_supervisor_exit(pid, Duration::from_secs(2));
 }
 
+/// Returns whether the tracked root PID is still alive.
+fn tracked_service_alive(pid: libc::pid_t) -> bool {
+    let target = Pid::from_raw(pid);
+    match signal::kill(target, None) {
+        Ok(_) => true,
+        Err(err) => err != nix::Error::from(nix::errno::Errno::ESRCH),
+    }
+}
+
+/// Waits for a tracked service root PID to exit.
+fn wait_for_tracked_service_exit(pid: libc::pid_t, timeout: Duration) -> bool {
+    let deadline = Instant::now() + timeout;
+    while Instant::now() < deadline {
+        if !tracked_service_alive(pid) {
+            return true;
+        }
+        thread::sleep(Duration::from_millis(100));
+    }
+    !tracked_service_alive(pid)
+}
+
+/// Sends a signal to a tracked service using its process group when available and falling back
+/// to the root PID when group signaling is not possible.
+fn signal_tracked_service(
+    service: &str,
+    pid: libc::pid_t,
+    pgid: Option<libc::pid_t>,
+    signal: libc::c_int,
+) {
+    let mut delivered = false;
+
+    if let Some(group_id) = pgid {
+        unsafe {
+            if libc::killpg(group_id, signal) == 0 {
+                delivered = true;
+            } else {
+                let err = std::io::Error::last_os_error();
+                match err.raw_os_error() {
+                    Some(code) if code == libc::ESRCH => {
+                        delivered = true;
+                    }
+                    Some(code) if code == libc::EPERM => {}
+                    _ => eprintln!(
+                        "systemg: failed to send signal {signal} to '{service}' (pgid {group_id}): {err}"
+                    ),
+                }
+            }
+        }
+    }
+
+    if delivered {
+        return;
+    }
+
+    unsafe {
+        if libc::kill(pid, signal) == -1 {
+            let err = std::io::Error::last_os_error();
+            match err.raw_os_error() {
+                Some(code) if code == libc::ESRCH => {}
+                _ => eprintln!(
+                    "systemg: failed to send signal {signal} to '{service}' (pid {pid}): {err}"
+                ),
+            }
+        }
+    }
+}
+
+/// Terminates tracked services from the pid file during foreground Ctrl-C shutdown.
+fn terminate_tracked_services_on_shutdown() {
+    let mut tracked: Vec<(String, libc::pid_t, Option<libc::pid_t>)> = Vec::new();
+    if let Ok(pid_file) = PidFile::load() {
+        for (service, pid) in pid_file.services() {
+            tracked.push((
+                service.clone(),
+                *pid as libc::pid_t,
+                pid_file.pgid_for(service).map(|group| group as libc::pid_t),
+            ));
+        }
+    }
+
+    for (service, pid, pgid) in &tracked {
+        signal_tracked_service(service, *pid, *pgid, libc::SIGTERM);
+    }
+
+    for (service, pid, pgid) in &tracked {
+        if wait_for_tracked_service_exit(*pid, Duration::from_secs(2)) {
+            continue;
+        }
+        signal_tracked_service(service, *pid, *pgid, libc::SIGKILL);
+    }
+
+    for (service, pid, _) in &tracked {
+        if !wait_for_tracked_service_exit(*pid, Duration::from_secs(2)) {
+            eprintln!(
+                "systemg: service '{service}' (pid {pid}) survived foreground shutdown"
+            );
+        }
+    }
+}
+
 /// Handles process exited.
 fn process_exited(pid: libc::pid_t) -> bool {
     let proc_root = PathBuf::from(format!("/proc/{pid}"));
@@ -2280,55 +2380,8 @@ fn daemonize_systemg() -> std::io::Result<()> {
 fn register_signal_handler() -> Result<(), Box<dyn Error>> {
     ctrlc::set_handler(move || {
         println!("systemg is shutting down... terminating child services");
-
-        let mut service_pids: Vec<(String, libc::pid_t)> = Vec::new();
-        if let Ok(pid_file) = PidFile::load() {
-            for (service, pid) in pid_file.services() {
-                service_pids.push((service.clone(), *pid as libc::pid_t));
-            }
-        }
-
-        for (service, pgid) in &service_pids {
-            unsafe {
-                if libc::killpg(*pgid, libc::SIGTERM) == -1 {
-                    let err = std::io::Error::last_os_error();
-                    match err.raw_os_error() {
-                        Some(code) if code == libc::ESRCH => {}
-                        Some(code) if code == libc::EPERM => {
-                            let _ = libc::kill(*pgid, libc::SIGTERM);
-                        }
-                        _ => eprintln!(
-                            "systemg: failed to send SIGTERM to '{service}' (pgid {pgid}): {err}"
-                        ),
-                    }
-                }
-            }
-        }
-
-        std::thread::sleep(std::time::Duration::from_millis(150));
-
-        for (service, pgid) in &service_pids {
-            unsafe {
-                if libc::killpg(*pgid, libc::SIGKILL) == -1 {
-                    let err = std::io::Error::last_os_error();
-                    match err.raw_os_error() {
-                        Some(code) if code == libc::ESRCH => {}
-                        Some(code) if code == libc::EPERM => {
-                            let _ = libc::kill(*pgid, libc::SIGKILL);
-                        }
-                        _ => eprintln!(
-                            "systemg: failed to send SIGKILL to '{service}' (pgid {pgid}): {err}"
-                        ),
-                    }
-                }
-            }
-        }
-
-        unsafe {
-            let pgid = getpgrp();
-            killpg(pgid, SIGKILL);
-        }
-
+        terminate_tracked_services_on_shutdown();
+        let _ = ipc::cleanup_runtime();
         std::process::exit(0);
     })?;
 
diff --git a/src/logs.rs b/src/logs.rs
@@ -9,13 +9,14 @@ use std::{
     fs::{self, OpenOptions},
     io::{IsTerminal, Read, Write},
     path::{Path, PathBuf},
-    sync::{Arc, Mutex, OnceLock, mpsc},
+    sync::{Arc, Mutex, OnceLock, mpsc, mpsc::RecvTimeoutError},
     thread,
+    time::Duration,
 };
 #[cfg(any(target_os = "linux", target_os = "macos"))]
 use std::{
     os::unix::{
-        io::{FromRawFd, IntoRawFd},
+        io::{AsRawFd, FromRawFd, IntoRawFd},
         net::UnixStream,
     },
     process::{Command, Stdio},
@@ -106,6 +107,35 @@ fn subscribe_live_log(service: &str, kind: &str) -> mpsc::Receiver<Vec<u8>> {
     rx
 }
 
+/// Returns whether the client side of a Unix socket has disconnected.
+#[cfg(any(target_os = "linux", target_os = "macos"))]
+fn socket_peer_disconnected(stream: &UnixStream) -> bool {
+    let fd = stream.as_raw_fd();
+    let mut byte = 0_u8;
+    let result = unsafe {
+        libc::recv(
+            fd,
+            &mut byte as *mut u8 as *mut libc::c_void,
+            1,
+            libc::MSG_PEEK | libc::MSG_DONTWAIT,
+        )
+    };
+
+    if result == 0 {
+        return true;
+    }
+
+    if result < 0 {
+        let err = std::io::Error::last_os_error();
+        return !matches!(
+            err.raw_os_error(),
+            Some(code) if code == libc::EAGAIN || code == libc::EWOULDBLOCK
+        );
+    }
+
+    false
+}
+
 /// Normalizes this item.
 fn normalize(name: &str) -> String {
     name.chars()
@@ -630,9 +660,30 @@ impl LogManager {
                 }
                 if matches!(mode, TailMode::Follow) {
                     let receiver = subscribe_live_log(service_name, kind_name);
-                    while let Ok(chunk) = receiver.recv() {
-                        socket.write_all(&chunk)?;
-                        socket.flush()?;
+                    loop {
+                        match receiver.recv_timeout(Duration::from_millis(250)) {
+                            Ok(chunk) => match socket.write_all(&chunk) {
+                                Ok(()) => {
+                                    socket.flush()?;
+                                }
+                                Err(err)
+                                    if matches!(
+                                        err.kind(),
+                                        std::io::ErrorKind::BrokenPipe
+                                            | std::io::ErrorKind::ConnectionReset
+                                    ) =>
+                                {
+                                    break;
+                                }
+                                Err(err) => return Err(err.into()),
+                            },
+                            Err(RecvTimeoutError::Timeout) => {
+                                if socket_peer_disconnected(&socket) {
+                                    break;
+                                }
+                            }
+                            Err(RecvTimeoutError::Disconnected) => break,
+                        }
                     }
                 }
                 return Ok(());
diff --git a/src/supervisor.rs b/src/supervisor.rs
@@ -648,12 +648,41 @@ impl Supervisor {
                             follow,
                         } = command
                         {
-                            if let Err(err) = self.handle_logs_command(
-                                service, lines, &kind, follow, &stream,
-                            ) {
-                                error!("Supervisor logs command failed: {err}");
-                                let _ = writeln!(stream, "{err}");
-                            }
+                            let snapshot = match self.collect_live_snapshot() {
+                                Ok(snapshot) => {
+                                    self.status_cache.replace(snapshot.clone());
+                                    snapshot
+                                }
+                                Err(err) => {
+                                    error!("Supervisor logs command failed: {err}");
+                                    let _ = writeln!(stream, "{err}");
+                                    continue;
+                                }
+                            };
+                            let pid_file = self.daemon.pid_file_handle();
+                            let mut log_stream = match stream.try_clone() {
+                                Ok(stream) => stream,
+                                Err(err) => {
+                                    error!(
+                                        "Failed to clone supervisor log stream: {err}"
+                                    );
+                                    continue;
+                                }
+                            };
+                            thread::spawn(move || {
+                                if let Err(err) = Supervisor::handle_logs_command(
+                                    snapshot,
+                                    pid_file,
+                                    service,
+                                    lines,
+                                    &kind,
+                                    follow,
+                                    &log_stream,
+                                ) {
+                                    error!("Supervisor logs command failed: {err}");
+                                    let _ = writeln!(log_stream, "{err}");
+                                }
+                            });
                             continue;
                         }
                         match self.handle_command(command) {
@@ -698,16 +727,14 @@ impl Supervisor {
 
     /// Streams logs through the supervisor-owned control socket.
     fn handle_logs_command(
-        &mut self,
+        snapshot: crate::status::StatusSnapshot,
+        pid_file: std::sync::Arc<std::sync::Mutex<crate::daemon::PidFile>>,
         service: Option<String>,
         lines: usize,
         kind: &str,
         follow: bool,
         stream: &std::os::unix::net::UnixStream,
     ) -> Result<(), SupervisorError> {
-        let snapshot = self.collect_live_snapshot()?;
-        self.status_cache.replace(snapshot.clone());
-        let pid_file = self.daemon.pid_file_handle();
         let manager = LogManager::new(pid_file);
         let requested_kind = Some(kind);
 
