Add an equivalent of river.WorkFunc for Ruby

ruby/CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- 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).
+
 ## [0.13.0] - 2026-10-08
 
 ### Added

ruby/docs/README.md

@@ -116,6 +116,31 @@ for nested expressions and `{ ... }` blocks where they make binding clear.
 
 ## Core features
 
+### Block workers
+
+Pass a kind and a block to `Workers#add` to define a worker inline:
+
+```ruby
+workers = River::Workers.new
+workers.add(:message) do |job|
+  puts "Message: #{job.args.fetch("message")}"
+end
+
+# With client configured to use this registry and the ruby queue:
+client.insert River::JobArgsHash.new(:message, message: "Hello!"), queue: :ruby
+```
+
+Pass this registry as `workers:` in `River::Config`. The kind accepts a string
+or symbol and must match the inserted job's kind. The block receives a
+`River::Job` and uses the client's retry and timeout policies. Returning normally
+completes the job; the block's return value is ignored. Raise exceptions to fail,
+cancel, or snooze work as with a worker class. Optionally use `job.output =` to
+persist a result. Aliases are supported through `aliases:` as for other workers.
+Supply a worker or a block, not both. Existing procs or methods can be passed as
+the block with `&`.
+The same block may run concurrently for multiple jobs, so any captured mutable
+state must be safe to share across threads.
+
 ### [Accessing the client from workers](https://riverqueue.com/docs/context-client)
 
 Every running `River::Job` exposes the client that claimed it. Workers can use

ruby/lib/worker.rb

@@ -12,8 +12,18 @@ def initialize
     #
     # With one argument, the kind is read from +worker.kind+ or
     # +worker.class.kind+. Pass a kind and worker separately to override it.
-    # Kinds and aliases accept symbols or strings.
-    def add(kind_or_worker, worker = nil, aliases: [])
+    # Alternatively, pass a kind and a block that receives a River::Job on each
+    # attempt. Block workers use the client's retry and timeout policies.
+    # Supply a worker or a block, not both. Kinds and aliases accept symbols or
+    # strings.
+    def add(kind_or_worker, worker = nil, aliases: [], &block)
+      if block
+        raise ArgumentError, "use a worker or a block, not both" unless worker.nil?
+        raise ArgumentError, "block workers require a string or symbol kind" unless kind_or_worker.is_a?(String) || kind_or_worker.is_a?(Symbol)
+
+        worker = BlockWorker.new(block)
+      end
+
       if worker
         kind = kind_or_worker.to_s
       else
@@ -54,6 +64,18 @@ def include?(kind)
     def kinds
       @workers.keys.freeze
     end
+
+    # Adapts a registered block to the runtime's worker interface.
+    class BlockWorker
+      def initialize(work_func)
+        @work_func = work_func
+      end
+
+      def work(job)
+        @work_func.call(job)
+      end
+    end
+    private_constant :BlockWorker
   end
 
   # A job being worked, with access to its persisted row and attempt-local

ruby/sig/runtime.rbs

@@ -77,12 +77,19 @@ module River
     @workers: Hash[String, untyped]
     def initialize: () -> void
     def add: (untyped, ?untyped, ?aliases: Array[untyped]) -> self
+           | (String | Symbol, ?aliases: Array[untyped]) { (Job) -> untyped } -> self
     def []: (String | Symbol) -> untyped
     def fetch: (String | Symbol) -> untyped
              | (String | Symbol, untyped) -> untyped
              | (String | Symbol) { (String) -> untyped } -> untyped
     def include?: (String | Symbol) -> bool
     def kinds: () -> Array[String]
+
+    class BlockWorker
+      @work_func: ^(Job) -> untyped
+      def initialize: (^(Job) -> untyped) -> void
+      def work: (Job) -> untyped
+    end
   end
 
   class Job

ruby/spec/runtime_spec.rb

@@ -51,6 +51,25 @@ def build_client(worker, queue: "runtime", **config_overrides)
     RUNTIME_DB[:river_job].delete
   end
 
