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)