From 18efcc80b84c56ab3069b22b41845a769e611abe Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Sun, 11 Oct 2026 13:43:02 -0700 Subject: [PATCH] Ruby: generate default client ID per process `River::Config` builds its default client ID (`host-pid-random`) when the config is constructed. Preforking servers such as Puma, Unicorn, or Pitchfork with application preloading construct the config before they fork, so every worker process inherits the same ID. Anything keyed by client ID then conflates sibling processes: `attempted_by`, leader election, and producer bookkeeping such as River Pro's dead-producer rescue, where a live sibling can vouch for a dead sibling's jobs. Go clients always get a unique ID per process. Generate the default ID lazily, on first use in each process. `Config#id` compares the current `Process.pid` against the PID that generated the cached ID and regenerates it in a forked child. A PID check needs no global `Process._fork` hook, covers every fork path, and matches the per-process client handling already used by the Rails integration. The cached PID and ID are swapped in together as one frozen value behind a mutex, so every reader in a process (`Config#id`, `Client#id`, job claims, leadership, and plugins such as River Pro) sees the same stable ID. Host names are truncated so the generated ID always fits the 127-character limit. An explicit `id:` is used unchanged in every process, preserving the Go contract that callers who set IDs keep them unique. Copies made with `Config#with` share the original's ID, including its per-process default, unless the copy overrides `id:`. New specs fork child processes that build clients from a shared config and verify that each child gets a distinct ID tied to its PID, that the parent's ID is unchanged, that explicit IDs survive forks, that the ID is stable across threads, and that claims, leadership, and accessors all agree on the generated ID. --- ruby/CHANGELOG.md | 4 ++ ruby/lib/config.rb | 63 ++++++++++++++++++--- ruby/sig/runtime.rbs | 19 ++++++- ruby/spec/config_spec.rb | 115 ++++++++++++++++++++++++++++++++++++++ ruby/spec/runtime_spec.rb | 18 ++++++ 5 files changed, 209 insertions(+), 10 deletions(-) diff --git a/ruby/CHANGELOG.md b/ruby/CHANGELOG.md index ff57759aa..9a6be3ee6 100644 --- a/ruby/CHANGELOG.md +++ b/ruby/CHANGELOG.md @@ -11,6 +11,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Allow `River::Workers#add` to register a kind and a work block, using the client's default retry and timeout policies. [PR #1486](https://github.com/riverqueue/river/pull/1486). +### Fixed + +- A `River::Config` without an explicit `id:` generates its default client ID on first use in each process, so clients built before a fork (as with preloading Puma, Unicorn, or Pitchfork servers) no longer share one ID across forked workers. Explicit IDs are unchanged, and `Config#with` copies share the original's ID. [PR #1498](https://github.com/riverqueue/river/pull/1498). + ## [0.13.0] - 2026-10-08 ### Added diff --git a/ruby/lib/config.rb b/ruby/lib/config.rb index 7fd698a81..0e1512350 100644 --- a/ruby/lib/config.rb +++ b/ruby/lib/config.rb @@ -40,12 +40,37 @@ def resolved_fetch_poll_interval(config) end end + # Lazily generates a client ID once per process. The process ID check + # regenerates it in forked children without needing a fork hook. + class DefaultClientID # :nodoc: + HOSTNAME_MAX_LENGTH = 80 + + def initialize + @mutex = Mutex.new + @state = nil + end + + def value + pid = Process.pid + @state&.[](pid) || @mutex.synchronize { @state&.[](pid) || generate(pid) } + end + + # The PID and ID are swapped in as one frozen hash so a lock-free reader + # never pairs one process's PID with another's ID. + private def generate(pid) + id = "#{Socket.gethostname[0, HOSTNAME_MAX_LENGTH]}-#{pid}-#{SecureRandom.hex(6)}".freeze + @state = {pid => id}.freeze + id + end + end + private_constant :DefaultClientID + # Complete configuration for a River client and its worker runtime. class Config # Configured values used by Client and its runtime. attr_reader :cancelled_job_retention_period, :completed_job_retention_period, :discarded_job_retention_period, :error_handler, :fetch_cooldown, - :fetch_only_known_kinds, :fetch_poll_interval, :id, :job_timeout, + :fetch_only_known_kinds, :fetch_poll_interval, :job_timeout, :leader_election_disabled, :logger, :maintenance_services, :periodic_jobs, :plugins, :queues, :retry_policy, :workers @@ -53,9 +78,12 @@ class Config # # Configure queues and workers here to enable job processing; a client with # neither may still be used for insertion and administrative operations. - # +fetch_only_known_kinds+ leaves unregistered kinds untouched in shared - # queues. +leader_election_disabled+ prevents this client's maintenance; - # another eligible client must run scheduling, rescue, and cleanup. + # An explicit +id+ is used exactly as given and must be unique among all + # running clients. Without one, each process using this config generates + # its own ID; see #id. +fetch_only_known_kinds+ leaves unregistered kinds + # untouched in shared queues. +leader_election_disabled+ prevents this + # client's maintenance; another eligible client must run scheduling, + # rescue, and cleanup. def initialize( queues: {}, workers: Workers.new, @@ -82,7 +110,8 @@ def initialize( @fetch_cooldown = Float(fetch_cooldown) @fetch_only_known_kinds = fetch_only_known_kinds @fetch_poll_interval = Float(fetch_poll_interval) - @id = id || "#{Socket.gethostname}-#{Process.pid}-#{SecureRandom.hex(6)}" + @default_id = DefaultClientID.new + @id = id @job_timeout = job_timeout.nil? ? nil : Float(job_timeout) @leader_election_disabled = leader_election_disabled @logger = logger || Logger.new($stdout, level: Logger::WARN) @@ -96,10 +125,22 @@ def initialize( validate end + # Returns the client ID recorded in +attempted_by+ and used for leadership. + # + # An explicitly configured ID is returned unchanged. Otherwise, the ID is + # generated on first use in each process and stays stable for the life of + # that process, so clients built from a config before a fork (for example, + # by a preloading Puma, Unicorn, or Pitchfork server) get distinct IDs in + # each forked child instead of sharing the parent's. + def id + @id || @default_id.value + end + # Returns a new Config with the supplied values replacing this config's - # corresponding settings. + # corresponding settings. Without an +id+ override, the copy shares this + # config's ID, including its per-process generated default. def with(**overrides) - Config.new(id: id, + copy = Config.new(id: @id, cancelled_job_retention_period: cancelled_job_retention_period, completed_job_retention_period: completed_job_retention_period, discarded_job_retention_period: discarded_job_retention_period, @@ -116,6 +157,12 @@ def with(**overrides) queues: queues, retry_policy: retry_policy, workers: workers, **overrides) + copy.adopt_default_id(@default_id) unless overrides.key?(:id) + copy + end + + protected def adopt_default_id(default_id) + @default_id = default_id end private def normalize_queues(queues) @@ -136,7 +183,7 @@ def with(**overrides) private def validate raise ArgumentError, "fetch intervals and job_timeout must be finite" unless [fetch_cooldown, fetch_poll_interval, job_timeout].compact.all?(&:finite?) - raise ArgumentError, "id must be between 1 and 127 characters" unless (1...128).cover?(id.length) + raise ArgumentError, "id must be between 1 and 127 characters" if (explicit_id = @id) && !(1...128).cover?(explicit_id.length) raise ArgumentError, "fetch_cooldown must be at least 0.001 seconds" if fetch_cooldown < 0.001 raise ArgumentError, "fetch_poll_interval cannot be less than fetch_cooldown" if fetch_poll_interval < fetch_cooldown raise ArgumentError, "job_timeout must be greater than zero or nil" if job_timeout && job_timeout <= 0 diff --git a/ruby/sig/runtime.rbs b/ruby/sig/runtime.rbs index ed40bf87d..ab66b9d02 100644 --- a/ruby/sig/runtime.rbs +++ b/ruby/sig/runtime.rbs @@ -159,15 +159,28 @@ module River def resolved_fetch_poll_interval: (Config) -> Float end + class DefaultClientID + HOSTNAME_MAX_LENGTH: Integer + @mutex: Mutex + @state: Hash[Integer, String]? + def initialize: () -> void + def value: () -> String + + private + + def generate: (Integer) -> String + end + class Config @cancelled_job_retention_period: Float? @completed_job_retention_period: Float? + @default_id: DefaultClientID @discarded_job_retention_period: Float? @error_handler: untyped @fetch_cooldown: Float @fetch_only_known_kinds: bool @fetch_poll_interval: Float - @id: String + @id: String? @job_timeout: Float? @leader_election_disabled: bool @logger: untyped @@ -185,7 +198,6 @@ module River attr_reader fetch_cooldown: Float attr_reader fetch_only_known_kinds: bool attr_reader fetch_poll_interval: Float - attr_reader id: String attr_reader job_timeout: Float? attr_reader leader_election_disabled: bool attr_reader logger: untyped @@ -203,8 +215,11 @@ module River ?maintenance_services: untyped, ?periodic_jobs: untyped, ?cancelled_job_retention_period: untyped, ?completed_job_retention_period: untyped, ?discarded_job_retention_period: untyped, ?logger: untyped) -> void + def id: () -> String def with: (**untyped) -> Config + def adopt_default_id: (DefaultClientID) -> void + private def normalize_queues: (untyped) -> Hash[String, QueueConfig] diff --git a/ruby/spec/config_spec.rb b/ruby/spec/config_spec.rb index 82da8cc2c..4cba15f8b 100644 --- a/ruby/spec/config_spec.rb +++ b/ruby/spec/config_spec.rb @@ -150,6 +150,121 @@ .to raise_error(ArgumentError, /periodic_jobs must be empty/) end + describe "#id" do + # Runs the block in a forked child and returns its result, marshaled back + # through a pipe. The child exits with exit! to skip RSpec and coverage + # at_exit hooks. + def in_fork + reader, writer = IO.pipe + pid = Process.fork do + reader.close + writer.binmode.write(Marshal.dump(yield)) + writer.close + exit!(0) + end + writer.close + result = Marshal.load(reader.binmode.read) # rubocop:disable Security/MarshalLoad + _, status = Process.wait2(pid) + raise "forked child failed: #{status.inspect}" unless status.success? + + result + ensure + reader&.close + end + + def skip_unless_fork_supported + skip "Process.fork is unsupported on this platform" unless Process.respond_to?(:fork) + end + + it "generates a distinct default ID for each config" do + expect(described_class.new.id).not_to eq(described_class.new.id) + end + + it "generates a default ID including the host name and process ID" do + hostname = Socket.gethostname[0, 80] + + expect(described_class.new.id) + .to match(/\A#{Regexp.escape(hostname)}-#{Process.pid}-\h{12}\z/) + end + + it "truncates long host names to keep the default ID valid" do + gethostname = Socket.method(:gethostname) + Socket.define_singleton_method(:gethostname) { "h" * 200 } + + expect(described_class.new.id).to match(/\Ah{80}-#{Process.pid}-\h{12}\z/) + ensure + Socket.define_singleton_method(:gethostname, gethostname) + end + + it "generates a new default ID in each forked process" do + skip_unless_fork_supported + + config = described_class.new + parent_id = config.id + children = Array.new(2) do + in_fork do + client = River::Client.new(Object.new, config: config) + {client_id: client.id, config_id: config.id, copy_id: config.with(job_timeout: 5).id, pid: Process.pid} + end + end + + children.each do |child| + expect(child.fetch(:client_id)).to eq(child.fetch(:config_id)) + expect(child.fetch(:copy_id)).to eq(child.fetch(:config_id)) + expect(child.fetch(:config_id)).to include("-#{child.fetch(:pid)}-") + end + expect(children.map { |child| child.fetch(:config_id) }.push(parent_id).uniq.length).to eq(3) + expect(config.id).to eq(parent_id) + end + + it "generates a forked process's ID even when the parent never read one" do + skip_unless_fork_supported + + config = described_class.new + child_id = in_fork { config.id } + + expect(config.id).not_to eq(child_id) + end + + it "keeps a default ID stable within a process" do + config = described_class.new + client = River::Client.new(Object.new, config: config) + ids = Array.new(8) { Thread.new { Array.new(50) { config.id } } }.flat_map(&:value) + + expect(ids.uniq).to eq([client.id]) + expect(config.id).to be_frozen + end + + it "regenerates a default ID left over from another process" do + config = described_class.new + config.instance_variable_get(:@default_id).instance_variable_set(:@state, {Process.pid + 1 => "inherited"}.freeze) + + expect(config.id).not_to eq("inherited") + expect(config.id).to equal(config.id) + end + + it "preserves an explicit ID across forks" do + skip_unless_fork_supported + + config = described_class.new(id: "explicit-client") + child = in_fork { [config.id, River::Client.new(Object.new, config: config).id] } + + expect(child).to eq(%w[explicit-client explicit-client]) + expect(config.id).to eq("explicit-client") + end + + it "shares a default ID with copies unless overridden" do + config = described_class.new + copy = config.with(job_timeout: 5) + reset = config.with(id: nil) + + expect(copy.id).to eq(config.id) + expect(config.with(id: "explicit").id).to eq("explicit") + expect(reset.id).not_to eq(config.id) + expect(reset.with(job_timeout: 5).id).to eq(reset.id) + end + end + ["", "a" * 128].each do |id| it "rejects an ID with length #{id.length}" do expect { described_class.new(id: id) } diff --git a/ruby/spec/runtime_spec.rb b/ruby/spec/runtime_spec.rb index bb544348d..5ed490870 100644 --- a/ruby/spec/runtime_spec.rb +++ b/ruby/spec/runtime_spec.rb @@ -109,6 +109,24 @@ def work(job) client&.stop_and_cancel end + it "uses one generated client ID for claims, leadership, and accessors" do + client = build_client(Class.new { + def work(_job) + end + }) + inserted = client.insert(RuntimeArgs.new, insert_opts: River::InsertOpts.new(queue: "runtime")).job + client.start + + completed = wait_until { (row = client.job_get(inserted.id)).state == River::JOB_STATE_COMPLETED && row } + leader_ids = wait_until { (ids = RUNTIME_DB[:river_leader].select_map(:leader_id)).any? && ids } + + expect(client.id).to eq(client.config.id) + expect(completed.attempted_by).to eq([client.id]) + expect(leader_ids).to eq([client.id]) + ensure + client&.stop_and_cancel + end + it "records failures and discards jobs at max attempts" do worker = Class.new do def work(_job) = raise("nope")