diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/session.rs | 187 |
1 files changed, 168 insertions, 19 deletions
diff --git a/src/session.rs b/src/session.rs index d537a0f9..56330aa8 100644 --- a/src/session.rs +++ b/src/session.rs @@ -199,34 +199,27 @@ pub fn stop_session(name: Option<&str>) -> Result<SessionInfo, String> { } fn stop_session_with_timeout(name: Option<&str>, timeout: Duration) -> Result<SessionInfo, String> { + let deadline = Instant::now() + timeout; let socket_path = api_socket_path_for(name); let request = serde_json::json!({ "id": "cli:session:stop", "method": "server.stop", "params": {} }); - let mut stream = UnixStream::connect(&socket_path).map_err(|err| { + let stream = UnixStream::connect(&socket_path).map_err(|err| { format!( "session {} is not running or cannot be reached at {}: {err}", name.unwrap_or(DEFAULT_SESSION_NAME), socket_path.display() ) })?; - stream - .write_all(request.to_string().as_bytes()) - .map_err(|err| err.to_string())?; - stream.write_all(b"\n").map_err(|err| err.to_string())?; - stream.flush().map_err(|err| err.to_string())?; - - let mut line = String::new(); - BufReader::new(stream) - .read_line(&mut line) - .map_err(|err| err.to_string())?; - let response: serde_json::Value = serde_json::from_str(&line).map_err(|err| err.to_string())?; - if let Some(error) = response.get("error") { - return Err(error.to_string()); + let stop_response = send_stop_request(stream, &request, deadline)?; + if let Some(response) = stop_response { + if let Some(error) = response.get("error") { + return Err(error.to_string()); + } } - if !wait_until_stopped(&socket_path, timeout) { + if !wait_until_stopped_until(&socket_path, deadline) { return Err(format!( "session {} did not stop within {}ms; socket is still reachable at {}", name.unwrap_or(DEFAULT_SESSION_NAME), @@ -257,21 +250,83 @@ pub fn delete_session(name: &str) -> Result<SessionInfo, String> { } } +fn send_stop_request( + mut stream: UnixStream, + request: &serde_json::Value, + deadline: Instant, +) -> Result<Option<serde_json::Value>, String> { + let write_timeout = time_until(deadline); + if write_timeout.is_zero() { + return Ok(None); + } + stream + .set_write_timeout(Some(write_timeout)) + .map_err(|err| err.to_string())?; + + let response = send_stop_request_inner(&mut stream, request, deadline); + match response { + Ok(Some(line)) => serde_json::from_str(&line) + .map(Some) + .map_err(|err| err.to_string()), + Ok(None) => Ok(None), + Err(err) if stop_request_error_allows_wait(&err) => Ok(None), + Err(err) => Err(err.to_string()), + } +} + +fn send_stop_request_inner( + stream: &mut UnixStream, + request: &serde_json::Value, + deadline: Instant, +) -> std::io::Result<Option<String>> { + stream.write_all(request.to_string().as_bytes())?; + stream.write_all(b"\n")?; + stream.flush()?; + + let read_timeout = time_until(deadline); + if read_timeout.is_zero() { + return Ok(None); + } + stream.set_read_timeout(Some(read_timeout))?; + + let mut line = String::new(); + let bytes_read = BufReader::new(stream).read_line(&mut line)?; + if bytes_read == 0 { + return Ok(None); + } + Ok(Some(line)) +} + +fn stop_request_error_allows_wait(err: &std::io::Error) -> bool { + matches!( + err.kind(), + std::io::ErrorKind::BrokenPipe + | std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::UnexpectedEof + | std::io::ErrorKind::NotConnected + | std::io::ErrorKind::TimedOut + | std::io::ErrorKind::WouldBlock + ) +} + fn is_running_at(socket_path: &Path) -> bool { socket_path.exists() && UnixStream::connect(socket_path).is_ok() } -fn wait_until_stopped(socket_path: &Path, timeout: Duration) -> bool { - let deadline = Instant::now() + timeout; +fn wait_until_stopped_until(socket_path: &Path, deadline: Instant) -> bool { while Instant::now() < deadline { if !is_running_at(socket_path) { return true; } - std::thread::sleep(STOP_WAIT_POLL); + std::thread::sleep(STOP_WAIT_POLL.min(time_until(deadline))); } !is_running_at(socket_path) } +fn time_until(deadline: Instant) -> Duration { + deadline.saturating_duration_since(Instant::now()) +} + pub fn validate_name(name: &str) -> Result<(), String> { if name.is_empty() { return Err("session name cannot be empty".to_string()); @@ -325,6 +380,95 @@ mod tests { } #[test] + fn stop_request_errors_wait_for_socket_state() { + for kind in [ + std::io::ErrorKind::BrokenPipe, + std::io::ErrorKind::ConnectionReset, + std::io::ErrorKind::UnexpectedEof, + std::io::ErrorKind::NotConnected, + std::io::ErrorKind::TimedOut, + std::io::ErrorKind::WouldBlock, + ] { + let err = std::io::Error::from(kind); + assert!(stop_request_error_allows_wait(&err), "{kind:?}"); + } + } + + #[test] + fn stop_request_empty_response_waits_for_socket_state() { + let (client, server) = UnixStream::pair().unwrap(); + let handle = std::thread::spawn(move || { + let mut request = String::new(); + let _ = BufReader::new(server).read_line(&mut request); + request + }); + let request = serde_json::json!({ + "id": "cli:session:stop", + "method": "server.stop", + "params": {} + }); + + assert_eq!( + send_stop_request( + client, + &request, + Instant::now() + Duration::from_millis(100) + ) + .unwrap(), + None + ); + assert!(handle.join().unwrap().contains("server.stop")); + } + + #[test] + fn stop_session_times_out_when_socket_stays_open_without_response() { + let _guard = env_lock().lock().unwrap(); + let config_home = PathBuf::from(format!("/tmp/hs-stop-open-{}", std::process::id())); + std::env::set_var("XDG_CONFIG_HOME", &config_home); + let session_name = "silent"; + let socket_path = api_socket_path_for(Some(session_name)); + std::fs::create_dir_all(socket_path.parent().unwrap()).unwrap(); + let _ = std::fs::remove_file(&socket_path); + let listener = std::os::unix::net::UnixListener::bind(&socket_path).unwrap(); + listener.set_nonblocking(true).unwrap(); + let keep_running = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true)); + let keep_running_for_thread = keep_running.clone(); + let handle = std::thread::spawn(move || { + let mut held_streams = Vec::new(); + while keep_running_for_thread.load(Ordering::Relaxed) { + match listener.accept() { + Ok((stream, _)) => { + if let Ok(reader_stream) = stream.try_clone() { + let mut request = String::new(); + match BufReader::new(reader_stream).read_line(&mut request) { + Ok(0) => continue, + Ok(_) if request.contains("server.stop") => { + held_streams.push(stream) + } + Ok(_) => {} + Err(_) => continue, + } + } + } + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => { + std::thread::sleep(Duration::from_millis(5)); + } + Err(_) => break, + } + } + }); + + let err = stop_session_with_timeout(Some(session_name), Duration::from_millis(75)) + .expect_err("silent session should fail after timeout"); + + assert!(err.contains("did not stop"), "{err}"); + keep_running.store(false, Ordering::Relaxed); + handle.join().unwrap(); + let _ = std::fs::remove_dir_all(&config_home); + std::env::remove_var("XDG_CONFIG_HOME"); + } + + #[test] fn configure_from_args_removes_global_session_option() { let _guard = env_lock().lock().unwrap(); std::env::remove_var(SESSION_ENV_VAR); @@ -617,7 +761,12 @@ mod tests { Ok((mut stream, _)) => { if let Ok(reader_stream) = stream.try_clone() { let mut request = String::new(); - let _ = BufReader::new(reader_stream).read_line(&mut request); + match BufReader::new(reader_stream).read_line(&mut request) { + Ok(0) => continue, + Ok(_) if request.trim().is_empty() => continue, + Ok(_) => {} + Err(_) => continue, + } } let _ = stream.write_all(b"{\"id\":\"cli:session:stop\",\"result\":{}}\n"); let _ = stream.flush(); |