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")