refactor(client): streamline request execution

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

src/client.rs

@@ -1,8 +1,8 @@
 mod body;
 mod param;
 mod query;
-mod req;
-pub mod resp;
+mod request;
+pub mod response;
 
 use std::{net::IpAddr, time::Duration};
 
@@ -12,7 +12,7 @@ use wreq::Proxy;
 
 use crate::{
     arch::{ProcessLocal, SUPPORTS_INTERFACE, SUPPORTS_TCP_USER_TIMEOUT},
-    client::{req::execute_request, resp::Response},
+    client::{request::execute_request, response::Response},
     cookie::Jar,
     emulate::Emulation,
     error::wreq_error,
@@ -123,6 +123,14 @@ struct Builder {
 #[magnus::wrap(class = "Wreq::Client", free_immediately, size)]
 pub struct Client(ProcessLocal<wreq::Client>);
 
+/// Borrow the process-local wrapper without bypassing its ownership check.
+impl AsRef<ProcessLocal<wreq::Client>> for Client {
+    #[inline]
+    fn as_ref(&self) -> &ProcessLocal<wreq::Client> {
+        &self.0
+    }
+}
+
 // ===== impl Builder =====
 
 impl Builder {
@@ -204,8 +212,10 @@ impl Client {
     ///
     /// Returns `Wreq::BuilderError`, `Wreq::TlsError`, or another mapped native
     /// initialization error without unwinding through Ruby.
-    pub(crate) fn default_client(ruby: &Ruby) -> Result<wreq::Client, magnus::Error> {
+    pub(crate) fn default(ruby: &Ruby) -> Result<Self, magnus::Error> {
         Self::build(ruby, Builder::default())
+            .map(ProcessLocal::new)
+            .map(Self)
     }
 
     /// Apply validated parameters and build the native client without the GVL.
@@ -381,155 +391,97 @@ impl Client {
         // Ruby exceptions must be created after the GVL has been reacquired.
         result.map_err(|err| wreq_error(ruby, err))
     }
-
-    /// Clone the native client handle in the process that created it.
-    ///
-    /// # Errors
-    ///
-    /// Returns `Wreq::ForkError` when the client was inherited from a parent
-    /// process.
-    fn native_client(&self, ruby: &Ruby) -> Result<wreq::Client, magnus::Error> {
-        self.0.get(ruby).cloned()
-    }
 }
 
 impl Client {
     /// Send a request through a newly built default client.
     ///
     /// Request arguments are validated before the native client is built, so
     /// invalid options fail without initializing a connection pool.
-    pub(crate) fn request_with_default_client(
+    pub(crate) fn request_once_from_args(
         ruby: &Ruby,
         args: &[Value],
     ) -> Result<Response, magnus::Error> {
         let ((method, url), request) = extract_request!(ruby, args, (Obj<Method>, String));
-        let client = Self::default_client(ruby)?;
-        execute_request(ruby, client, *method, url, request)
+        let client = Self::default(ruby)?;
+        execute_request(ruby, &client, *method, url, request)
     }
 
     /// Send a request with `method` through a newly built default client.
     ///
     /// Request arguments are validated before the native client is built, so
     /// invalid options fail without initializing a connection pool.
-    pub(crate) fn execute_with_default_client(
+    pub(crate) fn request_once(
         ruby: &Ruby,
         method: Method,
         args: &[Value],
     ) -> Result<Response, magnus::Error> {
         let ((url,), request) = extract_request!(ruby, args, (String,));
-        let client = Self::default_client(ruby)?;
-        execute_request(ruby, client, method, url, request)
+        let client = Self::default(ruby)?;
+        execute_request(ruby, &client, method, url, request)
     }
 
     /// Send a HTTP request.
     #[inline]
     pub fn request(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result<Response, magnus::Error> {
         let ((method, url), request) = extract_request!(ruby, args, (Obj<Method>, String));
-        execute_request(ruby, rb_self.native_client(ruby)?, *method, url, request)
+        execute_request(ruby, rb_self, *method, url, request)
     }
 
     /// Send a GET request.
     #[inline]
     pub fn get(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result<Response, magnus::Error> {
         let ((url,), request) = extract_request!(ruby, args, (String,));
-        execute_request(
-            ruby,
-            rb_self.native_client(ruby)?,
-            Method::GET,
-            url,
-            request,
-        )
+        execute_request(ruby, rb_self, Method::GET, url, request)
     }
 
     /// Send a POST request.
     #[inline]
     pub fn post(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result<Response, magnus::Error> {
         let ((url,), request) = extract_request!(ruby, args, (String,));
-        execute_request(
-            ruby,
-            rb_self.native_client(ruby)?,
-            Method::POST,
-            url,
-            request,
-        )
+        execute_request(ruby, rb_self, Method::POST, url, request)
     }
 
     /// Send a PUT request.
     #[inline]
     pub fn put(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result<Response, magnus::Error> {
         let ((url,), request) = extract_request!(ruby, args, (String,));
-        execute_request(
-            ruby,
-            rb_self.native_client(ruby)?,
-            Method::PUT,
-            url,
-            request,
-        )
+        execute_request(ruby, rb_self, Method::PUT, url, request)
     }
 
     /// Send a DELETE request.
     #[inline]
     pub fn delete(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result<Response, magnus::Error> {
         let ((url,), request) = extract_request!(ruby, args, (String,));
-        execute_request(
-            ruby,
-            rb_self.native_client(ruby)?,
-            Method::DELETE,
-            url,
-            request,
-        )
+        execute_request(ruby, rb_self, Method::DELETE, url, request)
     }
 
     /// Send a HEAD request.
     #[inline]
     pub fn head(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result<Response, magnus::Error> {
         let ((url,), request) = extract_request!(ruby, args, (String,));
-        execute_request(
-            ruby,
-            rb_self.native_client(ruby)?,
-            Method::HEAD,
-            url,
-            request,
-        )
+        execute_request(ruby, rb_self, Method::HEAD, url, request)
     }
 
     /// Send an OPTIONS request.
     #[inline]
     pub fn options(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result<Response, magnus::Error> {
         let ((url,), request) = extract_request!(ruby, args, (String,));
-        execute_request(
-            ruby,
-            rb_self.native_client(ruby)?,
-            Method::OPTIONS,
-            url,
-            request,
-        )
+        execute_request(ruby, rb_self, Method::OPTIONS, url, request)
     }
 
     /// Send a TRACE request.
     #[inline]
     pub fn trace(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result<Response, magnus::Error> {
         let ((url,), request) = extract_request!(ruby, args, (String,));
-        execute_request(
-            ruby,
-            rb_self.native_client(ruby)?,
-            Method::TRACE,
-            url,
-            request,
-        )
+        execute_request(ruby, rb_self, Method::TRACE, url, request)
     }
 
     /// Send a PATCH request.
     #[inline]
     pub fn patch(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result<Response, magnus::Error> {
         let ((url,), request) = extract_request!(ruby, args, (String,));
-        execute_request(
-            ruby,
-            rb_self.native_client(ruby)?,
-            Method::PATCH,
-            url,
-            request,
-        )
+        execute_request(ruby, rb_self, Method::PATCH, url, request)
     }
 }
 
@@ -546,7 +498,7 @@ pub fn include(ruby: &Ruby, gem_module: &RModule) -> Result<(), magnus::Error> {
     client_class.define_method("trace", method!(Client::trace, -1))?;
     client_class.define_method("patch", method!(Client::patch, -1))?;
 
-    resp::include(ruby, gem_module)?;
+    response::include(ruby, gem_module)?;
     body::include(ruby, gem_module)?;
     Ok(())
 }

src/client/req.rs → src/client/request.rs

@@ -1,18 +1,21 @@
+//! Request option parsing, native request construction, and execution.
+
 use std::{net::IpAddr, time::Duration};
 
 use ::serde::Deserialize;
 use http::header;
 use magnus::{RHash, TryConvert, typed_data::Obj, value::ReprValue};
-use wreq::{Client, Proxy};
+use wreq::Proxy;
 
 use super::body::{Body, form::Form, json::Json};
 use crate::{
-    arch::SUPPORTS_INTERFACE,
-    client::{query::Query, resp::Response},
+    arch::{ProcessLocal, SUPPORTS_INTERFACE},
+    client::{query::Query, response::Response},
     cookie::Cookies,
     emulate::Emulation,
     error::wreq_error,
     extractor::Extractor,
+    gvl,
     header::{Headers, OrigHeaders},
     http::{Method, Version},
     options::{NativeOption, Options},
@@ -21,7 +24,6 @@ use crate::{
 
 /// The parameters for a request.
 #[derive(Default, Deserialize)]
-#[non_exhaustive]
 pub struct Request {
     /// The emulation option for the request.
     #[serde(default)]
@@ -177,14 +179,31 @@ impl Request {
     }
 }
 
-pub fn execute_request<U: AsRef<str>>(
+/// Build and execute one request with a process-local client.
+///
+/// Request builder configuration is synchronous and runs without the GVL.
+/// Only the completed request's network future is submitted to Tokio.
+///
+/// # Errors
+///
+/// Returns `Wreq::ForkError` for an inherited client, a mapped native error when
+/// building or sending the request fails, or an interruption/runtime error from
+/// [`rt::block_on`].
+pub fn execute_request<U, C>(
     ruby: &magnus::Ruby,
-    client: Client,
+    client: &C,
     method: Method,
     url: U,
     mut request: Request,
-) -> Result<Response, magnus::Error> {
-    rt::block_on(ruby, async move {
+) -> Result<Response, magnus::Error>
+where
+    U: AsRef<str>,
+    C: AsRef<ProcessLocal<wreq::Client>>,
+{
+    // Process ownership failures construct a Ruby exception and require the GVL.
+    let client = client.as_ref().get(ruby)?;
+
+    let request = gvl::nogvl(|| {
         let mut builder = client.request(method.into_ffi(), url.as_ref());
 
         // Emulation options.
@@ -304,8 +323,11 @@ pub fn execute_request<U: AsRef<str>>(
             builder = builder.body(wreq::Body::from(body));
         }
 
-        // Send request.
-        builder.send().await.map(Response::new)
-    })?
-    .map_err(|err| wreq_error(ruby, err))
+        builder.build()
+    })
+    .map_err(|err| wreq_error(ruby, err))?;
+
+    rt::block_on(ruby, client.execute(request))?
+        .map(Response::new)
+        .map_err(|err| wreq_error(ruby, err))
 }

src/client/resp.rs → src/client/response.rs

@@ -1,3 +1,5 @@
+//! Ruby response bindings and response-body lifecycle management.
+
 use std::{net::SocketAddr, sync::Arc};
 
 use arc_swap::ArcSwapOption;