Ruby: generate default client ID per process

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

ruby/lib/config.rb

@@ -40,22 +40,50 @@ 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
 
     # Creates a client configuration.
     #
     # 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

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]

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

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