From 072994d84044ca8f1a7b201f9f7a6e22d3c221e0 Mon Sep 17 00:00:00 2001 From: gngpp Date: Mon, 17 Aug 2026 02:07:20 +0800 Subject: [PATCH] fix(runtime): allow prefork loading before runtime init --- docs/fork-safety.md | 81 ++++++----- docs/interrupt-handling.md | 7 +- lib/wreq.rb | 48 +++++-- lib/wreq_ruby/body.rb | 15 +- lib/wreq_ruby/client.rb | 32 +++-- lib/wreq_ruby/cookie.rb | 9 ++ lib/wreq_ruby/error.rb | 12 +- lib/wreq_ruby/response.rb | 30 +++- src/arch.rs | 120 +++++++++++----- src/client.rs | 114 ++++++++++----- src/client/body/stream.rs | 39 ++---- src/client/req.rs | 241 ++++++++++++++++---------------- src/client/resp.rs | 156 +++++++++++++-------- src/cookie.rs | 58 ++++++-- src/error.rs | 2 +- src/lib.rs | 2 - src/macros.rs | 1 - src/rt.rs | 56 +++----- test/fork_test.rb | 36 ++++- test/scripts/fork_safety.rb | 109 +++++++++------ test/scripts/prefork_runtime.rb | 95 +++++++++++++ 21 files changed, 810 insertions(+), 453 deletions(-) create mode 100644 test/scripts/prefork_runtime.rb diff --git a/docs/fork-safety.md b/docs/fork-safety.md index 0dfce5c..c0a2a52 100644 --- a/docs/fork-safety.md +++ b/docs/fork-safety.md @@ -1,34 +1,51 @@ # Fork safety -## Why inherited clients are rejected - -wreq-ruby uses a process-wide Tokio runtime and connection pool. `fork` copies -the parent's memory, but only the thread that called `fork` continues in the -child. Tokio's worker threads are gone, and its inherited tasks, locks, and -connections are not safe to reuse. - -If the parent has already loaded wreq-ruby, native HTTP operations in the child -raise `Wreq::ForkError`. This applies to new and existing clients, module -request methods, streaming request bodies, and response methods backed by native -state. Retrying the operation in the same child raises the same error. Read-only -response metadata such as status, headers, and captured TLS information remains -available. - -The parent can continue using its clients. When inherited Ruby objects are -collected in the child, their native runtime state is left for the operating -system to reclaim when the process exits. - -## HTTP work in forked children is unsupported - -A process created with `fork` must not start or continue HTTP work through -wreq-ruby, even when it first loads the extension after the fork. If the parent -loaded wreq-ruby, native HTTP operations in the child raise `Wreq::ForkError`. - -When the extension was not present in the parent, no wreq-ruby state or fork -marker reaches the child. The extension cannot reliably distinguish that child -from a newly started process, so this unsupported path cannot guarantee a Ruby -error and may fail inside platform libraries. - -Prefork servers should use an `exec`- or spawn-based worker model when workers -need wreq-ruby. Requiring the extension again does not reset inherited runtime -state, and there is no `after_fork!` hook. +## Prefork checklist + +- `require "wreq"` may run in the parent before workers fork. +- Create each `Wreq::Client`, `Wreq::Jar`, and `Wreq::BodySender` in the worker + that will use it. +- Keep each `Wreq::Response` in the process that received it. +- Do not start requests or push streaming body data in the parent before workers + fork. +- If the parent must use wreq-ruby first, start workers with `spawn` or `exec` + instead of `fork`. + +wreq-ruby checks process ownership whenever it exposes guarded native state. It +does not copy, reset, or rebuild inherited objects. + +## Loading before fork + +wreq-ruby creates its process-wide Tokio runtime on the first operation that +needs it. Requiring the gem does not initialize the runtime, so a prefork server +may load wreq-ruby during boot. Each worker can then create its own runtime on +its first request without an `after_fork!` hook. + +Create clients and other native-backed objects inside the worker. Each client, +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. + +## Forking after runtime initialization + +Once the parent starts an HTTP operation or otherwise uses the Tokio runtime, a +forked child must not reuse it. Tokio's worker threads do not survive `fork`, and +the inherited connection pool may refer to those missing threads. + +Operations that need the inherited runtime raise `Wreq::ForkError`. This +includes requests through new or existing clients, module request methods, and +streaming request writes. Constructing a new client, body sender, or cookie jar +does not use the runtime, but runtime-backed operations remain unavailable in +that child. Retrying them raises the same error. + +An inherited `Wreq::Response` cannot be used at all. This includes status, +headers, socket addresses, TLS information, and body methods. Values copied out +before the fork, such as a `Wreq::StatusCode` or `Wreq::TlsInfo`, are separate +objects and do not retain access to the response. + +The parent remains usable. Native objects collected in the child do not destroy +state owned by the parent process. + +Use a spawn- or exec-based worker when the parent must perform HTTP work before +workers start. Requiring the extension again cannot replace an inherited +runtime. diff --git a/docs/interrupt-handling.md b/docs/interrupt-handling.md index 660be00..a47924c 100644 --- a/docs/interrupt-handling.md +++ b/docs/interrupt-handling.md @@ -84,9 +84,10 @@ to the Ruby-owned thread with the GVL may [`src/rt.rs`](../src/rt.rs) map a wreq-owned cancellation to the `Wreq::InterruptError` defined in [`src/error.rs`](../src/error.rs). -Keep this conversion centralized in `rt::try_block_on`. Request, response, and -body operations may call `try_block_on`, but they must not construct their own -Ruby cancellation exception. +Keep cancellation conversion centralized in `rt::block_on`. Request, response, +and body operations may call `block_on`, but they must not construct their own +Ruby cancellation exception. `block_on` returns a future's native error +unchanged so the caller can convert it after the GVL has been reacquired. These forms are forbidden for wreq-owned cancellation: diff --git a/lib/wreq.rb b/lib/wreq.rb index 8fca103..3d8de6f 100644 --- a/lib/wreq.rb +++ b/lib/wreq.rb @@ -19,6 +19,33 @@ require_relative "wreq_ruby/cookie" unless defined?(Wreq) + # An HTTP client backed by a lazily initialized, process-wide Tokio runtime. + # + # Loading wreq-ruby before `fork` is supported. The parent must not send a + # request or perform another operation that starts the runtime before workers + # are forked. Create clients and begin HTTP work inside each worker so it gets + # its own runtime and connection pool. Clients, responses, body senders, and + # cookie jars belong to the process that created them and must be recreated + # in the worker. wreq-ruby does not rebuild inherited objects. + # + # Accessing an inherited native-backed object raises Wreq::ForkError even if + # the parent did not start the runtime. If the parent did start it, the child + # also cannot perform new runtime-backed operations. Retrying does not replace + # either kind of inherited state. Use `spawn` or `exec`, or move the parent's + # HTTP work until after the workers have been forked. + # + # @example Preload the extension, then start HTTP work in the worker + # require "wreq" + # + # Process.fork do + # client = Wreq::Client.new + # response = client.get("https://example.com") + # puts response.status + # end + # + # @note Fork safety Create clients, cookie jars, and body senders inside the + # worker that uses them. Do not carry responses across `fork`. + # @see https://github.com/SearchApi/wreq-ruby/blob/main/docs/fork-safety.md module Wreq # Current wreq gem version. # @return [String] @@ -29,9 +56,6 @@ 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. - # - # If a child process inherits wreq-ruby from its parent, requests raise - # Wreq::ForkError. Require wreq after the worker has been forked. # Send an HTTP request. # @@ -64,7 +88,7 @@ module Wreq # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the child inherited an initialized wreq-ruby runtime def self.request(method, url, **options) end @@ -98,7 +122,7 @@ def self.request(method, url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the child inherited an initialized wreq-ruby runtime def self.get(url, **options) end @@ -132,7 +156,7 @@ def self.get(url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the child inherited an initialized wreq-ruby runtime def self.head(url, **options) end @@ -166,7 +190,7 @@ def self.head(url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the child inherited an initialized wreq-ruby runtime def self.post(url, **options) end @@ -200,7 +224,7 @@ def self.post(url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the child inherited an initialized wreq-ruby runtime def self.put(url, **options) end @@ -234,7 +258,7 @@ def self.put(url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the child inherited an initialized wreq-ruby runtime def self.delete(url, **options) end @@ -268,7 +292,7 @@ def self.delete(url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the child inherited an initialized wreq-ruby runtime def self.options(url, **options) end @@ -302,7 +326,7 @@ def self.options(url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the child inherited an initialized wreq-ruby runtime def self.trace(url, **options) end @@ -336,7 +360,7 @@ def self.trace(url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the child inherited an initialized wreq-ruby runtime def self.patch(url, **options) end end diff --git a/lib/wreq_ruby/body.rb b/lib/wreq_ruby/body.rb index 1860d93..a78c976 100644 --- a/lib/wreq_ruby/body.rb +++ b/lib/wreq_ruby/body.rb @@ -17,8 +17,12 @@ 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. - # Creating or using a sender raises Wreq::ForkError if the child inherited - # wreq-ruby from its parent. + # Creating a sender does not initialize Tokio. An inherited sender raises + # Wreq::ForkError before its channel is accessed. A new sender can be + # created in a child, but pushing data also requires a usable runtime. + # + # @note Fork safety Create each sender in the worker that writes to it. + # Do not pass a sender through `fork`. class BodySender # Create a bounded request-body sender. # @@ -27,7 +31,6 @@ class BodySender # @return [Wreq::BodySender] A streaming request body sender # @raise [ArgumentError] if capacity is zero, negative, or too large # @raise [TypeError] if capacity is not an Integer - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent def self.new(capacity = 8) end @@ -36,7 +39,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 child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the sender or runtime belongs to the parent process def push(data) end @@ -45,7 +48,7 @@ def push(data) # This operation is idempotent. # # @return [nil] - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the sender belongs to the parent process def close end @@ -55,7 +58,7 @@ def close # the receiving side. # # @return [Boolean] - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the sender belongs to the parent process def closed? end end diff --git a/lib/wreq_ruby/client.rb b/lib/wreq_ruby/client.rb index 31cbeb1..9e62269 100644 --- a/lib/wreq_ruby/client.rb +++ b/lib/wreq_ruby/client.rb @@ -17,9 +17,14 @@ module Wreq # native conversion, such as TypeError or Wreq::BuilderError. Request # validation finishes before network I/O. # - # A child process cannot create or use a client if it inherited wreq-ruby - # from its parent. These calls raise Wreq::ForkError before accessing the - # native runtime. Require wreq after the worker has been forked. + # A client belongs to the process that created it. An inherited client + # raises Wreq::ForkError before its connection pool is accessed. Loading + # the gem before fork is supported, but clients must be created inside the + # worker. If the parent already started the runtime, new clients can be + # constructed in the child but cannot send requests. + # + # @note Fork safety Create each client in the worker that uses it. An + # inherited client is never rebuilt automatically. # # @example Basic usage # client = Wreq::Client.new @@ -174,8 +179,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 child inherited wreq-ruby from its parent - # + # @raise [Wreq::ForkError] if :cookie_provider belongs to a parent process. # @example Minimal client # client = Wreq::Client.new # @@ -290,7 +294,7 @@ def self.new(**options) # or unavailable on the current platform # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the client or runtime belongs to the parent process def request(method, url, **options) end @@ -324,7 +328,7 @@ def request(method, url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the client or runtime belongs to the parent process def get(url, **options) end @@ -358,7 +362,7 @@ def get(url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the client or runtime belongs to the parent process def head(url, **options) end @@ -392,7 +396,7 @@ def head(url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the client or runtime belongs to the parent process def post(url, **options) end @@ -426,7 +430,7 @@ def post(url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the client or runtime belongs to the parent process def put(url, **options) end @@ -460,7 +464,7 @@ def put(url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the client or runtime belongs to the parent process def delete(url, **options) end @@ -494,7 +498,7 @@ def delete(url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the client or runtime belongs to the parent process def options(url, **options) end @@ -528,7 +532,7 @@ def options(url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the client or runtime belongs to the parent process def trace(url, **options) end @@ -562,7 +566,7 @@ def trace(url, **options) # @return [Wreq::Response] HTTP response # @raise [TypeError, ArgumentError, Wreq::BuilderError] if a known option # value cannot be converted, validated, or built - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the client or runtime belongs to the parent process def patch(url, **options) end end diff --git a/lib/wreq_ruby/cookie.rb b/lib/wreq_ruby/cookie.rb index e13fdb7..87babaf 100644 --- a/lib/wreq_ruby/cookie.rb +++ b/lib/wreq_ruby/cookie.rb @@ -158,6 +158,11 @@ def to_s # Stores cookies for reuse across requests. # # Pass a Jar to Wreq::Client as `cookie_provider` to share its cookies. + # A jar belongs to the process that created it and cannot be inherited + # across `fork`. + # + # @note Fork safety Create a new jar in each worker. wreq-ruby does not + # copy cookies from an inherited jar. class Jar # Creates an empty cookie jar. # @return [Wreq::Jar] @@ -166,6 +171,7 @@ def self.new # Returns all stored cookies. # @return [Array] + # @raise [Wreq::ForkError] if the jar belongs to the parent process def get_all end @@ -174,6 +180,7 @@ def get_all # @param url [String] URL that scopes the cookie # @return [void] # @raise [TypeError] if cookie is neither a String nor Wreq::Cookie + # @raise [Wreq::ForkError] if the jar belongs to the parent process def add(cookie, url) end @@ -181,11 +188,13 @@ def add(cookie, url) # @param name [String] # @param url [String] # @return [void] + # @raise [Wreq::ForkError] if the jar belongs to the parent process def remove(name, url) end # Clear all cookies from the jar. # @return [void] + # @raise [Wreq::ForkError] if the jar belongs to the parent process def clear end end diff --git a/lib/wreq_ruby/error.rb b/lib/wreq_ruby/error.rb index f1e42f1..4a9c121 100644 --- a/lib/wreq_ruby/error.rb +++ b/lib/wreq_ruby/error.rb @@ -11,15 +11,17 @@ class InterruptError < Interrupt; end # Memory allocation failed. class MemoryError < StandardError; end - # The child process inherited wreq-ruby from its parent. + # The child process tried to use native state created by its parent. # - # Tokio worker threads do not survive fork, and inherited pooled - # connections are not safe to reuse. This error is raised before a child - # can access that state. + # Tokio worker threads do not survive fork. Inherited connection pools, + # locks, channels, and response state are also unsafe to use. wreq-ruby + # raises this error before exposing them. Loading the gem before fork is + # supported, but native-backed objects must be created in each worker. # # @example + # client = Wreq::Client.new # Process.fork do - # Wreq::Client.new # Raises if the parent loaded wreq-ruby. + # client.get("https://example.com") # Raises in the child. # end # @see https://github.com/SearchApi/wreq-ruby/blob/main/docs/fork-safety.md class ForkError < RuntimeError; end diff --git a/lib/wreq_ruby/response.rb b/lib/wreq_ruby/response.rb index 4985495..1dd6729 100644 --- a/lib/wreq_ruby/response.rb +++ b/lib/wreq_ruby/response.rb @@ -8,8 +8,12 @@ module Wreq # access to HTTP response data including status codes, headers, body # content, and streaming capabilities. # - # Body methods raise Wreq::ForkError if the child inherited wreq-ruby from - # its parent. + # A response belongs to the process that received it. Accessing its metadata + # or body after inheriting it from a parent raises Wreq::ForkError. + # + # @note Fork safety Keep each response in the process that received it. + # Issue a new request in the worker instead of carrying a response through + # `fork`. # # @example Basic response handling # response = client.get("https://api.example.com") @@ -29,6 +33,7 @@ class Response # Get the HTTP status code as an integer. # # @return [Integer] Status code (e.g., 200, 404, 500) + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example # response.code # => 200 def code @@ -37,6 +42,7 @@ def code # Get the HTTP status code object. # # @return [Wreq::StatusCode] Status code wrapper with helper methods + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example # status = response.status # status.success? # => true @@ -46,6 +52,7 @@ def status # Get the HTTP protocol version used. # # @return [Wreq::Version] HTTP version (HTTP/1.1, HTTP/2, etc.) + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example # response.version # => Wreq::Version::HTTP_11 def version @@ -54,6 +61,7 @@ def version # Get the final URL after redirects. # # @return [String] The final URL + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example # response.url # => "https://example.com/final-page" def url @@ -62,6 +70,7 @@ def url # Get the content length if known. # # @return [Integer, nil] Content length in bytes, or nil if unknown + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example # response.content_length # => 1024 def content_length @@ -75,6 +84,7 @@ def content_length # response or a later snapshot, and object identity is not guaranteed. # # @return [Wreq::Headers] Response headers + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example # response.headers.get("content-type") # => "application/json" def headers @@ -83,6 +93,7 @@ def headers # Get the local socket address. # # @return [String, nil] Local address (e.g., "127.0.0.1:54321"), or nil + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example # response.local_addr # => "192.168.1.100:54321" def local_addr @@ -91,6 +102,7 @@ def local_addr # Get the remote socket address. # # @return [String, nil] Remote address (e.g., "93.184.216.34:443"), or nil + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example # response.remote_addr # => "93.184.216.34:443" def remote_addr @@ -101,6 +113,7 @@ def remote_addr # Invalid `Set-Cookie` values are skipped. # # @return [Array] Parsed response cookies + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example # response.cookies.each do |cookie| # puts "#{cookie.name}=#{cookie.value}" @@ -110,7 +123,7 @@ def cookies # Get the response bytes as a binary string. # @return [String] Response body as binary data - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example # binary_data = response.bytes # puts binary_data.size # => 1024 @@ -126,7 +139,7 @@ def bytes # html = response.text("ISO-8859-1") # puts html # @raise [Wreq::DecodingError] if body cannot be decoded with the specified encoding - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the response belongs to the parent process def text(default_encoding = "UTF-8") end @@ -137,7 +150,7 @@ def text(default_encoding = "UTF-8") # # @return [Object] Parsed JSON (Hash, Array, String, Integer, Float, Boolean, nil) # @raise [Wreq::DecodingError] if body is not valid JSON - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example # data = response.json # puts data["key"] @@ -155,7 +168,7 @@ def json # @raise [LocalJumpError] if called without a block # @raise [Wreq::TimeoutError, Wreq::BodyError, Wreq::ConnectionResetError, Wreq::RequestError] # if streaming fails while reading the response body - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example Save response to file # File.open("output.bin", "wb") do |f| # response.chunks { |chunk| f.write(chunk) } @@ -172,7 +185,7 @@ def chunks # Close the response and free associated resources. # # @return [void] - # @raise [Wreq::ForkError] if the child inherited wreq-ruby from its parent + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example # response.close def close @@ -186,6 +199,7 @@ def close # # @return [Wreq::TlsInfo, nil] TLS information for this response, or +nil+ # when unavailable + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example # client = Wreq::Client.new(tls_info: true) # response = client.get("https://example.com") @@ -208,6 +222,7 @@ class Response # Returns the response body as a string. # # @return [String] Response body text + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example # puts response.to_s # puts response @@ -221,6 +236,7 @@ def to_s # Format: # # # @return [String] Compact formatted response information + # @raise [Wreq::ForkError] if the response belongs to the parent process # @example # p response # # => # diff --git a/src/arch.rs b/src/arch.rs index afbbc81..a7e4f37 100644 --- a/src/arch.rs +++ b/src/arch.rs @@ -8,35 +8,63 @@ use std::mem::ManuallyDrop; -/// Native state that belongs to the process where the extension was loaded. +use magnus::Ruby; + +#[cfg(unix)] +use crate::error::fork_error; + +/// Native state that belongs to the process where it was created. /// /// A forked child must not destroy inherited clients, channels, or response /// bodies because their synchronization state may belong to threads that no /// longer exist. The child intentionally leaks the value and lets the operating /// system reclaim it when the process exits. /// -/// This wrapper only controls destruction. Call [`crate::rt::ensure_current`] -/// before using process-bound state stored inside it. -#[derive(Clone)] -pub(crate) struct ProcessLocal(ManuallyDrop); +/// `Send` and `Sync` only describe access between threads in one process. They +/// do not make a runtime, lock, channel, or connection pool safe after `fork`. +/// +/// [`ProcessLocal::get`] is the only access path and checks the object's own +/// process generation before exposing its value. +pub(crate) struct ProcessLocal { + value: ManuallyDrop, + #[cfg(unix)] + owner: unix::ProcessToken, +} impl ProcessLocal { /// Wrap native state created by the current process. pub(crate) fn new(value: T) -> Self { - Self(ManuallyDrop::new(value)) + Self { + value: ManuallyDrop::new(value), + #[cfg(unix)] + owner: unix::ProcessToken::current(), + } } -} -impl AsRef for ProcessLocal { - fn as_ref(&self) -> &T { - &self.0 + /// Borrow native state only from the process that created it. + /// + /// # Errors + /// + /// Returns `Wreq::ForkError` when the value was inherited from a parent + /// process. + #[inline] + pub(crate) fn get(&self, ruby: &Ruby) -> Result<&T, magnus::Error> { + #[cfg(unix)] + if let Some((owner_pid, current_pid)) = self.owner.forked_process_ids() { + return Err(fork_error(ruby, owner_pid, current_pid)); + } + + #[cfg(not(unix))] + let _ = ruby; + + Ok(&self.value) } } impl Drop for ProcessLocal { fn drop(&mut self) { #[cfg(unix)] - if forked_process_ids().is_some() { + if self.owner.forked_process_ids().is_some() { return; } @@ -44,7 +72,7 @@ impl Drop for ProcessLocal { // prevents an automatic second drop, and this wrapper's `Drop` // implementation runs at most once. unsafe { - ManuallyDrop::drop(&mut self.0); + ManuallyDrop::drop(&mut self.value); } } } @@ -74,54 +102,72 @@ pub(crate) const SUPPORTS_INTERFACE: bool = cfg!(any( mod unix { use std::{io, process, sync::OnceLock}; - /// Process state captured when the extension initializes. + /// Identity of the process generation that created native state. /// - /// The atfork guard uses a POSIX child handler to advance an atomic fork - /// generation without running Ruby code. + /// Forkguard's child callback only advances an atomic generation counter. + /// The PID is retained for diagnostics and as a fallback if registering + /// the callback fails. /// https://pubs.opengroup.org/onlinepubs/9799919799/functions/pthread_atfork.html - struct ForkGuard { - detector: forkguard::Guard, + pub(super) struct ProcessToken { + detector: Option, owner_pid: u32, } - impl ForkGuard { - /// Create a guard and register fork detection with the process. - fn new() -> io::Result { + impl ProcessToken { + /// Capture the current process and fork generation. + pub(super) fn current() -> Self { + Self::try_current().unwrap_or_else(|_| Self { + detector: None, + owner_pid: process::id(), + }) + } + + /// Capture the current process after registering fork detection. + fn try_current() -> io::Result { forkguard::Guard::try_new() .map(|detector| Self { - detector, + detector: Some(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())) + /// Return process IDs when this token was inherited through a fork. + pub(super) fn forked_process_ids(&self) -> Option<(u32, u32)> { + if let Some(detector) = &self.detector { + // Keep the stored generation unchanged so repeated accesses + // continue to reject the same inherited object. Cloning the + // detector copies one usize. + return detector + .clone() + .detected_fork() + .then(|| (self.owner_pid, process::id())); + } + + let current_pid = process::id(); + (self.owner_pid != current_pid).then_some((self.owner_pid, current_pid)) } } - static FORK_GUARD: OnceLock = OnceLock::new(); + /// Runtime owner captured when Tokio first initializes. + static RUNTIME_OWNER: OnceLock = OnceLock::new(); - /// Register process fork tracking before the extension exposes its API. + /// Register process fork tracking before the Tokio runtime is initialized. pub(crate) fn initialize_fork_tracking() -> io::Result<()> { - if FORK_GUARD.get().is_some() { + if RUNTIME_OWNER.get().is_some() { return Ok(()); } - let guard = ForkGuard::new()?; - let _ = FORK_GUARD.set(guard); + let owner = ProcessToken::try_current()?; + let _ = RUNTIME_OWNER.set(owner); Ok(()) } - /// Return process IDs only when this process inherited the extension. + /// Return process IDs when this process inherited an initialized runtime. pub(crate) fn forked_process_ids() -> Option<(u32, u32)> { - FORK_GUARD.get().and_then(ForkGuard::forked_process_ids) + RUNTIME_OWNER + .get() + .and_then(ProcessToken::forked_process_ids) } } @@ -175,7 +221,7 @@ mod tests { { let value = ProcessLocal::new(DropCounter(&drops)); - assert_eq!(value.as_ref().0.get(), 0); + assert_eq!(value.value.0.get(), 0); } assert_eq!(drops.get(), 1); diff --git a/src/client.rs b/src/client.rs index 9a2b5c4..e42b488 100644 --- a/src/client.rs +++ b/src/client.rs @@ -21,7 +21,6 @@ use crate::{ header::{Headers, OrigHeaders, UserAgent}, http::Method, options::{NativeOption, Options}, - rt, }; /// A builder for `Client`. @@ -51,7 +50,7 @@ struct Builder { cookie_store: Option, /// Whether to use cookie store provider. #[serde(default)] - cookie_provider: NativeOption, + cookie_provider: NativeOption>, // ========= Timeout options ========= /// The timeout to use for the client. (in seconds) @@ -121,7 +120,6 @@ struct Builder { zstd: Option, } -#[derive(Clone)] #[magnus::wrap(class = "Wreq::Client", free_immediately, size)] pub struct Client(ProcessLocal); @@ -171,12 +169,7 @@ impl Builder { extract_native_option!(options, builder, user_agent); extract_native_option!(options, builder, headers); extract_native_option!(options, builder, orig_headers); - extract_native_option!( - options, - builder, - cookie_provider, - Obj => |value| (*value).clone() - ); + extract_native_option!(options, builder, cookie_provider); builder .proxy .set(Extractor::::try_convert(options.as_value())?.into_inner()); @@ -193,11 +186,9 @@ impl Client { /// # Errors /// /// Returns Ruby configuration errors from [`Builder::from_options`] or the - /// native fallible client builder. Extra positional arguments return - /// `ArgumentError`. + /// native fallible client builder. An inherited cookie provider returns + /// `Wreq::ForkError`, and extra positional arguments return `ArgumentError`. pub fn new(ruby: &Ruby, args: &[Value]) -> Result { - rt::ensure_current(ruby)?; - Options::from_args(ruby, args, "client")? .map(Builder::from_options) .transpose() @@ -221,12 +212,15 @@ impl Client { /// /// # Errors /// - /// 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. + /// Returns `Wreq::ForkError` if the cookie provider belongs to a parent + /// process. Native build failures are mapped only after the GVL has been + /// reacquired. fn build(ruby: &Ruby, mut params: Builder) -> Result { - rt::ensure_current(ruby)?; - + let mut cookie_provider = params + .cookie_provider + .take() + .map(|jar| jar.clone_store(ruby)) + .transpose()?; let result = gvl::nogvl(|| { let mut builder = wreq::Client::builder(); @@ -270,12 +264,7 @@ impl Client { // Cookie options. apply_option!(set_if_some, builder, params.cookie_store, cookie_store); - apply_option!( - set_if_some_inner, - builder, - params.cookie_provider, - cookie_provider - ); + apply_option!(set_if_some, builder, cookie_provider, cookie_provider); // TCP options. apply_option!( @@ -393,9 +382,14 @@ impl Client { result.map_err(|err| wreq_error(ruby, err)) } - /// Clone the native client handle for a request future. - fn native_client(&self) -> wreq::Client { - self.0.as_ref().clone() + /// 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 { + self.0.get(ruby).cloned() } } @@ -431,63 +425,111 @@ impl Client { #[inline] pub fn request(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((method, url), request) = extract_request!(ruby, args, (Obj, String)); - execute_request(ruby, rb_self.native_client(), *method, url, request) + execute_request(ruby, rb_self.native_client(ruby)?, *method, url, request) } /// Send a GET request. #[inline] pub fn get(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request(ruby, rb_self.native_client(), Method::GET, url, request) + execute_request( + ruby, + rb_self.native_client(ruby)?, + Method::GET, + url, + request, + ) } /// Send a POST request. #[inline] pub fn post(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request(ruby, rb_self.native_client(), Method::POST, url, request) + execute_request( + ruby, + rb_self.native_client(ruby)?, + Method::POST, + url, + request, + ) } /// Send a PUT request. #[inline] pub fn put(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request(ruby, rb_self.native_client(), Method::PUT, url, request) + execute_request( + ruby, + rb_self.native_client(ruby)?, + Method::PUT, + url, + request, + ) } /// Send a DELETE request. #[inline] pub fn delete(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request(ruby, rb_self.native_client(), Method::DELETE, url, request) + execute_request( + ruby, + rb_self.native_client(ruby)?, + Method::DELETE, + url, + request, + ) } /// Send a HEAD request. #[inline] pub fn head(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request(ruby, rb_self.native_client(), Method::HEAD, url, request) + execute_request( + ruby, + rb_self.native_client(ruby)?, + Method::HEAD, + url, + request, + ) } /// Send an OPTIONS request. #[inline] pub fn options(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request(ruby, rb_self.native_client(), Method::OPTIONS, url, request) + execute_request( + ruby, + rb_self.native_client(ruby)?, + Method::OPTIONS, + url, + request, + ) } /// Send a TRACE request. #[inline] pub fn trace(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request(ruby, rb_self.native_client(), Method::TRACE, url, request) + execute_request( + ruby, + rb_self.native_client(ruby)?, + Method::TRACE, + url, + request, + ) } /// Send a PATCH request. #[inline] pub fn patch(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { let ((url,), request) = extract_request!(ruby, args, (String,)); - execute_request(ruby, rb_self.native_client(), Method::PATCH, url, request) + execute_request( + ruby, + rb_self.native_client(ruby)?, + Method::PATCH, + url, + request, + ) } } diff --git a/src/client/body/stream.rs b/src/client/body/stream.rs index d3d2807..0f15e83 100644 --- a/src/client/body/stream.rs +++ b/src/client/body/stream.rs @@ -65,17 +65,14 @@ impl BodyReceiver { /// Read the next body chunk, converting stream errors into Ruby errors. pub fn next(&self, ruby: &Ruby) -> Result, Error> { - rt::try_block_on( - ruby, - async { - match self.0.lock().await.as_mut().next().await { - Some(Ok(data)) => Ok(Some(data)), - Some(Err(err)) => Err(err), - None => Ok(None), - } - }, - wreq_error, - ) + rt::block_on(ruby, async { + match self.0.lock().await.as_mut().next().await { + Some(Ok(data)) => Ok(Some(data)), + Some(Err(err)) => Err(err), + None => Ok(None), + } + })? + .map_err(|err| wreq_error(ruby, err)) } } @@ -90,10 +87,8 @@ impl BodySender { /// # Errors /// /// Returns `TypeError` for a non-Integer capacity and `ArgumentError` for - /// an invalid range or argument count. Returns `Wreq::ForkError` before - /// creating a channel in a child that inherited the extension. + /// an invalid range or argument count. pub fn new(ruby: &Ruby, args: &[Value]) -> Result { - rt::ensure_current(ruby)?; let capacity = parse_capacity(ruby, args)?; // Create the Tokio channel without allowing an unwind to cross the Ruby FFI boundary. @@ -121,8 +116,6 @@ impl BodySender { /// wait raises `Wreq::InterruptError`. Returns `Wreq::ForkError` before /// reading an inherited channel. pub fn push(ruby: &Ruby, rb_self: &Self, data: RString) -> Result<(), Error> { - rt::ensure_current(ruby)?; - // Clone during the shared borrow, then release it before waiting // for capacity. Request attachment needs a mutable borrow. let tx = match &rb_self.read_inner(ruby)?.tx { @@ -130,7 +123,8 @@ impl BodySender { _ => return Err(closed_body_sender_error(ruby)), }; - rt::try_block_on(ruby, tx.send(data.to_bytes()), body_sender_send_error) + rt::block_on(ruby, tx.send(data.to_bytes()))? + .map_err(|err| body_sender_send_error(ruby, err)) } /// Close the producing side while retaining the receiver and queued chunks. @@ -142,7 +136,6 @@ impl BodySender { /// Returns `Wreq::ForkError` before reading an inherited channel, or /// `Wreq::BodyError` if the internal state is already borrowed. pub fn close(ruby: &Ruby, rb_self: &Self) -> Result<(), Error> { - rt::ensure_current(ruby)?; let mut inner = rb_self.write_inner(ruby)?; inner.tx.take(); Ok(()) @@ -155,22 +148,21 @@ impl BodySender { /// Returns `Wreq::ForkError` before reading an inherited channel, or /// `Wreq::BodyError` if the internal state is already borrowed. pub fn is_closed(ruby: &Ruby, rb_self: &Self) -> Result { - rt::ensure_current(ruby)?; rb_self.read_inner(ruby).map(|r| r.is_closed()) } - /// Borrow the channel state without panicking on accidental re-entry. + /// Borrow channel state only in the process that created this sender. fn read_inner(&self, ruby: &Ruby) -> Result, Error> { self.0 - .as_ref() + .get(ruby)? .try_borrow() .map_err(|err| body_sender_borrow_error(ruby, err)) } - /// Mutably borrow the channel state without panicking on accidental re-entry. + /// Mutably borrow channel state only in the process that created this sender. fn write_inner(&self, ruby: &Ruby) -> Result, Error> { self.0 - .as_ref() + .get(ruby)? .try_borrow_mut() .map_err(|err| body_sender_borrow_mut_error(ruby, err)) } @@ -182,7 +174,6 @@ impl BodySender { /// Returns `Wreq::MemoryError` if the receiver was already consumed, or /// `Wreq::BodyError` if Ruby re-enters while the state is borrowed. pub(super) fn take_receiver(&self, ruby: &Ruby) -> Result, Error> { - rt::ensure_current(ruby)?; self.write_inner(ruby)? .rx .take() diff --git a/src/client/req.rs b/src/client/req.rs index 0603d9e..64ce764 100644 --- a/src/client/req.rs +++ b/src/client/req.rs @@ -184,131 +184,128 @@ pub fn execute_request>( url: U, mut request: Request, ) -> Result { - rt::try_block_on( - ruby, - async move { - let mut builder = client.request(method.into_ffi(), url.as_ref()); - - // Emulation options. - apply_option!(set_if_some_inner, builder, request.emulation, emulation); - - // Version options. - apply_option!( - set_if_some_map, - builder, - request.version, - version, - Version::into_ffi - ); - - // Timeout options. - apply_option!( - set_if_some_map, - builder, - request.timeout, - timeout, - Duration::from_secs - ); - apply_option!( - set_if_some_map, - builder, - request.read_timeout, - read_timeout, - Duration::from_secs - ); - - // Network options. - apply_option!(set_if_some, builder, request.proxy, proxy); - apply_option!(set_if_some, builder, request.local_address, local_address); - #[cfg(any( - target_os = "android", - target_os = "fuchsia", - target_os = "illumos", - target_os = "ios", - target_os = "linux", - target_os = "macos", - target_os = "solaris", - target_os = "tvos", - target_os = "visionos", - target_os = "watchos", - ))] - apply_option!(set_if_some, builder, request.interface, interface); - - // Headers options. - apply_option!(set_if_some_into_inner, builder, request.headers, headers); - apply_option!( - set_if_some_inner, - builder, - request.orig_headers, - orig_headers - ); - apply_option!( - set_if_some, - builder, - request.default_headers, - default_headers - ); - - // Cookies options. - if let Some(cookies) = request.cookies.take() { - for cookie in cookies.0 { - builder = builder.header(header::COOKIE, cookie); - } - } + rt::block_on(ruby, async move { + let mut builder = client.request(method.into_ffi(), url.as_ref()); + + // Emulation options. + apply_option!(set_if_some_inner, builder, request.emulation, emulation); + + // Version options. + apply_option!( + set_if_some_map, + builder, + request.version, + version, + Version::into_ffi + ); + + // Timeout options. + apply_option!( + set_if_some_map, + builder, + request.timeout, + timeout, + Duration::from_secs + ); + apply_option!( + set_if_some_map, + builder, + request.read_timeout, + read_timeout, + Duration::from_secs + ); - // Authentication options. - apply_option!( - set_if_some_map_ref, - builder, - request.auth, - auth, - AsRef::::as_ref - ); - apply_option!(set_if_some, builder, request.bearer_auth, bearer_auth); - if let Some(basic_auth) = request.basic_auth.take() { - builder = builder.basic_auth(basic_auth.0, basic_auth.1); + // Network options. + apply_option!(set_if_some, builder, request.proxy, proxy); + apply_option!(set_if_some, builder, request.local_address, local_address); + #[cfg(any( + target_os = "android", + target_os = "fuchsia", + target_os = "illumos", + target_os = "ios", + target_os = "linux", + target_os = "macos", + target_os = "solaris", + target_os = "tvos", + target_os = "visionos", + target_os = "watchos", + ))] + apply_option!(set_if_some, builder, request.interface, interface); + + // Headers options. + apply_option!(set_if_some_into_inner, builder, request.headers, headers); + apply_option!( + set_if_some_inner, + builder, + request.orig_headers, + orig_headers + ); + apply_option!( + set_if_some, + builder, + request.default_headers, + default_headers + ); + + // Cookies options. + if let Some(cookies) = request.cookies.take() { + for cookie in cookies.0 { + builder = builder.header(header::COOKIE, cookie); } + } - // Allow redirects options. - match request.allow_redirects { - Some(false) => { - builder = builder.redirect(wreq::redirect::Policy::none()); - } - Some(true) => { - builder = builder.redirect( - request - .max_redirects - .take() - .map(wreq::redirect::Policy::limited) - .unwrap_or_default(), - ); - } - None => {} - }; - - // Compression options. - apply_option!(set_if_some, builder, request.gzip, gzip); - apply_option!(set_if_some, builder, request.brotli, brotli); - apply_option!(set_if_some, builder, request.deflate, deflate); - apply_option!(set_if_some, builder, request.zstd, zstd); - - // Query options. - apply_option!(set_if_some_ref, builder, request.query, query); - - // Form options. - apply_option!(set_if_some_ref, builder, request.form, form); - - // JSON options. - apply_option!(set_if_some_ref, builder, request.json, json); - - // Body options. - if let Some(body) = request.body.take() { - builder = builder.body(wreq::Body::from(body)); + // Authentication options. + apply_option!( + set_if_some_map_ref, + builder, + request.auth, + auth, + AsRef::::as_ref + ); + apply_option!(set_if_some, builder, request.bearer_auth, bearer_auth); + if let Some(basic_auth) = request.basic_auth.take() { + builder = builder.basic_auth(basic_auth.0, basic_auth.1); + } + + // Allow redirects options. + match request.allow_redirects { + Some(false) => { + builder = builder.redirect(wreq::redirect::Policy::none()); + } + Some(true) => { + builder = builder.redirect( + request + .max_redirects + .take() + .map(wreq::redirect::Policy::limited) + .unwrap_or_default(), + ); } + None => {} + }; + + // Compression options. + apply_option!(set_if_some, builder, request.gzip, gzip); + apply_option!(set_if_some, builder, request.brotli, brotli); + apply_option!(set_if_some, builder, request.deflate, deflate); + apply_option!(set_if_some, builder, request.zstd, zstd); + + // Query options. + apply_option!(set_if_some_ref, builder, request.query, query); + + // Form options. + apply_option!(set_if_some_ref, builder, request.form, form); + + // JSON options. + apply_option!(set_if_some_ref, builder, request.json, json); + + // Body options. + if let Some(body) = request.body.take() { + builder = builder.body(wreq::Body::from(body)); + } - // Send request. - builder.send().await.map(Response::new) - }, - wreq_error, - ) + // Send request. + builder.send().await.map(Response::new) + })? + .map_err(|err| wreq_error(ruby, err)) } diff --git a/src/client/resp.rs b/src/client/resp.rs index a2f25d4..6dff164 100644 --- a/src/client/resp.rs +++ b/src/client/resp.rs @@ -22,7 +22,10 @@ use crate::{ /// A response from a request. #[magnus::wrap(class = "Wreq::Response", free_immediately, size)] -pub struct Response { +pub struct Response(ProcessLocal); + +/// Inner response state owned by the process that received it. +struct ResponseInner { uri: Uri, version: Version, status: StatusCode, @@ -30,7 +33,8 @@ pub struct Response { headers: HeaderMap, local_addr: Option, remote_addr: Option, - state: ProcessLocal, + body: ArcSwapOption, + extensions: Extensions, } /// Represents the state of the HTTP response body. @@ -41,12 +45,6 @@ enum Body { Reusable(Bytes), } -/// Response state that may contain handles owned by the native runtime. -struct NativeResponseState { - body: ArcSwapOption, - extensions: Extensions, -} - impl Response { /// Create a new [`Response`] instance. pub fn new(response: wreq::Response) -> Self { @@ -57,7 +55,7 @@ impl Response { let response = HttpResponse::from(response); let (parts, body) = response.into_parts(); - Response { + Response(ProcessLocal::new(ResponseInner { uri, local_addr, remote_addr, @@ -65,23 +63,20 @@ impl Response { version: Version::from_ffi(parts.version), status: StatusCode::from(parts.status), headers: parts.headers, - state: ProcessLocal::new(NativeResponseState { - body: ArcSwapOption::from_pointee(Body::Streamable(body)), - extensions: parts.extensions, - }), - } + body: ArcSwapOption::from_pointee(Body::Streamable(body)), + extensions: parts.extensions, + })) } /// Internal method to get the wreq::Response, optionally streaming the body. fn response(&self, ruby: &Ruby, stream: bool) -> Result { - rt::ensure_current(ruby)?; - let state = self.state.as_ref(); + let state = self.0.get(ruby)?; let build_response = |body: wreq::Body| -> wreq::Response { let mut response = HttpResponse::new(body); - *response.version_mut() = self.version.into_ffi(); - *response.status_mut() = self.status.0; - *response.headers_mut() = self.headers.clone(); + *response.version_mut() = state.version.into_ffi(); + *response.status_mut() = state.status.0; + *response.headers_mut() = state.headers.clone(); *response.extensions_mut() = state.extensions.clone(); wreq::Response::from(response) }; @@ -92,11 +87,11 @@ impl Response { return if stream { Ok(build_response(body)) } else { - let bytes = rt::try_block_on( + let bytes = rt::block_on( ruby, BodyExt::collect(body).map_ok(|buf| buf.to_bytes()), - wreq_error, - )?; + )? + .map_err(|err| wreq_error(ruby, err))?; state .body @@ -124,38 +119,66 @@ impl Response { impl Response { /// Get the response status code as a u16. + /// + /// # Errors + /// + /// Returns `Wreq::ForkError` when the response belongs to a parent process. #[inline] - pub fn code(&self) -> u16 { - self.status.0.as_u16() + pub fn code(ruby: &Ruby, rb_self: &Self) -> Result { + rb_self + .0 + .get(ruby) + .map(|response| response.status.0.as_u16()) } /// Get the response status code. + /// + /// # Errors + /// + /// Returns `Wreq::ForkError` when the response belongs to a parent process. #[inline] - pub fn status(&self) -> StatusCode { - self.status + pub fn status(ruby: &Ruby, rb_self: &Self) -> Result { + rb_self.0.get(ruby).map(|response| response.status) } /// Get the response HTTP version. + /// + /// # Errors + /// + /// Returns `Wreq::ForkError` when the response belongs to a parent process. #[inline] - pub fn version(&self) -> Version { - self.version + pub fn version(ruby: &Ruby, rb_self: &Self) -> Result { + rb_self.0.get(ruby).map(|response| response.version) } /// Get the response URL. + /// + /// # Errors + /// + /// Returns `Wreq::ForkError` when the response belongs to a parent process. #[inline] - pub fn url(&self) -> String { - self.uri.to_string() + pub fn url(ruby: &Ruby, rb_self: &Self) -> Result { + rb_self.0.get(ruby).map(|response| response.uri.to_string()) } /// Get the content length of the response, if known. + /// + /// # Errors + /// + /// Returns `Wreq::ForkError` when the response belongs to a parent process. #[inline] - pub fn content_length(&self) -> Option { - self.content_length + pub fn content_length(ruby: &Ruby, rb_self: &Self) -> Result, Error> { + rb_self.0.get(ruby).map(|response| response.content_length) } /// Get the response cookies. + /// + /// # Errors + /// + /// Returns `Wreq::ForkError` when the response belongs to a parent process. pub fn cookies(ruby: &Ruby, rb_self: &Self) -> Result { - let cookies = Cookie::extract_headers_cookies(&rb_self.headers); + let response = rb_self.0.get(ruby)?; + let cookies = Cookie::extract_headers_cookies(&response.headers); let ary = ruby.ary_new_capa(cookies.len()); for cookie in cookies { ary.push(cookie)?; @@ -164,63 +187,86 @@ impl Response { } /// Get the response headers. + /// + /// # Errors + /// + /// Returns `Wreq::ForkError` when the response belongs to a parent process. #[inline] - pub fn headers(&self) -> Headers { - Headers::from(self.headers.clone()) + pub fn headers(ruby: &Ruby, rb_self: &Self) -> Result { + rb_self + .0 + .get(ruby) + .map(|response| Headers::from(response.headers.clone())) } /// Get the local socket address, if available. + /// + /// # Errors + /// + /// Returns `Wreq::ForkError` when the response belongs to a parent process. #[inline] - pub fn local_addr(&self) -> Option { - self.local_addr.map(|addr| addr.to_string()) + pub fn local_addr(ruby: &Ruby, rb_self: &Self) -> Result, Error> { + rb_self + .0 + .get(ruby) + .map(|response| response.local_addr.map(|addr| addr.to_string())) } /// Get the remote socket address, if available. + /// + /// # Errors + /// + /// Returns `Wreq::ForkError` when the response belongs to a parent process. #[inline] - pub fn remote_addr(&self) -> Option { - self.remote_addr.map(|addr| addr.to_string()) + pub fn remote_addr(ruby: &Ruby, rb_self: &Self) -> Result, Error> { + rb_self + .0 + .get(ruby) + .map(|response| response.remote_addr.map(|addr| addr.to_string())) } /// Return peer certificate data retained for this response. - fn tls_info(&self) -> Option { - self.state - .as_ref() + /// + /// # Errors + /// + /// Returns `Wreq::ForkError` when the response belongs to a parent process. + fn tls_info(ruby: &Ruby, rb_self: &Self) -> Result, Error> { + Ok(rb_self + .0 + .get(ruby)? .extensions .get::() .cloned() - .map(TlsInfo) + .map(TlsInfo)) } /// Get the response body as bytes. pub fn bytes(ruby: &Ruby, rb_self: &Self) -> Result { let response = rb_self.response(ruby, false)?; - rt::try_block_on(ruby, response.bytes(), wreq_error) + rt::block_on(ruby, response.bytes())?.map_err(|err| wreq_error(ruby, err)) } /// Get the full response text given a specific encoding. pub fn text(ruby: &Ruby, rb_self: &Self, args: &[Value]) -> Result { - rt::ensure_current(ruby)?; let args = scan_args::<(), (Option,), (), (), (), ()>(args)?; let response = rb_self.response(ruby, false)?; match args.optional.0 { - Some(encoding) => { - rt::try_block_on(ruby, response.text_with_charset(encoding), wreq_error) - } - None => rt::try_block_on(ruby, response.text(), wreq_error), + Some(encoding) => rt::block_on(ruby, response.text_with_charset(encoding))? + .map_err(|err| wreq_error(ruby, err)), + None => rt::block_on(ruby, response.text())?.map_err(|err| wreq_error(ruby, err)), } } /// Get the response body as JSON. pub fn json(ruby: &Ruby, rb_self: &Self) -> Result { let response = rb_self.response(ruby, false)?; - let json = rt::try_block_on(ruby, response.json::(), wreq_error)?; + let json = + rt::block_on(ruby, response.json::())?.map_err(|err| wreq_error(ruby, err))?; crate::serde::serialize(ruby, &json) } /// Yield response body chunks to the given Ruby block. pub fn chunks(ruby: &Ruby, rb_self: &Self) -> Result<(), Error> { - rt::ensure_current(ruby)?; - if !ruby.block_given() { return Err(no_block_given_error(ruby)); } @@ -241,11 +287,11 @@ impl Response { /// /// # Errors /// - /// Returns `Wreq::ForkError` before touching a body inherited from the + /// Returns `Wreq::ForkError` before touching a response inherited from the /// parent process. pub fn close(ruby: &Ruby, rb_self: &Self) -> Result<(), Error> { - rt::ensure_current(ruby)?; - gvl::nogvl(|| rb_self.state.as_ref().body.swap(None)); + let response = rb_self.0.get(ruby)?; + gvl::nogvl(|| response.body.swap(None)); Ok(()) } } diff --git a/src/cookie.rs b/src/cookie.rs index fdb16f6..a1029d8 100644 --- a/src/cookie.rs +++ b/src/cookie.rs @@ -18,6 +18,7 @@ use magnus::{ use wreq::header::{self, HeaderMap, HeaderValue}; use crate::{ + arch::ProcessLocal, error::{header_value_error, type_error}, gvl, options::{NativeOption, Options}, @@ -78,9 +79,14 @@ struct Builder { /// A cookie jar that can be shared with a Ruby `Wreq::Client`. /// /// Pass a populated jar as the client's `cookie_provider` option. -#[derive(Clone, Default)] #[magnus::wrap(class = "Wreq::Jar", free_immediately, size)] -pub struct Jar(pub Arc); +pub struct Jar(ProcessLocal>); + +impl Default for Jar { + fn default() -> Self { + Self(ProcessLocal::new(Arc::new(wreq::cookie::Jar::default()))) + } +} // ===== impl Builder ===== @@ -309,13 +315,18 @@ impl TryConvert for Cookies { impl Jar { /// Create a new [`Jar`] with an empty cookie store. pub fn new() -> Self { - Self(Arc::new(wreq::cookie::Jar::default())) + Self::default() } /// Get all cookies. + /// + /// # Errors + /// + /// Returns `Wreq::ForkError` when the jar belongs to a parent process. pub fn get_all(ruby: &Ruby, rb_self: &Self) -> Result { let cookies: Vec = rb_self .0 + .get(ruby)? .get_all() .map(RawCookie::from) .map(Cookie) @@ -331,28 +342,53 @@ impl Jar { /// /// # Errors /// - /// Returns `TypeError` when `cookie` is neither a [`Cookie`] nor a String. + /// Returns `Wreq::ForkError` when the jar belongs to a parent process, or + /// `TypeError` when `cookie` is neither a [`Cookie`] nor a String. pub fn add(&self, cookie: Value, url: String) -> Result<(), Error> { + let ruby = Ruby::get_with(cookie); + let jar = self.0.get(&ruby)?; + if let Ok(cookie) = Obj::::try_convert(cookie) { - gvl::nogvl(|| self.0.add(cookie.clone_for_jar(), &url)); + gvl::nogvl(|| jar.add(cookie.clone_for_jar(), &url)); return Ok(()); } - let ruby = Ruby::get_with(cookie); let cookie = String::try_convert(cookie) .map_err(|_| type_error(&ruby, "cookie must be a Wreq::Cookie or String"))?; - gvl::nogvl(|| self.0.add(cookie.as_ref(), &url)); + gvl::nogvl(|| jar.add(cookie.as_ref(), &url)); Ok(()) } /// Remove a cookie from this jar by name and URL. - pub fn remove(&self, name: String, url: String) { - gvl::nogvl(|| self.0.remove(name, &url)) + /// + /// # Errors + /// + /// Returns `Wreq::ForkError` when the jar belongs to a parent process. + pub fn remove(ruby: &Ruby, rb_self: &Self, name: String, url: String) -> Result<(), Error> { + let jar = rb_self.0.get(ruby)?; + gvl::nogvl(|| jar.remove(name, &url)); + Ok(()) } /// Clear all cookies in this jar. - pub fn clear(&self) { - gvl::nogvl(|| self.0.clear()) + /// + /// # Errors + /// + /// Returns `Wreq::ForkError` when the jar belongs to a parent process. + pub fn clear(ruby: &Ruby, rb_self: &Self) -> Result<(), Error> { + let jar = rb_self.0.get(ruby)?; + gvl::nogvl(|| jar.clear()); + Ok(()) + } + + /// Clone the shared native store in the process that created this jar. + /// + /// # Errors + /// + /// Returns `Wreq::ForkError` when the jar was inherited from a parent + /// process. + pub(crate) fn clone_store(&self, ruby: &Ruby) -> Result, Error> { + self.0.get(ruby).cloned() } } diff --git a/src/error.rs b/src/error.rs index 1620a40..57c1935 100644 --- a/src/error.rs +++ b/src/error.rs @@ -105,7 +105,7 @@ pub fn fork_error(ruby: &Ruby, owner_pid: u32, current_pid: u32) -> MagnusError MagnusError::new( ruby.get_inner(&FORK_ERROR), format!( - "wreq-ruby was loaded in process {owner_pid} and cannot be used after fork in process {current_pid}" + "wreq-ruby native state was created in process {owner_pid} and cannot be used after fork in process {current_pid}" ), ) } diff --git a/src/lib.rs b/src/lib.rs index 549a1f2..0503557 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -102,7 +102,5 @@ fn init(ruby: &Ruby) -> Result<(), Error> { tls::include(ruby, &gem_module)?; client::include(ruby, &gem_module)?; emulate::include(ruby, &gem_module)?; - #[cfg(unix)] - rt::initialize(ruby)?; Ok(()) } diff --git a/src/macros.rs b/src/macros.rs index 412318b..18d97c6 100644 --- a/src/macros.rs +++ b/src/macros.rs @@ -149,7 +149,6 @@ macro_rules! define_ruby_enum { macro_rules! extract_request { ($ruby:expr, $args:expr, $required:ty) => {{ - crate::rt::ensure_current($ruby)?; let args = magnus::scan_args::scan_args::<$required, (), (), (), magnus::RHash, ()>($args)?; let required = args.required; let request = crate::client::req::Request::new($ruby, args.keywords)?; diff --git a/src/rt.rs b/src/rt.rs index a385f37..3cb8555 100644 --- a/src/rt.rs +++ b/src/rt.rs @@ -17,29 +17,13 @@ use crate::{ /// Initialize the global runtime lazily and preserve failures for Ruby. static RUNTIME: OnceLock> = OnceLock::new(); -enum BlockOnError { - Interrupted, - Future(E), -} - -/// Register fork tracking while the native extension is being loaded. +/// Reject a child process that inherited an initialized native runtime. /// /// # 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> { +/// Returns `Wreq::ForkError` when the global runtime belongs to the parent +/// process. +fn ensure_runtime_owner(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)); @@ -54,21 +38,25 @@ pub fn ensure_current(ruby: &Ruby) -> Result<(), magnus::Error> { /// 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. +/// Ruby exceptions. Its output is returned unchanged. If that output is a +/// `Result`, convert its error after this function returns and reacquires the +/// GVL. /// /// # Errors /// -/// Returns `Wreq::ForkError` if the extension belongs to a parent process, +/// Returns `Wreq::ForkError` if the runtime belongs to a parent process, /// `Wreq::BuilderError` if the Tokio runtime cannot be initialized, -/// `Wreq::InterruptError` if Ruby interrupts the request, or the error produced -/// by `map_err` if the future fails. -pub fn try_block_on(ruby: &Ruby, future: F, map_err: M) -> Result +/// or `Wreq::InterruptError` if Ruby interrupts the operation. +pub(crate) fn block_on(ruby: &Ruby, future: F) -> Result where - F: Future>, - M: FnOnce(&Ruby, E) -> magnus::Error, + F: Future, { - ensure_current(ruby)?; + // Install fork tracking at the same point as the lazy runtime. Loading the + // extension alone must not claim the runtime for the parent process. + #[cfg(unix)] + arch::initialize_fork_tracking().map_err(|err| fork_handler_error(ruby, &err))?; + + ensure_runtime_owner(ruby)?; let runtime = RUNTIME .get_or_init(|| { let mut builder = Builder::new_multi_thread(); @@ -80,15 +68,11 @@ where runtime.block_on(async move { tokio::select! { biased; - _ = flag.cancelled() => Err(BlockOnError::Interrupted), - result = future => result.map_err(BlockOnError::Future), + _ = flag.cancelled() => None, + result = future => Some(result), } }) }); - match result { - Ok(value) => Ok(value), - Err(BlockOnError::Interrupted) => Err(interrupt_error(ruby)), - Err(BlockOnError::Future(err)) => Err(map_err(ruby, err)), - } + result.ok_or_else(|| interrupt_error(ruby)) } diff --git a/test/fork_test.rb b/test/fork_test.rb index 7fc7b1d..7a772fb 100644 --- a/test/fork_test.rb +++ b/test/fork_test.rb @@ -7,15 +7,16 @@ class ForkTest < Minitest::Test FORK_ERROR_LABELS = %w[ - before_runtime - invalid_client - invalid_request - fresh_body_sender + module_request + fresh_client_request + fresh_body_sender_push inherited_body_sender_push inherited_body_sender_close inherited_body_sender_closed - fresh_client inherited_client + inherited_jar + inherited_cookie_provider + inherited_response_metadata inherited_response inherited_response_text inherited_response_chunks @@ -26,13 +27,36 @@ def test_fork_error_is_a_runtime_error assert_operator Wreq::ForkError, :<, RuntimeError end - def test_loaded_extension_is_rejected_after_fork + def test_loaded_extension_can_initialize_runtime_after_fork + skip "fork is not supported on this platform" unless Process.respond_to?(:fork) + + stdout, stderr, status = run_fork_script("prefork_runtime.rb") + + assert status.success?, "subprocess failed with #{status.inspect}: #{stderr}" + assert_equal "ok\n", stdout + assert_match(/loaded_only=ok/, stderr) + assert_match(/before_runtime=ok/, stderr) + assert_match(/parent_after_children=ok/, stderr) + %w[ + inherited_client_before_runtime + inherited_sender_before_runtime + inherited_jar_before_runtime + inherited_cookie_provider_before_runtime + ].each do |label| + assert_match(/#{label}=Wreq::ForkError:.*cannot be used after fork/, stderr) + end + refute_match(/\[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) stdout, stderr, status = run_fork_script("fork_safety.rb") assert status.success?, "subprocess failed with #{status.inspect}: #{stderr}" assert_equal "ok\n", stdout + assert_match(/non_runtime_construction=ok/, stderr) + assert_match(/inherited_snapshots=ok/, stderr) FORK_ERROR_LABELS.each do |label| assert_match(/#{label}=Wreq::ForkError:.*cannot be used after fork/, stderr) assert_match(/#{label}_retry=Wreq::ForkError:.*cannot be used after fork/, stderr) diff --git a/test/scripts/fork_safety.rb b/test/scripts/fork_safety.rb index 7593b40..311d56e 100644 --- a/test/scripts/fork_safety.rb +++ b/test/scripts/fork_safety.rb @@ -9,36 +9,22 @@ $stderr.sync = true def expect_fork_error(label) - child_pid = fork do - 2.times do |attempt| - attempt_label = attempt.zero? ? label : "#{label}_retry" - - begin - Timeout.timeout(5) { yield } - rescue Wreq::ForkError => error - warn "#{attempt_label}=#{error.class}: #{error.message}" - next - rescue => error - warn "#{attempt_label}=unexpected #{error.class}: #{error.message}" - exit! 2 - end + 2.times do |attempt| + attempt_label = attempt.zero? ? label : "#{label}_retry" - warn "#{attempt_label}=missing Wreq::ForkError" - exit! 3 + begin + yield + rescue Wreq::ForkError => error + warn "#{attempt_label}=#{error.class}: #{error.message}" + next + rescue => error + abort "#{attempt_label}=unexpected #{error.class}: #{error.message}" end - exit! 0 + abort "#{attempt_label}=missing Wreq::ForkError" end - - _, status = Process.wait2(child_pid) - abort "#{label} child failed with #{status.inspect}" unless status.success? end -expect_fork_error("before_runtime") { Wreq::Client.new } -expect_fork_error("invalid_client") { Wreq::Client.new(unknown: true) } -expect_fork_error("invalid_request") { Wreq.get(1) } -expect_fork_error("fresh_body_sender") { Wreq::BodySender.new(0) } - server = TCPServer.new("127.0.0.1", 0) port = server.addr[1] server_pid = fork do @@ -51,7 +37,7 @@ def expect_fork_error(label) while (line = socket.gets) break if line == "\r\n" end - socket.write("HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok") + socket.write("HTTP/1.1 200 OK\r\nX-Fork-Test: ok\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok") ensure socket.close end @@ -63,29 +49,66 @@ def expect_fork_error(label) server.close url = "http://127.0.0.1:#{port}/" + +# Start the server process before Tokio creates worker threads in the parent. +runtime_probe = Wreq::BodySender.new(1) +runtime_probe.push("warmup") + client = Wreq::Client.new abort "parent warm-up failed" unless client.get(url).bytes == "ok" -def build_inherited_objects(client, url) - [Wreq::Client.new, Wreq::BodySender.new, client.get(url)] +inherited_objects = { + client: Wreq::Client.new, + sender: Wreq::BodySender.new, + response: client.get(url), + jar: Wreq::Jar.new +} +inherited_weak_refs = inherited_objects.values.map { |object| WeakRef.new(object) } +status_snapshot = inherited_objects[:response].status +headers_snapshot = inherited_objects[:response].headers + +guard_pid = fork do + Timeout.timeout(10) do + jar = Wreq::Jar.new + jar.add("child=1; Path=/", url) + abort "fresh child jar failed" unless jar.get_all.one? + + Wreq::Client.new(cookie_provider: jar) + Wreq::BodySender.new + warn "non_runtime_construction=ok" + + abort "status snapshot changed" unless status_snapshot.to_i == 200 + abort "headers snapshot changed" unless headers_snapshot["X-Fork-Test"] == "ok" + warn "inherited_snapshots=ok" + + expect_fork_error("module_request") { Wreq.get(url) } + expect_fork_error("fresh_client_request") { Wreq::Client.new.get(url) } + expect_fork_error("fresh_body_sender_push") { Wreq::BodySender.new.push("chunk") } + expect_fork_error("inherited_body_sender_push") do + inherited_objects[:sender].push("chunk") + end + expect_fork_error("inherited_body_sender_close") { inherited_objects[:sender].close } + expect_fork_error("inherited_body_sender_closed") { inherited_objects[:sender].closed? } + expect_fork_error("inherited_client") { inherited_objects[:client].get(url) } + expect_fork_error("inherited_jar") { inherited_objects[:jar].get_all } + expect_fork_error("inherited_cookie_provider") do + Wreq::Client.new(cookie_provider: inherited_objects[:jar]) + end + expect_fork_error("inherited_response_metadata") { inherited_objects[:response].status } + expect_fork_error("inherited_response") { inherited_objects[:response].bytes } + expect_fork_error("inherited_response_text") { inherited_objects[:response].text } + expect_fork_error("inherited_response_chunks") { inherited_objects[:response].chunks { nil } } + expect_fork_error("inherited_response_close") { inherited_objects[:response].close } + end + exit! 0 +rescue => error + warn "guard_checks=unexpected #{error.class}: #{error.message}" + exit! 2 end +_, guard_status = Process.wait2(guard_pid) +abort "guard checks child failed with #{guard_status.inspect}" unless guard_status.success? -inherited_objects = build_inherited_objects(client, url) -inherited_weak_refs = inherited_objects.map { |object| WeakRef.new(object) } - -expect_fork_error("inherited_body_sender_push") do - inherited_objects[1].push("chunk") -end -expect_fork_error("inherited_body_sender_close") { inherited_objects[1].close } -expect_fork_error("inherited_body_sender_closed") { inherited_objects[1].closed? } -expect_fork_error("fresh_client") { Wreq::Client.new } -expect_fork_error("inherited_client") { client.get(url) } -expect_fork_error("inherited_response") { inherited_objects[2].bytes } -expect_fork_error("inherited_response_text") { inherited_objects[2].text(1) } -expect_fork_error("inherited_response_chunks") { inherited_objects[2].chunks } -expect_fork_error("inherited_response_close") { inherited_objects[2].close } - -# Release the earlier test blocks so this array is the only strong reference. +# The hash is now the only strong reference to the inherited native objects. GC.start(full_mark: true, immediate_sweep: true) gc_pid = fork do inherited_objects = nil diff --git a/test/scripts/prefork_runtime.rb b/test/scripts/prefork_runtime.rb new file mode 100644 index 0000000..8daa267 --- /dev/null +++ b/test/scripts/prefork_runtime.rb @@ -0,0 +1,95 @@ +# frozen_string_literal: true + +require "socket" +require "timeout" +require "wreq" + +$stdout.sync = true +$stderr.sync = true + +def expect_fork_error(label) + yield + abort "#{label}=missing Wreq::ForkError" +rescue Wreq::ForkError => error + warn "#{label}=#{error.class}: #{error.message}" +end + +def run_child(label) + child_pid = fork do + Timeout.timeout(10) { yield } + warn "#{label}=ok" + exit! 0 + rescue => error + warn "#{label}=unexpected #{error.class}: #{error.message}" + exit! 2 + end + + _, status = Process.wait2(child_pid) + abort "#{label} child failed with #{status.inspect}" unless status.success? +end + +server = TCPServer.new("127.0.0.1", 0) +url = "http://127.0.0.1:#{server.addr[1]}/" +server_pid = fork do + 3.times do + ready = IO.select([server], nil, nil, 10) + exit! 4 unless ready + + socket = server.accept + begin + while (line = socket.gets) + break if line == "\r\n" + end + socket.write("HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok") + ensure + socket.close + end + end + exit! 0 +ensure + server.close +end +server.close + +# Requiring the extension is the only parent-side Wreq operation before this fork. +run_child("loaded_only") do + abort "module request failed" unless Wreq.get(url).bytes == "ok" +end + +inherited_client = Wreq::Client.new +inherited_sender = Wreq::BodySender.new +inherited_jar = Wreq::Jar.new + +run_child("before_runtime") do + expect_fork_error("inherited_client_before_runtime") do + inherited_client.get(url) + end + expect_fork_error("inherited_sender_before_runtime") { inherited_sender.closed? } + expect_fork_error("inherited_jar_before_runtime") { inherited_jar.get_all } + expect_fork_error("inherited_cookie_provider_before_runtime") do + Wreq::Client.new(cookie_provider: inherited_jar) + end + + sender = Wreq::BodySender.new + sender.push("child") + + jar = Wreq::Jar.new + jar.add("child=1; Path=/", url) + abort "child jar failed" unless jar.get_all.one? + + client = Wreq::Client.new(cookie_provider: jar) + abort "client request failed" unless client.get(url).bytes == "ok" +end + +Timeout.timeout(10) do + abort "parent client failed" unless inherited_client.get(url).bytes == "ok" + inherited_sender.push("parent") + inherited_jar.add("parent=1; Path=/", url) + abort "parent jar failed" unless inherited_jar.get_all.one? +end +warn "parent_after_children=ok" + +_, server_status = Process.wait2(server_pid) +abort "server failed with #{server_status.inspect}" unless server_status.success? + +puts "ok"