aggressive unix socket cleanup handling

alistairjevans · alistairjevans · commit a771bd837ca4 · 2025-03-18T21:34:45.000Z
diff --git a/bin/bench-server.rb b/bin/bench-server.rb
@@ -0,0 +1,55 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+$LOAD_PATH.unshift File.expand_path("../lib", __dir__)
+
+puts "Loading hyper_ruby"
+
+require "hyper_ruby"
+require "json"
+
+# Create and configure the server
+server = HyperRuby::Server.new
+config = {
+  bind_address: ENV.fetch("BIND_ADDRESS", "127.0.0.1:3000"),
+  tokio_threads: ENV.fetch("TOKIO_THREADS", "1").to_i,
+  debug: ENV.fetch("DEBUG", "false") == "true",
+  recv_timeout: ENV.fetch("RECV_TIMEOUT", "30000").to_i
+}
+server.configure(config)
+
+puts "Starting server with config: #{config}"
+
+accept_response = HyperRuby::Response.new(
+  200,
+  { "Content-Type" => "application/json" },
+  { "message" => "Accepted" }.to_json
+)
+
+# Start the server
+server.start
+
+puts "Server started"
+
+# Create a worker thread to handle requests
+worker = Thread.new do
+  server.run_worker do |request|
+    # read the body into a buffer, to simulate some work
+    buffer = String.new(capacity: 1024)
+    request.fill_body(buffer)
+
+    accept_response
+  end
+end
+
+puts "Server running at #{config[:bind_address]}"
+puts "Press Ctrl+C to stop"
+
+# Wait for Ctrl+C
+begin
+  sleep
+rescue Interrupt
+  puts "\nShutting down..."
+  server.stop
+  worker.join
+end 
diff --git a/bin/run-server.rb b/bin/run-server.rb
@@ -47,7 +47,7 @@
   end
 end
 
-puts "Server running at #{config[:bind_address]}"
+puts "Server running at #{server.config.bind_address}"
 puts "Press Ctrl+C to stop"
 
 # Wait for Ctrl+C
