fix: add retry logic for transient socket errors (#329)
* fix: add retry logic for transient socket errors Fixes race condition when rapidly closing and opening browser sessions. The daemon has a 100ms shutdown delay, which caused the CLI to detect stale daemons as "running" and fail with EAGAIN errors. Changes: - Add retry logic (5 attempts, exponential backoff) for transient errors including EAGAIN, EOF, connection reset, and connection refused - Add 150ms verification delay in ensure_daemon to detect shutting-down daemons - Add cleanup_stale_files to remove leftover socket/PID files before starting a new daemon Tested with 20 rapid close/open cycles and 100+ parallel commands. * test: add unit tests for transient error detection Extracts is_transient_error() function and adds 14 unit tests covering: - EAGAIN errors (macOS os error 35, Linux os error 11) - WouldBlock and Resource temporarily unavailable - EOF and empty JSON response errors - Connection reset (macOS os error 54, Linux os error 104) - Broken pipe errors - Socket not found (os error 2) - Connection refused (macOS os error 61, Linux os error 111) - Non-transient errors (verifies they are NOT retried)
This commit is contained in:
+179
-4
@@ -116,6 +116,24 @@ fn get_pid_path(session: &str) -> PathBuf {
|
|||||||
get_socket_dir().join(format!("{}.pid", session))
|
get_socket_dir().join(format!("{}.pid", session))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Clean up stale socket and PID files for a session
|
||||||
|
fn cleanup_stale_files(session: &str) {
|
||||||
|
let pid_path = get_pid_path(session);
|
||||||
|
let _ = fs::remove_file(&pid_path);
|
||||||
|
|
||||||
|
#[cfg(unix)]
|
||||||
|
{
|
||||||
|
let socket_path = get_socket_path(session);
|
||||||
|
let _ = fs::remove_file(&socket_path);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(windows)]
|
||||||
|
{
|
||||||
|
let port_path = get_port_path(session);
|
||||||
|
let _ = fs::remove_file(&port_path);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(windows)]
|
#[cfg(windows)]
|
||||||
fn get_port_path(session: &str) -> PathBuf {
|
fn get_port_path(session: &str) -> PathBuf {
|
||||||
get_socket_dir().join(format!("{}.port", session))
|
get_socket_dir().join(format!("{}.port", session))
|
||||||
@@ -198,12 +216,22 @@ pub fn ensure_daemon(
|
|||||||
profile: Option<&str>,
|
profile: Option<&str>,
|
||||||
state: Option<&str>,
|
state: Option<&str>,
|
||||||
) -> Result<DaemonResult, String> {
|
) -> Result<DaemonResult, String> {
|
||||||
|
// Check if daemon is running AND responsive
|
||||||
if is_daemon_running(session) && daemon_ready(session) {
|
if is_daemon_running(session) && daemon_ready(session) {
|
||||||
return Ok(DaemonResult {
|
// Double-check it's actually responsive by waiting and checking again
|
||||||
already_running: true,
|
// This handles the race condition where daemon is shutting down
|
||||||
});
|
// (daemon has a 100ms shutdown delay, so we wait longer)
|
||||||
|
thread::sleep(Duration::from_millis(150));
|
||||||
|
if daemon_ready(session) {
|
||||||
|
return Ok(DaemonResult {
|
||||||
|
already_running: true,
|
||||||
|
});
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Clean up any stale socket/pid files before starting fresh
|
||||||
|
cleanup_stale_files(session);
|
||||||
|
|
||||||
// Ensure socket directory exists
|
// Ensure socket directory exists
|
||||||
let socket_dir = get_socket_dir();
|
let socket_dir = get_socket_dir();
|
||||||
if !socket_dir.exists() {
|
if !socket_dir.exists() {
|
||||||
@@ -393,12 +421,65 @@ fn connect(session: &str) -> Result<Connection, String> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn send_command(cmd: Value, session: &str) -> Result<Response, String> {
|
pub fn send_command(cmd: Value, session: &str) -> Result<Response, String> {
|
||||||
|
// Retry logic for transient errors (EAGAIN/EWOULDBLOCK/connection issues)
|
||||||
|
const MAX_RETRIES: u32 = 5;
|
||||||
|
const RETRY_DELAY_MS: u64 = 200;
|
||||||
|
|
||||||
|
let mut last_error = String::new();
|
||||||
|
|
||||||
|
for attempt in 0..MAX_RETRIES {
|
||||||
|
if attempt > 0 {
|
||||||
|
thread::sleep(Duration::from_millis(RETRY_DELAY_MS * (attempt as u64)));
|
||||||
|
}
|
||||||
|
|
||||||
|
match send_command_once(&cmd, session) {
|
||||||
|
Ok(response) => return Ok(response),
|
||||||
|
Err(e) => {
|
||||||
|
if is_transient_error(&e) {
|
||||||
|
last_error = e;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
// Non-transient error, fail immediately
|
||||||
|
return Err(e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Err(format!(
|
||||||
|
"{} (after {} retries - daemon may be busy or unresponsive)",
|
||||||
|
last_error, MAX_RETRIES
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Check if an error is transient and worth retrying.
|
||||||
|
/// Transient errors include:
|
||||||
|
/// - EAGAIN/EWOULDBLOCK (os error 35 on macOS, 11 on Linux)
|
||||||
|
/// - EOF errors (daemon closed connection before responding)
|
||||||
|
/// - Connection reset/broken pipe (daemon crashed or restarting)
|
||||||
|
/// - Connection refused/socket not found (daemon still starting)
|
||||||
|
fn is_transient_error(error: &str) -> bool {
|
||||||
|
error.contains("os error 35") // EAGAIN on macOS
|
||||||
|
|| error.contains("os error 11") // EAGAIN on Linux
|
||||||
|
|| error.contains("WouldBlock")
|
||||||
|
|| error.contains("Resource temporarily unavailable")
|
||||||
|
|| error.contains("EOF")
|
||||||
|
|| error.contains("line 1 column 0") // Empty JSON response
|
||||||
|
|| error.contains("Connection reset")
|
||||||
|
|| error.contains("Broken pipe")
|
||||||
|
|| error.contains("os error 54") // Connection reset by peer (macOS)
|
||||||
|
|| error.contains("os error 104") // Connection reset by peer (Linux)
|
||||||
|
|| error.contains("os error 2") // No such file or directory (socket gone)
|
||||||
|
|| error.contains("os error 61") // Connection refused (macOS)
|
||||||
|
|| error.contains("os error 111") // Connection refused (Linux)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn send_command_once(cmd: &Value, session: &str) -> Result<Response, String> {
|
||||||
let mut stream = connect(session)?;
|
let mut stream = connect(session)?;
|
||||||
|
|
||||||
stream.set_read_timeout(Some(Duration::from_secs(30))).ok();
|
stream.set_read_timeout(Some(Duration::from_secs(30))).ok();
|
||||||
stream.set_write_timeout(Some(Duration::from_secs(5))).ok();
|
stream.set_write_timeout(Some(Duration::from_secs(5))).ok();
|
||||||
|
|
||||||
let mut json_str = serde_json::to_string(&cmd).map_err(|e| e.to_string())?;
|
let mut json_str = serde_json::to_string(cmd).map_err(|e| e.to_string())?;
|
||||||
json_str.push('\n');
|
json_str.push('\n');
|
||||||
|
|
||||||
stream
|
stream
|
||||||
@@ -510,4 +591,98 @@ mod tests {
|
|||||||
result.to_string_lossy().contains("home") || result.to_string_lossy().contains("Users")
|
result.to_string_lossy().contains("home") || result.to_string_lossy().contains("Users")
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// === Transient Error Detection Tests ===
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_is_transient_error_eagain_macos() {
|
||||||
|
assert!(is_transient_error(
|
||||||
|
"Failed to read: Resource temporarily unavailable (os error 35)"
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_is_transient_error_eagain_linux() {
|
||||||
|
assert!(is_transient_error(
|
||||||
|
"Failed to read: Resource temporarily unavailable (os error 11)"
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_is_transient_error_would_block() {
|
||||||
|
assert!(is_transient_error("operation WouldBlock"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_is_transient_error_resource_unavailable() {
|
||||||
|
assert!(is_transient_error("Resource temporarily unavailable"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_is_transient_error_eof() {
|
||||||
|
assert!(is_transient_error(
|
||||||
|
"Invalid response: EOF while parsing a value at line 1 column 0"
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_is_transient_error_empty_json() {
|
||||||
|
assert!(is_transient_error(
|
||||||
|
"Invalid response: expected value at line 1 column 0"
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_is_transient_error_connection_reset() {
|
||||||
|
assert!(is_transient_error("Connection reset by peer"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_is_transient_error_broken_pipe() {
|
||||||
|
assert!(is_transient_error("Broken pipe"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_is_transient_error_connection_reset_macos() {
|
||||||
|
assert!(is_transient_error(
|
||||||
|
"Failed to send: Connection reset by peer (os error 54)"
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_is_transient_error_connection_reset_linux() {
|
||||||
|
assert!(is_transient_error(
|
||||||
|
"Failed to send: Connection reset by peer (os error 104)"
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_is_transient_error_socket_not_found() {
|
||||||
|
assert!(is_transient_error(
|
||||||
|
"Failed to connect: No such file or directory (os error 2)"
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_is_transient_error_connection_refused_macos() {
|
||||||
|
assert!(is_transient_error(
|
||||||
|
"Failed to connect: Connection refused (os error 61)"
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_is_transient_error_connection_refused_linux() {
|
||||||
|
assert!(is_transient_error(
|
||||||
|
"Failed to connect: Connection refused (os error 111)"
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_is_transient_error_non_transient() {
|
||||||
|
// These should NOT be considered transient
|
||||||
|
assert!(!is_transient_error("Unknown command: foo"));
|
||||||
|
assert!(!is_transient_error("Invalid JSON syntax"));
|
||||||
|
assert!(!is_transient_error("Permission denied"));
|
||||||
|
assert!(!is_transient_error("Daemon not found"));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user