From a234416b945a8501c29c5d663237144d89ed5e1a Mon Sep 17 00:00:00 2001 From: Brandur Date: Fri, 9 Oct 2026 09:52:56 -0500 Subject: [PATCH] Add an equivalent of `river.WorkFunc` for Ruby Just noticed while doing a manual review on docs changes that Ruby doesn't have an equivalent to Go's `river.WorkFunc` whereas the rest of the languages do. Here, let `workers.add` take a kind and closure: workers.add(:send_message) do |job| puts "Message: #{job.args.fetch("message")}" end Insertions would have to come from another language or use JobArgsHash: client.insert River::JobArgsHash.new(:send_message, message: "Hello!") This admittedly isn't a huge improvement, but I think it's probably worth doing just because both Rust and TypeScript have something extremely similar. This puts all languages on equal footing. Here, add one in. It's arguable as to whether it adds all that much compared to a normal class which is already pretty terse, but it does let you --- ruby/CHANGELOG.md | 4 ++ ruby/docs/README.md | 25 ++++++++++++ ruby/lib/worker.rb | 26 ++++++++++++- ruby/sig/runtime.rbs | 7 ++++ ruby/spec/runtime_spec.rb | 19 ++++++++++ ruby/spec/worker_spec.rb | 80 +++++++++++++++++++++++++++++++++++++++ 6 files changed, 159 insertions(+), 2 deletions(-) diff --git a/ruby/CHANGELOG.md b/ruby/CHANGELOG.md index 1e04757e2..ff57759aa 100644 --- a/ruby/CHANGELOG.md +++ b/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 diff --git a/ruby/docs/README.md b/ruby/docs/README.md index 84836132f..1c9c7dece 100644 --- a/ruby/docs/README.md +++ b/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 diff --git a/ruby/lib/worker.rb b/ruby/lib/worker.rb index 0db37e2ec..082595537 100644 --- a/ruby/lib/worker.rb +++ b/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 diff --git a/ruby/sig/runtime.rbs b/ruby/sig/runtime.rbs index bbb21cf83..ed40bf87d 100644 --- a/ruby/sig/runtime.rbs +++ b/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 diff --git a/ruby/spec/runtime_spec.rb b/ruby/spec/runtime_spec.rb index 8e438347f..bb544348d 100644 --- a/ruby/spec/runtime_spec.rb +++ b/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) diff --git a/ruby/spec/worker_spec.rb b/ruby/spec/worker_spec.rb index f89bb12c9..3a9513a02 100644 --- a/ruby/spec/worker_spec.rb +++ b/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)