test(fork): cover concurrent prefork clients

docs/fork-safety.md

@@ -26,6 +26,42 @@ response, body sender, and cookie jar belongs to the process that created it.
 Using an inherited object raises `Wreq::ForkError`, even when the parent never
 started the runtime. wreq-ruby does not rebuild these objects.
 
+## Concurrent prefork workers on Linux
+
+The [`multiprocess_client.rb`](../test/scripts/multiprocess_client.rb) regression
+test covers a Linux prefork layout in which the parent loads wreq-ruby before it
+starts the workers, but does not create a client or initialize the request
+runtime. The master is clean and single-threaded at the fork boundary. The local
+HTTP server is forked before `require "wreq"`, so the server process never
+inherits the extension or any of its native state.
+
+The test uses two barriers. The first releases four workers to create a fresh
+`Wreq::Client` in each process. After all four workers report that their client
+exists, the second barrier releases them together to send requests. Each worker
+uses the same fresh client for two requests. After every worker exits, the
+parent creates its own fresh client and sends one final request.
+
+This works because the parent has not initialized the Tokio runtime when the
+workers fork. Under copy-on-write process semantics, each child initializes a
+separate runtime and its threads on its first request, in its own address space.
+`ProcessLocal` rejects native-backed objects inherited from another process; it
+does not stop separate children from creating their own objects and runtime
+after the fork.
+
+The test completed 10 consecutive runs on WSL2 x86_64 with Ruby 3.4.8. Those
+runs covered 40 worker processes, 80 worker requests, and 10 parent requests.
+They produced no failures, deadlocks, or `Wreq::ForkError` exceptions.
+The complete fork test file also passed with 4 runs and 100 assertions.
+
+The test server replies with `Connection: close`, so this result does not prove
+connection pooling or connection reuse across requests. It verifies that
+several prefork workers can create process-local clients and send requests when
+the parent has only loaded the extension. If the parent initializes the runtime
+before forking, the child remains unsupported even if it creates a new client.
+The server waits on a release pipe after the final request so that its `SIGCHLD`
+does not interrupt a no-GVL request that is still returning. This pipe is test
+harness coordination, not an application requirement.
+
 ## Forking after runtime initialization
 
 Once the parent starts an HTTP operation or otherwise uses the Tokio runtime, a

test/fork_test.rb

@@ -48,6 +48,21 @@ def test_loaded_extension_can_initialize_runtime_after_fork
     refute_match(/\[BUG\]|segmentation fault|panicked/i, stderr)
   end
 
+  def test_fresh_clients_can_request_from_concurrent_forked_workers
+    skip "this regression test is Linux-only" unless RbConfig::CONFIG.fetch("host_os").include?("linux")
+    skip "fork is not supported on this platform" unless Process.respond_to?(:fork)
+
+    stdout, stderr, status = run_fork_script("multiprocess_client.rb", timeout: 60)
+
+    assert status.success?, "subprocess failed with #{status.inspect}: #{stderr}"
+    assert_equal "ok\n", stdout
+    4.times do |worker|
+      assert_match(/^worker_#{worker}=ok$/, stderr)
+    end
+    assert_match(/^parent_after_workers=ok$/, stderr)
+    refute_match(/Wreq::ForkError|\[BUG\]|segmentation fault|panicked/i, stderr)
+  end
+
   def test_initialized_runtime_is_rejected_after_fork
     skip "fork is not supported on this platform" unless Process.respond_to?(:fork)
 
@@ -67,7 +82,7 @@ def test_initialized_runtime_is_rejected_after_fork
 
   private
 
-  def run_fork_script(name)
+  def run_fork_script(name, timeout: 30)
     lib_dir = File.expand_path("../lib", __dir__)
     script = File.expand_path("scripts/#{name}", __dir__)
 
@@ -82,25 +97,29 @@ def run_fork_script(name)
           err: stderr,
           pgroup: true
         )
-        status = Timeout.timeout(30) { Process.wait2(pid).last }
+        status = Timeout.timeout(timeout) { Process.wait2(pid).last }
+        kill_process_group(pid) unless status.success?
         stdout.rewind
         stderr.rewind
         return [stdout.read, stderr.read, status]
       rescue Timeout::Error
-        begin
-          Process.kill("KILL", -pid)
-        rescue Errno::ESRCH
-          nil
-        end
-
-        begin
-          Process.wait(pid)
-        rescue Errno::ECHILD
-          nil
-        end
-
+        kill_process_group(pid)
         flunk "#{name} timed out"
       end
     end
   end
+
+  def kill_process_group(pid)
+    begin
+      Process.kill("KILL", -pid)
+    rescue Errno::ESRCH
+      nil
+    end
+
+    begin
+      Process.wait(pid)
+    rescue Errno::ECHILD
+      nil
+    end
+  end
 end

test/scripts/multiprocess_client.rb

