diff --git a/ext/hyper_ruby/src/lib.rs b/ext/hyper_ruby/src/lib.rs index dbd892d..7a89261 100644 --- a/ext/hyper_ruby/src/lib.rs +++ b/ext/hyper_ruby/src/lib.rs @@ -16,6 +16,7 @@ use tokio::io::{AsyncRead, AsyncWrite}; use std::cell::RefCell; use std::net::SocketAddr; +use std::os::unix::fs::MetadataExt; use std::sync::atomic::{AtomicU64, Ordering}; use tokio::net::{TcpListener, UnixListener}; @@ -113,6 +114,9 @@ struct Server { runtime: RefCell>>, shutdown: RefCell>>, total_connections: Arc, + // (dev, ino) of the Unix socket file this server bound, so stop() only removes + // the file if a replacement server hasn't taken over the path in the meantime. + socket_ident: RefCell>, } impl Server { @@ -126,6 +130,7 @@ impl Server { runtime: RefCell::new(None), shutdown: RefCell::new(None), total_connections: Arc::new(AtomicU64::new(0)), + socket_ident: RefCell::new(None), } } @@ -310,32 +315,40 @@ impl Server { // Create the listener with proper error handling let listener = if config.bind_address.starts_with("unix:") { let path = config.bind_address.trim_start_matches("unix:"); - - // Check if the socket file already exists and try to delete it - if std::path::Path::new(path).exists() { - debug!("Unix socket file {} already exists, attempting to remove it", path); - match std::fs::remove_file(path) { - Ok(_) => debug!("Successfully removed existing socket file"), - Err(e) => { - error!("Failed to remove existing Unix socket file {}: {}", path, e); - return Err(MagnusError::new( - magnus::exception::runtime_error(), - format!("Failed to remove existing Unix socket file {}: {}", path, e) - )); - } - } - } - - match UnixListener::bind(path) { - Ok(listener) => Listener::Unix(listener), + + // Bind to a unique temp path and atomically rename over the target, so the + // path always points at a live socket even when a replacement server takes + // over a path an older, still-draining server bound. + static SOCKET_TMP_SEQ: AtomicU64 = AtomicU64::new(0); + let tmp_path = format!("{}.{}.{}.tmp", path, std::process::id(), SOCKET_TMP_SEQ.fetch_add(1, Ordering::Relaxed)); + + let listener = match UnixListener::bind(&tmp_path) { + Ok(listener) => listener, Err(e) => { - error!("Failed to bind to Unix socket {}: {}", path, e); + error!("Failed to bind to Unix socket {}: {}", tmp_path, e); return Err(MagnusError::new( magnus::exception::runtime_error(), - format!("Failed to bind to Unix socket {}: {}", path, e) + format!("Failed to bind to Unix socket {}: {}", tmp_path, e) )); } + }; + + // The socket file's identity survives the rename; stop() compares against it + // so an older server generation never unlinks a newer generation's socket. + let ident = std::fs::symlink_metadata(&tmp_path).ok().map(|m| (m.dev(), m.ino())); + + if let Err(e) = std::fs::rename(&tmp_path, path) { + let _ = std::fs::remove_file(&tmp_path); + error!("Failed to install Unix socket file {}: {}", path, e); + return Err(MagnusError::new( + magnus::exception::runtime_error(), + format!("Failed to install Unix socket file {}: {}", path, e) + )); } + + *self.socket_ident.borrow_mut() = ident; + + Listener::Unix(listener) } else { match config.bind_address.parse::() { Ok(addr) => { @@ -474,9 +487,18 @@ impl Server { let bind_address = self.config.borrow().bind_address.clone(); if bind_address.starts_with("unix:") { let path = bind_address.trim_start_matches("unix:"); - std::fs::remove_file(path).unwrap_or_else(|e| { - warn!("Failed to remove socket file: {:?}", e); - }); + // Only remove the socket file if it's still the one this server bound; a + // replacement server may have taken over the path while we were draining. + match (self.socket_ident.borrow_mut().take(), std::fs::symlink_metadata(path)) { + (Some((dev, ino)), Ok(meta)) if (meta.dev(), meta.ino()) == (dev, ino) => { + std::fs::remove_file(path).unwrap_or_else(|e| { + warn!("Failed to remove socket file: {:?}", e); + }); + } + (Some(_), Ok(_)) => info!("Socket file {} was replaced by another server; leaving it in place", path), + (Some(_), Err(_)) => debug!("Socket file {} already removed", path), + (None, _) => {} + } } Ok(()) diff --git a/test/test_http.rb b/test/test_http.rb index 173a837..867d1d6 100644 --- a/test/test_http.rb +++ b/test/test_http.rb @@ -153,6 +153,65 @@ def test_unix_socket_cleanup end end + def test_unix_socket_takeover_preserves_new_generation_socket + # A replacement server (e.g. a redeployed container) can bind the same path while the + # old server is still draining; the old server's stop must not delete the new socket. + socket_path = "/tmp/hyper_ruby_test_takeover.sock" + File.unlink(socket_path) if File.exist?(socket_path) + + old_server = HyperRuby::Server.new + old_server.configure({ bind_address: "unix:#{socket_path}" }) + old_server.start + + new_server = HyperRuby::Server.new + new_server.configure({ bind_address: "unix:#{socket_path}" }) + new_server.start + + workers = 1.times.map do + Thread.new do + new_server.run_worker { |request| handler_simple(request) } + end + end + + old_server.stop + old_server = nil + assert File.exist?(socket_path), "old server's stop must not remove the new server's socket" + + client = HTTPX.with(transport: "unix", addresses: [socket_path], origin: "http://host") + response = client.get("/") + assert_equal 200, response.status + + new_server.stop + workers.each(&:join) + workers = nil + refute File.exist?(socket_path), "new server's stop should remove the socket it owns" + ensure + old_server.stop if old_server + new_server.stop if new_server && workers + workers&.each(&:join) + File.unlink(socket_path) if File.exist?(socket_path) + end + + def test_unix_socket_stop_leaves_foreign_file + # If something else has replaced our socket file, stop must leave it alone. + socket_path = "/tmp/hyper_ruby_test_foreign.sock" + File.unlink(socket_path) if File.exist?(socket_path) + + server = HyperRuby::Server.new + server.configure({ bind_address: "unix:#{socket_path}" }) + server.start + + File.unlink(socket_path) + FileUtils.touch(socket_path) + + server.stop + server = nil + assert File.exist?(socket_path), "stop must not remove a file it did not bind" + ensure + server.stop if server + File.unlink(socket_path) if File.exist?(socket_path) + end + # This test requires root permissions to create a file that can't be deleted. # Skip it unless we're running with proper permissions. def test_unix_socket_undeletable @@ -171,13 +230,13 @@ def test_unix_socket_undeletable server = HyperRuby::Server.new server.configure({ bind_address: "unix:#{socket_path}" }) - # This should raise an exception about not being able to remove the file + # This should raise an exception about not being able to install the socket file error = assert_raises(RuntimeError) do server.start end - + # Verify the error message - assert_match(/Failed to remove existing Unix socket file/, error.message) + assert_match(/Failed to install Unix socket file/, error.message) ensure # Clean up with sudo system("sudo rm -f #{socket_path}") if File.exist?(socket_path) @@ -201,14 +260,15 @@ def test_unix_socket_directory_error server.start end - # The error is from trying to remove the directory, not from binding - assert_match(/Failed to remove existing Unix socket file/, error.message) - - # It should include something about "Operation not permitted" or similar - assert(error.message.include?("Operation not permitted") || - error.message.include?("Permission denied") || - error.message.include?("not a socket"), - "Error should indicate issue with removing directory: #{error.message}") + # The error is from renaming the bound socket over the directory, not from binding + assert_match(/Failed to install Unix socket file/, error.message) + + # It should include something about the target being a directory or similar + assert(error.message.include?("Is a directory") || + error.message.include?("Operation not permitted") || + error.message.include?("Permission denied") || + error.message.include?("not a socket"), + "Error should indicate issue with replacing directory: #{error.message}") ensure # Clean up FileUtils.rm_rf(socket_dir) if Dir.exist?(socket_dir)