diff --git a/ext/hyper_ruby/src/lib.rs b/ext/hyper_ruby/src/lib.rs
@@ -260,23 +260,72 @@ impl Server {
 
 
         rt.block_on(async move {
-            let server_task = tokio::spawn(async move {
-                let timer = hyper_util::rt::TokioTimer::new();
-                let mut builder = auto::Builder::new(hyper_util::rt::TokioExecutor::new());
-                builder.http1()
-                    .header_read_timeout(std::time::Duration::from_millis(config.recv_timeout))
-                    .timer(timer.clone());
-                builder.http2()
-                    .keep_alive_interval(std::time::Duration::from_secs(10))
-                    .timer(timer);
-
-                let listener = if config.bind_address.starts_with("unix:") {
-                    Listener::Unix(UnixListener::bind(config.bind_address.trim_start_matches("unix:")).unwrap())
-                } else {
-                    let addr: SocketAddr = config.bind_address.parse().expect("invalid address format");
-                    Listener::Tcp(TcpListener::bind(addr).await.unwrap())
-                };
+            // Instead of spawning a task, we'll run the server setup inline first to catch binding errors
+            // Setup listener and http server components
+            let timer = hyper_util::rt::TokioTimer::new();
+            let mut builder = auto::Builder::new(hyper_util::rt::TokioExecutor::new());
+            builder.http1()
+                .header_read_timeout(std::time::Duration::from_millis(config.recv_timeout))
+                .timer(timer.clone());
+            builder.http2()
+                .keep_alive_interval(std::time::Duration::from_secs(10))
+                .timer(timer);
+
+            // 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),
+                    Err(e) => {
+                        error!("Failed to bind to Unix socket {}: {}", path, e);
+                        return Err(MagnusError::new(
+                            magnus::exception::runtime_error(),
+                            format!("Failed to bind to Unix socket {}: {}", path, e)
+                        ));
+                    }
+                }
+            } else {
+                match config.bind_address.parse::<SocketAddr>() {
+                    Ok(addr) => {
+                        match TcpListener::bind(addr).await {
+                            Ok(listener) => Listener::Tcp(listener),
+                            Err(e) => {
+                                error!("Failed to bind to address {}: {}", addr, e);
+                                return Err(MagnusError::new(
+                                    magnus::exception::runtime_error(),
+                                    format!("Failed to bind to address {}: {}", addr, e)
+                                ));
+                            }
+                        }
+                    },
+                    Err(e) => {
+                        error!("Invalid address format {}: {}", config.bind_address, e);
+                        return Err(MagnusError::new(
+                            magnus::exception::runtime_error(),
+                            format!("Invalid address format {}: {}", config.bind_address, e)
+                        ));
+                    }
+                }
+            };
 
+            // Now that we have successfully bound, spawn the server task
+            let server_task = tokio::spawn(async move {
                 let graceful_shutdown = GracefulShutdown::new();
                 let mut shutdown_rx = shutdown_rx;
 
diff --git a/test/test_http.rb b/test/test_http.rb
@@ -2,6 +2,7 @@
 
 require "test_helper"
 require "httpx"
+require "fileutils"
 
 class TestHttp < HyperRubyTest
   ACCEPT_RESPONSE = HyperRuby::Response.new(202, { 'Content-Type' => 'text/plain' }, '').freeze
@@ -103,6 +104,117 @@ def test_http2_request
     end
   end
 
+  def test_address_binding_error
+    # First server binds to the port
+    server1 = HyperRuby::Server.new
+    server1.configure({ bind_address: "127.0.0.1:3020" })
+    server1.start
+
+    begin
+      # Try to start a second server on the same port, which should fail
+      server2 = HyperRuby::Server.new
+      server2.configure({ bind_address: "127.0.0.1:3020" })
+      
+      # This should raise an exception
+      error = assert_raises(RuntimeError) do
+        server2.start
+      end
+      
+      # Verify that the error message contains information about the binding failure
+      assert_match(/Failed to bind to address/, error.message)
+    ensure
+      # Clean up
+      server1.stop if server1
+    end
+  end
+
+  def test_unix_socket_cleanup
+    # Create a temporary path for the socket
+    socket_path = "/tmp/hyper_ruby_test_cleanup.sock"
+    
+    # Clean up any leftover socket file from previous test runs
+    File.unlink(socket_path) if File.exist?(socket_path)
+    
+    # First ensure that automatic cleanup works when a socket exists
+    # but is deletable - this should work without errors
+    FileUtils.touch(socket_path)
+    File.chmod(0644, socket_path)  # Ensure we have permissions
+    
+    begin
+      server = HyperRuby::Server.new
+      server.configure({ bind_address: "unix:#{socket_path}" })
+      server.start
+      
+      # If we get here, the server started correctly
+      assert File.exist?(socket_path), "Socket file should exist after server starts"
+    ensure
+      server.stop if server
+      File.unlink(socket_path) if File.exist?(socket_path)
+    end
+  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
+    # Skip if we're not root or can't modify file permissions
+    skip unless Process.uid == 0 || system("sudo -n true 2>/dev/null")
+    
+    socket_path = "/tmp/hyper_ruby_test_undeletable.sock"
+    
+    # Clean up any leftover socket file
+    File.unlink(socket_path) if File.exist?(socket_path)
+    
+    # Create a file at the socket path that can't be deleted by the current user
+    system("sudo touch #{socket_path} && sudo chmod 0000 #{socket_path}")
+    
+    begin
+      server = HyperRuby::Server.new
+      server.configure({ bind_address: "unix:#{socket_path}" })
+      
+      # This should raise an exception about not being able to remove the file
+      error = assert_raises(RuntimeError) do
+        server.start
+      end
+      
+      # Verify the error message
+      assert_match(/Failed to remove existing Unix socket file/, error.message)
+    ensure
+      # Clean up with sudo
+      system("sudo rm -f #{socket_path}") if File.exist?(socket_path)
+    end
+  end
+
+  def test_unix_socket_directory_error
+    # Create a directory instead of a socket file
+    socket_dir = "/tmp/hyper_ruby_test_dir"
+    
+    # Ensure the directory exists
+    FileUtils.mkdir_p(socket_dir)
+    
+    # Try to bind to the directory (which should fail)
+    begin
+      server = HyperRuby::Server.new
+      server.configure({ bind_address: "unix:#{socket_dir}" })
+      
+      # This should raise an exception
+      error = assert_raises(RuntimeError) do
+        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}")
+    ensure
+      # Clean up
+      FileUtils.rm_rf(socket_dir) if Dir.exist?(socket_dir)
+    end
+  end
+
   private
 
   def handler_simple(request)
