fix(runtime): reject use after fork

Cargo.toml

@@ -47,6 +47,9 @@ http = "1.4.1"
 http-body-util = "0.1.3"
 futures-util = { version = "0.3.32", default-features = false }
 
+[target.'cfg(unix)'.dependencies]
+forkguard = { version = "0.1.3", features = ["atfork"] }
+
 [dev-dependencies]
 magnus = { version = "0.8.2", features = ["embed"] }
 

lib/wreq.rb

@@ -27,6 +27,10 @@ module Wreq
     # raise ArgumentError. Known values retain the error class from their Ruby
     # or native conversion, such as TypeError or Wreq::BuilderError. Validation
     # finishes before network I/O.
+    #
+    # Requests made in a child process forked after wreq-ruby was loaded raise
+    # Wreq::ForkError. Load wreq-ruby inside each worker after it has been
+    # forked.
 
     # Send an HTTP request.
     #

lib/wreq_ruby/body.rb

@@ -17,6 +17,8 @@ module Wreq
     #
     # A sender can be attached to one request. Closing it prevents further writes but
     # retains queued chunks so a request attached afterward can still drain them.
+    # Calling {#push} raises Wreq::ForkError in a child process forked after
+    # wreq-ruby was loaded.
     class BodySender
       # Create a bounded request-body sender.
       #
@@ -33,6 +35,7 @@ def self.new(capacity = 8)
       # @param data [String] binary chunk
       # @return [nil]
       # @raise [IOError] if the sender or receiving side is closed
+      # @raise [Wreq::ForkError] if the process inherited wreq-ruby through fork
       def push(data)
       end
 

lib/wreq_ruby/client.rb

@@ -17,6 +17,10 @@ module Wreq
     # native conversion, such as TypeError or Wreq::BuilderError. Request
     # validation finishes before network I/O.
     #
+    # Clients cannot be created or used in a child process forked after wreq-ruby
+    # was loaded. These operations raise Wreq::ForkError. Load wreq-ruby inside
+    # each worker after it has been forked.
+    #
     # @example Basic usage
     #   client = Wreq::Client.new
     #   # Use client for HTTP requests
@@ -165,6 +169,7 @@ class Client
       #   value cannot be converted or validated.
       # @raise [Wreq::BuilderError, Wreq::TlsError] if the native client cannot
       #   be initialized.
+      # @raise [Wreq::ForkError] if the process inherited wreq-ruby through fork.
       #
       # @example Minimal client
       #   client = Wreq::Client.new

lib/wreq_ruby/error.rb

@@ -7,6 +7,18 @@ module Wreq
     # Memory allocation failed.
     class MemoryError < StandardError; end
 
+    # The native extension was inherited from a parent process.
+    #
+    # Raised when wreq-ruby is used in a child process forked after the
+    # extension was loaded. Its process-global native state cannot be reused
+    # safely in the child.
+    #
+    # @example
+    #   Process.fork do
+    #     Wreq::Client.new # Raises when the parent loaded wreq-ruby.
+    #   end
+    class ForkError < RuntimeError; end
+
     # Network connection errors
 
     # Connection to the server failed.

lib/wreq_ruby/response.rb

@@ -8,6 +8,9 @@ module Wreq
     # access to HTTP response data including status codes, headers, body
     # content, and streaming capabilities.
     #
+    # Reading or streaming the body raises Wreq::ForkError in a child process
+    # forked after wreq-ruby was loaded.
+    #
     # @example Basic response handling
     #   response = client.get("https://api.example.com")
     #   puts response.status.as_int  # => 200

src/arch.rs

@@ -27,6 +27,64 @@ pub(crate) const SUPPORTS_INTERFACE: bool = cfg!(any(
     target_os = "watchos",
 ));
 
