fix: recover from stale daemon/socket state (#1136)
When a daemon is killed or crashes without cleaning up, stale .sock/.pid files are left behind. Previously, `close --all` would fail to connect to these zombie daemons and simply report an error, leaving the stale files in place and poisoning all future sessions. Three fixes: 1. `close --all` now force-kills unreachable daemon processes and removes all stale files (pid, sock, stream) instead of reporting failure. It also cleans up dead-but-lingering PID files during enumeration and scans for orphaned .sock files without corresponding .pid files. 2. `ensure_daemon` handles concurrent startup races: when a spawned daemon exits with "Address already in use" (another instance won the bind race), it checks whether the winner is accepting connections and piggybacks on it instead of failing. 3. `cleanup_stale_files` is now public so `close --all` can reuse it. Fixes #1118 Co-authored-by: ctate <366502+ctate@users.noreply.github.com>
This commit is contained in:
+16
-1
@@ -119,7 +119,7 @@ fn get_pid_path(session: &str) -> PathBuf {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Clean up stale socket and PID files for a session
|
/// Clean up stale socket and PID files for a session
|
||||||
fn cleanup_stale_files(session: &str) {
|
pub fn cleanup_stale_files(session: &str) {
|
||||||
let pid_path = get_pid_path(session);
|
let pid_path = get_pid_path(session);
|
||||||
let _ = fs::remove_file(&pid_path);
|
let _ = fs::remove_file(&pid_path);
|
||||||
let stream_path = get_socket_dir().join(format!("{}.stream", session));
|
let stream_path = get_socket_dir().join(format!("{}.stream", session));
|
||||||
@@ -429,6 +429,21 @@ pub fn ensure_daemon(session: &str, opts: &DaemonOptions) -> Result<DaemonResult
|
|||||||
let _ = stderr.read_to_string(&mut stderr_output);
|
let _ = stderr.read_to_string(&mut stderr_output);
|
||||||
}
|
}
|
||||||
let stderr_trimmed = stderr_output.trim();
|
let stderr_trimmed = stderr_output.trim();
|
||||||
|
|
||||||
|
// If the daemon failed because another instance won the bind
|
||||||
|
// race ("Address already in use"), check whether that winner is
|
||||||
|
// now accepting connections and piggyback on it.
|
||||||
|
if stderr_trimmed.contains("Address already in use")
|
||||||
|
|| stderr_trimmed.contains("Failed to bind")
|
||||||
|
{
|
||||||
|
thread::sleep(Duration::from_millis(200));
|
||||||
|
if daemon_ready(session) {
|
||||||
|
return Ok(DaemonResult {
|
||||||
|
already_running: true,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if !stderr_trimmed.is_empty() {
|
if !stderr_trimmed.is_empty() {
|
||||||
let msg = if stderr_trimmed.len() > 500 {
|
let msg = if stderr_trimmed.len() > 500 {
|
||||||
let mut end = 500;
|
let mut end = 500;
|
||||||
|
|||||||
+47
-5
@@ -21,7 +21,7 @@ use windows_sys::Win32::Foundation::CloseHandle;
|
|||||||
use windows_sys::Win32::System::Threading::{OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION};
|
use windows_sys::Win32::System::Threading::{OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION};
|
||||||
|
|
||||||
use commands::{gen_id, parse_command, ParseError};
|
use commands::{gen_id, parse_command, ParseError};
|
||||||
use connection::{ensure_daemon, get_socket_dir, send_command, DaemonOptions};
|
use connection::{cleanup_stale_files, ensure_daemon, get_socket_dir, send_command, DaemonOptions};
|
||||||
use flags::{clean_args, parse_flags, Flags};
|
use flags::{clean_args, parse_flags, Flags};
|
||||||
use install::run_install;
|
use install::run_install;
|
||||||
use output::{
|
use output::{
|
||||||
@@ -380,7 +380,7 @@ fn run_dashboard_stop(json_mode: bool) {
|
|||||||
|
|
||||||
fn run_close_all(flags: &Flags) {
|
fn run_close_all(flags: &Flags) {
|
||||||
let socket_dir = get_socket_dir();
|
let socket_dir = get_socket_dir();
|
||||||
let mut sessions: Vec<String> = Vec::new();
|
let mut sessions: Vec<(String, u32)> = Vec::new();
|
||||||
|
|
||||||
if let Ok(entries) = fs::read_dir(&socket_dir) {
|
if let Ok(entries) = fs::read_dir(&socket_dir) {
|
||||||
for entry in entries.flatten() {
|
for entry in entries.flatten() {
|
||||||
@@ -409,9 +409,33 @@ fn run_close_all(flags: &Flags) {
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
if running {
|
if running {
|
||||||
sessions.push(session_name.to_string());
|
sessions.push((session_name.to_string(), pid));
|
||||||
|
} else {
|
||||||
|
// Process is gone but stale files remain; clean them up
|
||||||
|
cleanup_stale_files(session_name);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
} else {
|
||||||
|
// PID file exists but is unreadable; clean up stale files
|
||||||
|
cleanup_stale_files(session_name);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Also scan for orphaned .sock files without corresponding .pid files
|
||||||
|
#[cfg(unix)]
|
||||||
|
if let Ok(entries) = fs::read_dir(&socket_dir) {
|
||||||
|
for entry in entries.flatten() {
|
||||||
|
let name = entry.file_name().to_string_lossy().to_string();
|
||||||
|
if let Some(session_name) = name.strip_suffix(".sock") {
|
||||||
|
if session_name.is_empty() {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
let pid_path = socket_dir.join(format!("{}.pid", session_name));
|
||||||
|
if !pid_path.exists() {
|
||||||
|
// Orphaned socket file with no PID file; remove it
|
||||||
|
cleanup_stale_files(session_name);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -432,7 +456,7 @@ fn run_close_all(flags: &Flags) {
|
|||||||
let mut closed: Vec<String> = Vec::new();
|
let mut closed: Vec<String> = Vec::new();
|
||||||
let mut failed: Vec<(String, String)> = Vec::new();
|
let mut failed: Vec<(String, String)> = Vec::new();
|
||||||
|
|
||||||
for session in &sessions {
|
for (session, pid) in &sessions {
|
||||||
let cmd = json!({ "id": gen_id(), "action": "close" });
|
let cmd = json!({ "id": gen_id(), "action": "close" });
|
||||||
match send_command(cmd, session) {
|
match send_command(cmd, session) {
|
||||||
Ok(resp) if resp.success => closed.push(session.clone()),
|
Ok(resp) if resp.success => closed.push(session.clone()),
|
||||||
@@ -440,7 +464,25 @@ fn run_close_all(flags: &Flags) {
|
|||||||
let err = resp.error.unwrap_or_else(|| "Unknown error".to_string());
|
let err = resp.error.unwrap_or_else(|| "Unknown error".to_string());
|
||||||
failed.push((session.clone(), err));
|
failed.push((session.clone(), err));
|
||||||
}
|
}
|
||||||
Err(e) => failed.push((session.clone(), e.to_string())),
|
Err(_) => {
|
||||||
|
// Daemon is unreachable despite its process existing.
|
||||||
|
// Force-kill the process and clean up stale files so future
|
||||||
|
// sessions are not poisoned.
|
||||||
|
#[cfg(unix)]
|
||||||
|
unsafe {
|
||||||
|
libc::kill(*pid as i32, libc::SIGKILL);
|
||||||
|
}
|
||||||
|
#[cfg(windows)]
|
||||||
|
unsafe {
|
||||||
|
let handle = OpenProcess(1, 0, *pid); // PROCESS_TERMINATE = 1
|
||||||
|
if handle != 0 {
|
||||||
|
windows_sys::Win32::System::Threading::TerminateProcess(handle, 1);
|
||||||
|
CloseHandle(handle);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
cleanup_stale_files(session);
|
||||||
|
closed.push(session.clone());
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user