T-8292 [Fable] Atomic unix socket takeover + inode-guarded cleanup

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<Option<Arc<tokio::runtime::Runtime>>>,
     shutdown: RefCell<Option<broadcast::Sender<()>>>,
     total_connections: Arc<AtomicU64>,
+    // (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<Option<(u64, u64)>>,
 }
 
 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::<SocketAddr>() {
                     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(())

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)