+  it "claims and completes jobs with a work block" do
+    workers = River::Workers.new
+    workers.add(:runtime) do |job|
+      job.output = {"doubled" => job.args.fetch("value") * 2}
+    end
+    client = build_client(workers.fetch(:runtime), workers: workers)
+    inserted = client.insert(River::JobArgsHash.new(:runtime, value: 3), queue: :runtime).job
+    client.start
+
+    completed = wait_until { (row = client.job_get(inserted.id)).state == River::JOB_STATE_COMPLETED && row }
+
+    expect(completed).to have_attributes(
+      attempt: 1,
+      metadata: have_attributes(to_h: include("output" => {"doubled" => 6}))
+    )
+  ensure
+    client&.stop_and_cancel
+  end
+
   it "claims, works, completes, emits events, and persists worker output" do
     worker = Class.new do
       def work(job)

ruby/spec/worker_spec.rb

@@ -3,6 +3,86 @@
 require "spec_helper"
 
 RSpec.describe River::Workers do
+  describe "block workers" do
+    it "accepts a proc as the work block" do
+      work_func = ->(job) { job.args.fetch("value") * 2 }
+      registry = described_class.new.add(:example, &work_func)
+
+      expect(registry.fetch(:example).work(Struct.new(:args).new({"value" => 3}))).to eq(6)
+    end
+
+    it "passes the job to the block and preserves its return value" do
+      received = []
+      result = Object.new
+      registry = described_class.new
+      registration = registry.add(:example) do |job|
+        received << job
+        result
+      end
+      job = Object.new
+
+      expect(registration).to equal(registry)
+      expect(received).to be_empty
+      expect(registry.fetch(:example).work(job)).to equal(result)
+      expect(received).to eq([job])
+    end
+
+    it "propagates exceptions including cancellation and snooze signals" do
+      [RuntimeError.new("failed"), River.job_cancel("cancelled"), River.job_snooze(60)].each do |error|
+        registry = described_class.new.add(:example) { |_job| raise error }
+
+        expect { registry.fetch(:example).work(Object.new) }.to raise_error { |raised| expect(raised).to equal(error) }
+      end
+    end
+
+    it "registers string and symbol kinds with aliases" do
+      ["example", :example].each do |kind|
+        registry = described_class.new.add(kind, aliases: [:old_example, "legacy"]) { |_job| :worked }
+
+        expect(registry.kinds).to contain_exactly("example", "old_example", "legacy")
+        expect(registry.fetch(:old_example)).to equal(registry.fetch(:example))
+        expect(registry.fetch(:legacy).work(Object.new)).to eq(:worked)
+      end
+    end
+
+    it "rejects a block without an explicit string or symbol kind" do
+      worker_class = Class.new { def self.kind = :example }
+      [worker_class, worker_class.new].each do |worker|
+        registry = described_class.new
+
+        expect { registry.add(worker) { |_job| } }
+          .to raise_error(ArgumentError, "block workers require a string or symbol kind")
+        expect(registry.kinds).to be_empty
+      end
+    end
+
+    it "rejects a duplicate alias without partially registering the block worker" do
+      worker = Object.new
+      registry = described_class.new.add(:existing, worker)
+
+      expect { registry.add(:fresh, aliases: [:existing]) { |_job| } }
+        .to raise_error(ArgumentError, 'worker for kind "existing" is already registered')
+      expect(registry.kinds).to eq(["existing"])
+      expect(registry.fetch(:existing)).to equal(worker)
+    end
+
+    it "rejects a duplicate primary kind without replacing the block worker" do
+      registry = described_class.new.add(:example) { |_job| :original }
+
+      expect { registry.add(:example) { |_job| :replacement } }
+        .to raise_error(ArgumentError, 'worker for kind "example" is already registered')
+      expect(registry.fetch(:example).work(Object.new)).to eq(:original)
+    end
+
+    it "rejects a worker and a block supplied together" do
+      registry = described_class.new
+
+      expect { registry.add(:example, Object.new, aliases: [:old_example]) { |_job| } }
+        .to raise_error(ArgumentError, "use a worker or a block, not both")
+      expect(registry.kinds).to be_empty
+    end
+  end
+
   it "registers and fetches a worker under an explicit kind" do
     worker = Object.new
     registry = described_class.new.add("email", worker)