+#[cfg(unix)]
+mod unix {
+    use std::{io, process, sync::OnceLock};
+
+    /// Process state captured when the extension initializes.
+    ///
+    /// The atfork guard uses a POSIX child handler to advance an atomic fork
+    /// generation without running Ruby code.
+    /// https://pubs.opengroup.org/onlinepubs/9799919799/functions/pthread_atfork.html
+    struct ForkGuard {
+        detector: forkguard::Guard,
+        owner_pid: u32,
+    }
+
+    impl ForkGuard {
+        /// Create a guard and register fork detection with the process.
+        fn new() -> io::Result<Self> {
+            forkguard::Guard::try_new()
+                .map(|detector| Self {
+                    detector,
+                    owner_pid: process::id(),
+                })
+                .map_err(|error| io::Error::from_raw_os_error(error.code().get()))
+        }
+
+        /// Return process IDs when this guard was inherited through a fork.
+        fn forked_process_ids(&self) -> Option<(u32, u32)> {
+            // Keep the stored generation unchanged so every runtime access in
+            // the child remains rejected. Cloning the detector copies one usize.
+            let mut detector = self.detector.clone();
+            detector
+                .detected_fork()
+                .then(|| (self.owner_pid, process::id()))
+        }
+    }
+
+    static FORK_GUARD: OnceLock<ForkGuard> = OnceLock::new();
+
+    /// Register process fork tracking before the extension exposes its API.
+    pub(crate) fn initialize_fork_tracking() -> io::Result<()> {
+        if FORK_GUARD.get().is_some() {
+            return Ok(());
+        }
+
+        let guard = ForkGuard::new()?;
+        let _ = FORK_GUARD.set(guard);
+        Ok(())
+    }
+
+    /// Return process IDs only when this process inherited the extension.
+    pub(crate) fn forked_process_ids() -> Option<(u32, u32)> {
+        FORK_GUARD.get().and_then(ForkGuard::forked_process_ids)
+    }
+}
+
+#[cfg(unix)]
+pub(crate) use unix::{forked_process_ids, initialize_fork_tracking};
+
 #[cfg(all(target_os = "windows", target_env = "gnu"))]
 mod windows_gnu {
     //! Windows GNU support.

src/client.rs

@@ -21,6 +21,7 @@ use crate::{
     header::{Headers, OrigHeaders, UserAgent},
     http::Method,
     options::{NativeOption, Options},
+    rt,
 };
 
 /// A builder for `Client`.
@@ -214,8 +215,12 @@ impl Client {
     ///
     /// # Errors
     ///
-    /// Maps native build failures only after the GVL has been reacquired.
+    /// Returns `Wreq::ForkError` before touching native client state when the
+    /// extension was inherited from a parent process. Maps native build
+    /// failures only after the GVL has been reacquired.
     fn build(ruby: &Ruby, mut params: Builder) -> Result<Self, magnus::Error> {
+        rt::ensure_current(ruby)?;
+
         let result = gvl::nogvl(|| {
             let mut builder = wreq::Client::builder();
 

src/client/resp.rs

@@ -65,6 +65,8 @@ impl Response {
 
     /// Internal method to get the wreq::Response, optionally streaming the body.
     fn response(&self, ruby: &Ruby, stream: bool) -> Result<wreq::Response, Error> {
+        rt::ensure_current(ruby)?;
+
         let build_response = |body: wreq::Body| -> wreq::Response {
             let mut response = HttpResponse::new(body);
             *response.version_mut() = self.version.into_ffi();

src/error.rs

@@ -56,6 +56,7 @@ macro_rules! map_wreq_error {
 
 // System-level and runtime errors
 define_exception!(MEMORY, "MemoryError", exception_runtime_error);
+define_exception!(FORK_ERROR, "ForkError", exception_runtime_error);
 
 // Network connection errors
 define_exception!(CONNECTION_ERROR, "ConnectionError", exception_runtime_error);
@@ -94,6 +95,26 @@ pub fn interrupt_error(ruby: &Ruby) -> MagnusError {
     MagnusError::new(ruby.exception_interrupt(), "request interrupted")
 }
 
+/// Build `Wreq::ForkError` without touching inherited native state.
+#[cfg(unix)]
+pub fn fork_error(ruby: &Ruby, owner_pid: u32, current_pid: u32) -> MagnusError {
+    MagnusError::new(
+        ruby.get_inner(&FORK_ERROR),
+        format!(
+            "wreq loaded in process {owner_pid} cannot be used after fork in process {current_pid}"
+        ),
+    )
+}
+
+/// Map a failed process-fork handler registration to `Wreq::ForkError`.
+#[cfg(unix)]
+pub fn fork_handler_error(ruby: &Ruby, err: &std::io::Error) -> MagnusError {
+    MagnusError::new(
+        ruby.get_inner(&FORK_ERROR),
+        format!("failed to initialize process fork tracking: {err}"),
+    )
+}
+
 /// Map a Tokio runtime initialization failure to `Wreq::BuilderError`.
 pub fn runtime_initialization_error(ruby: &Ruby, err: &std::io::Error) -> MagnusError {
     MagnusError::new(
@@ -235,6 +256,13 @@ pub fn include(ruby: &Ruby, gem_module: &RModule) -> Result<(), MagnusError> {
         "MemoryError",
         exception_runtime_error
     );
+    initialize_exception!(
+        ruby,
+        gem_module,
+        FORK_ERROR,
+        "ForkError",
+        exception_runtime_error
+    );
     initialize_exception!(
         ruby,
         gem_module,

src/lib.rs

@@ -85,6 +85,7 @@ pub fn patch(ruby: &Ruby, args: &[Value]) -> Result<Response, magnus::Error> {
 fn init(ruby: &Ruby) -> Result<(), Error> {
     let gem_module = ruby.define_module(RUBY_MODULE_NAME)?;
     gem_module.const_set("VERSION", VERSION)?;
+    error::include(ruby, &gem_module)?;
     gem_module.define_module_function("request", magnus::function!(request, -1))?;
     gem_module.define_module_function("get", magnus::function!(get, -1))?;
     gem_module.define_module_function("post", magnus::function!(post, -1))?;
@@ -99,6 +100,7 @@ fn init(ruby: &Ruby) -> Result<(), Error> {
     cookie::include(ruby, &gem_module)?;
     client::include(ruby, &gem_module)?;
     emulate::include(ruby, &gem_module)?;
-    error::include(ruby, &gem_module)?;
+    #[cfg(unix)]
+    rt::initialize(ruby)?;
     Ok(())
 }

src/macros.rs

@@ -134,6 +134,7 @@ macro_rules! extract_request {
         let args = magnus::scan_args::scan_args::<$required, (), (), (), magnus::RHash, ()>($args)?;
         let required = args.required;
         let ruby = magnus::Ruby::get_with(args.keywords);
+        crate::rt::ensure_current(&ruby)?;
         let request = crate::client::req::Request::new(&ruby, args.keywords)?;
         (required, request)
     }};

src/rt.rs

@@ -1,42 +1,79 @@
-use std::sync::LazyLock;
+use std::{io, sync::OnceLock};
 
 use magnus::Ruby;
-use tokio::runtime::{Builder, Runtime};
+use tokio::runtime::{Builder, Runtime as TokioRuntime};
 
 use crate::{
     error::{interrupt_error, runtime_initialization_error},
     gvl,
 };
 
-/// Initialize the global runtime lazily and preserve failures for Ruby.
-static RUNTIME: LazyLock<Result<Runtime, std::io::Error>> = LazyLock::new(|| {
-    let mut builder = Builder::new_multi_thread();
+#[cfg(unix)]
+use crate::{
+    arch,
+    error::{fork_error, fork_handler_error},
+};
 
-    builder.enable_all().build()
-});
+/// Initialize the global runtime lazily and preserve failures for Ruby.
+static RUNTIME: OnceLock<Result<TokioRuntime, io::Error>> = OnceLock::new();
 
 enum BlockOnError<E> {
     Interrupted,
     Future(E),
 }
 
-/// Block on a future to completion on the global Tokio runtime.
+/// Register fork tracking while the native extension is being loaded.
+///
+/// # Errors
+///
+/// Returns `Wreq::ForkError` if the platform cannot install its child-process
+/// callback.
+#[cfg(unix)]
+pub fn initialize(ruby: &Ruby) -> Result<(), magnus::Error> {
+    arch::initialize_fork_tracking().map_err(|err| fork_handler_error(ruby, &err))
+}
+
+/// Reject a child process that inherited the loaded native extension.
+///
+/// # Errors
+///
+/// Returns `Wreq::ForkError` when the extension was loaded before the current
+/// process was forked.
+pub fn ensure_current(ruby: &Ruby) -> Result<(), magnus::Error> {
+    #[cfg(unix)]
+    if let Some((owner_pid, current_pid)) = arch::forked_process_ids() {
+        return Err(fork_error(ruby, owner_pid, current_pid));
+    }
+
+    #[cfg(not(unix))]
+    let _ = ruby;
+
+    Ok(())
+}
+
+/// Block on a future to completion on the current process's global Tokio runtime.
 ///
 /// The future runs without Ruby's GVL, so it must not construct Ruby objects or
 /// Ruby exceptions. Convert Rust errors back into Ruby errors after the GVL has
 /// been reacquired.
 ///
 /// # Errors
 ///
-/// Returns `Wreq::BuilderError` if the Tokio runtime cannot be initialized,
-/// Ruby's standard `Interrupt` if Ruby interrupts the request, or the error produced
+/// Returns `Wreq::ForkError` if the extension belongs to a parent process,
+/// `Wreq::BuilderError` if the Tokio runtime cannot be initialized, Ruby's
+/// standard `Interrupt` if Ruby interrupts the request, or the error produced
 /// by `map_err` if the future fails.
 pub fn try_block_on<F, T, E, M>(ruby: &Ruby, future: F, map_err: M) -> Result<T, magnus::Error>
 where
     F: Future<Output = Result<T, E>>,
     M: FnOnce(&Ruby, E) -> magnus::Error,
 {
+    ensure_current(ruby)?;
     let runtime = RUNTIME
+        .get_or_init(|| {
+            let mut builder = Builder::new_multi_thread();
+            builder.enable_all().build()
+        })
         .as_ref()
         .map_err(|err| runtime_initialization_error(ruby, err))?;
     let result = gvl::nogvl_cancellable(|flag| {