@@ -0,0 +1,161 @@
+# frozen_string_literal: true
+
+require "rbconfig"
+require "socket"
+require "timeout"
+
+$stdout.sync = true
+$stderr.sync = true
+
+WORKER_COUNT = 4
+REQUESTS_PER_WORKER = 2
+PARENT_REQUEST_PATH = "/parent/request/0"
+
+abort "multiprocess client test requires Linux" unless RbConfig::CONFIG.fetch("host_os").include?("linux")
+
+server = TCPServer.new("127.0.0.1", 0)
+url = "http://127.0.0.1:#{server.addr[1]}"
+server_release_reader, server_release_writer = IO.pipe
+server_pid = fork do
+  server_release_writer.close
+  Timeout.timeout(55) do
+    observed_paths = Array.new((WORKER_COUNT * REQUESTS_PER_WORKER) + 1) do
+      socket = server.accept
+      begin
+        request_line = socket.gets
+        abort "server received an incomplete request" unless request_line
+
+        while (line = socket.gets)
+          break if line == "\r\n"
+        end
+
+        method, path, = request_line.split(" ", 3)
+        abort "server received an unexpected method: #{method.inspect}" unless method == "GET"
+
+        socket.write(
+          "HTTP/1.1 200 OK\r\n" \
+          "Content-Length: #{path.bytesize}\r\n" \
+          "Connection: close\r\n\r\n" \
+          "#{path}"
+        )
+        path
+      ensure
+        socket.close
+      end
+    end
+
+    expected_paths = WORKER_COUNT.times.flat_map do |worker|
+      REQUESTS_PER_WORKER.times.map { |request| "/worker/#{worker}/request/#{request}" }
+    end
+    expected_paths << PARENT_REQUEST_PATH
+    abort "server received unexpected paths: #{observed_paths.inspect}" unless observed_paths.sort == expected_paths.sort
+    abort "server did not receive its release signal" unless server_release_reader.read(1) == "."
+  end
+  exit! 0
+rescue => error
+  warn "server=unexpected #{error.class}: #{error.message}"
+  exit! 3
+ensure
+  server_release_reader.close
+  server.close
+end
+server.close
+server_release_reader.close
+
+# The parent only loads the extension. It does not create a Client or initialize
+# the request runtime before forking the workers.
+require "wreq"
+
+client_start_reader, client_start_writer = IO.pipe
+client_ready_reader, client_ready_writer = IO.pipe
+request_start_reader, request_start_writer = IO.pipe
+worker_pids = WORKER_COUNT.times.map do |worker|
+  fork do
+    server_release_writer.close
+    client_start_writer.close
+    client_ready_reader.close
+    request_start_writer.close
+
+    abort "worker #{worker} did not receive its client signal" unless client_start_reader.read(1) == "."
+    client_start_reader.close
+
+    client = Wreq::Client.new(no_proxy: true, http1_only: true, timeout: 5)
+    client_ready_writer.write(".")
+    client_ready_writer.close
+
+    abort "worker #{worker} did not receive its request signal" unless request_start_reader.read(1) == "."
+    request_start_reader.close
+
+    Timeout.timeout(30) do
+      REQUESTS_PER_WORKER.times do |request|
+        path = "/worker/#{worker}/request/#{request}"
+        response = client.get("#{url}#{path}")
+        abort "worker #{worker} request #{request} failed" unless response.bytes == path
+      end
+    end
+
+    warn "worker_#{worker}=ok"
+    exit! 0
+  rescue => error
+    warn "worker_#{worker}=unexpected #{error.class}: #{error.message}"
+    exit! 2
+  ensure
+    client_start_reader.close unless client_start_reader.closed?
+    client_ready_writer.close unless client_ready_writer.closed?
+    request_start_reader.close unless request_start_reader.closed?
+  end
+end
+
+client_start_reader.close
+client_ready_writer.close
+request_start_reader.close
+
+client_start_writer.write("." * WORKER_COUNT)
+client_start_writer.close
+
+ready_workers = client_ready_reader.read(WORKER_COUNT) || ""
+client_ready_reader.close
+abort "only #{ready_workers.bytesize} workers created a Client" unless ready_workers.bytesize == WORKER_COUNT
+
+request_start_writer.write("." * WORKER_COUNT)
+request_start_writer.close
+
+failed_workers = worker_pids.filter_map do |worker_pid|
+  pid, status = Process.wait2(worker_pid)
+  [pid, status] unless status.success?
+end
+
+unless failed_workers.empty?
+  server_release_writer.close
+  begin
+    Process.kill("TERM", server_pid)
+  rescue Errno::ESRCH
+    nil
+  end
+  Process.wait(server_pid)
+  abort "workers failed: #{failed_workers.map { |pid, status| "#{pid}=#{status.inspect}" }.join(", ")}"
+end
+
+begin
+  parent_client = Wreq::Client.new(no_proxy: true, http1_only: true, timeout: 5)
+  parent_response = parent_client.get("#{url}#{PARENT_REQUEST_PATH}")
+  raise "parent request after workers failed" unless parent_response.bytes == PARENT_REQUEST_PATH
+  warn "parent_after_workers=ok"
+rescue => error
+  server_release_writer.close
+  begin
+    Process.kill("TERM", server_pid)
+  rescue Errno::ESRCH
+    nil
+  end
+  Process.wait(server_pid)
+  abort "parent_after_workers=unexpected #{error.class}: #{error.message}"
+end
+
+server_release_writer.write(".")
+server_release_writer.close
+
+_, server_status = Process.wait2(server_pid)
+abort "server failed with #{server_status.inspect}" unless server_status.success?
+
+puts "ok"