Add transactional completion #job_complete_tx(job) for Ruby

ruby/CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### 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).
+- Added `Client#job_complete_tx(job)` to complete a running job in the caller's Active Record or Sequel transaction, including pending metadata and output. An active transaction on the driver's connection is required. Completion rolls back with the transaction; once committed, it takes precedence over subsequent worker errors and emits a completion event when the worker finishes. [PR #1490](https://github.com/riverqueue/river/pull/1490).
 
 ## [0.13.0] - 2026-10-08
 

ruby/Gemfile.lock

@@ -11,7 +11,7 @@ PATH
   remote: driver/riverqueue-activerecord
   specs:
     riverqueue-activerecord (0.13.0)
-      activerecord (> 0, < 1000)
+      activerecord (>= 7.2, < 1000)
       activesupport (> 0, < 1000)
       riverqueue (= 0.13.0)
 
@@ -20,7 +20,7 @@ PATH
   specs:
     riverqueue-sequel (0.13.0)
       riverqueue (= 0.13.0)
-      sequel (> 0, < 1000)
+      sequel (>= 5.20, < 1000)
 
 GEM
   remote: https://rubygems.org/

ruby/docs/README.md

@@ -206,6 +206,55 @@ end
 
 The equivalent works inside `ActiveRecord::Base.transaction` with the Active Record driver.
 
+### Transactional completion
+
+Call `job.client.job_complete_tx(job)` inside a transaction to commit application
+writes and job completion together (the equivalent of Go's `JobCompleteTx`):
+
+```ruby
+class FulfillOrderWorker
+  def work(job)
+    DB.transaction do
+      DB[:orders].where(id: job.args.fetch("order_id")).update(fulfilled: true)
+      job.output = {fulfilled: true}
+      job.client.job_complete_tx(job)
+    end
+  end
+end
+```
+
+With Active Record 7.2 or later, use it inside an `ApplicationRecord.transaction` block. As
+with insertion, the transaction must use the driver's connection on the same
+thread. No explicit transaction argument is needed. Calling `job_complete_tx`
+without an active transaction on that connection raises `River::Error` without
+changing the job.
+
+Sequel 5.20 or later is required so transaction callbacks respect savepoints.
+
+`job_complete_tx` accepts the running `River::Job` passed to the worker and returns
+an updated `River::JobRow`, leaving the original job unchanged. Set output and
+metadata before calling it so they're included in the transaction. A missing
+job raises `River::NotFoundError`; a pending cancellation raises
+`River::JobCancelError`, allowing the transaction to roll back.
+The supplied job must still belong to the current attempt; a stale worker cannot
+complete a replacement attempt.
+Metadata added afterward, including persisted logs, is saved separately when
+the worker finishes, provided the completed attempt has not been retried.
+
+Make completion the last operation in the transaction and let the transaction
+commit before returning from `work`. Propagate errors that roll back the
+transaction so River retries the job. If you swallow a rollback and return
+successfully, River completes the job normally. Once completion commits, later
+worker errors cannot retry or cancel the job. River emits `:job_completed` when
+the worker finishes, and skips its usual finalization hooks for the already
+completed job.
+If the completed row is deleted before the worker finishes, River emits the
+completion event using the snapshot retained when the transaction committed.
+
+If the job is manually retried before the original worker finishes, that worker
+leaves the replacement attempt untouched and does not emit a completion event
+for it.
+
 ### [Bulk insertion](https://riverqueue.com/docs/inserting-many-jobs)
 
 `#insert_many` inserts a batch atomically and returns one `River::JobInsertResult` per input. Use `River::InsertManyParams` when jobs need different options:

ruby/driver/riverqueue-activerecord/Gemfile.lock

@@ -11,7 +11,7 @@ PATH
   remote: .
   specs:
     riverqueue-activerecord (0.13.0)
-      activerecord (> 0, < 1000)
+      activerecord (>= 7.2, < 1000)
       activesupport (> 0, < 1000)
       riverqueue (= 0.13.0)
 

ruby/driver/riverqueue-activerecord/lib/driver.rb

@@ -48,6 +48,16 @@ def errors = {}
       end
     end
 
+    # Runs the callback only after this connection's outer transaction commits.
+    def after_commit(&block)
+      @connection_class.current_transaction.after_commit(&block)
+    end
+
+    # Reports whether this thread has a transaction on the configured connection.
+    def in_transaction?
+      @connection_class.connection_pool.with_connection(&:transaction_open?)
+    end
+
     def job_get_by_id(id)
       if @is_sqlite
         row = sqlite_job_rows("WHERE id = ? LIMIT 1", [id]).first

ruby/driver/riverqueue-activerecord/riverqueue-activerecord.gemspec

@@ -12,7 +12,7 @@ Gem::Specification.new do |s|
   s.license = "MPL-2.0"
   s.required_ruby_version = ">= 3.2"
   # The stupid version bounds are used to silence Ruby's extremely obnoxious warnings.
-  s.add_dependency "activerecord", "> 0", "< 1000"
+  s.add_dependency "activerecord", ">= 7.2", "< 1000" # transaction callbacks
   s.add_dependency "activesupport", "> 0", "< 1000" # required for ActiveRecord to load properly
   s.add_dependency "riverqueue", "= #{s.version}"
 end

ruby/driver/riverqueue-activerecord/spec/connection_class_spec.rb

@@ -69,6 +69,34 @@ class SelectedRiverConnection < ActiveRecord::Base
         expect(@selected.job_list).to be_empty
       end
 
+      it "completes jobs in the selected connection's transaction" do
+        row = @client.insert(River::JobArgsHash.new(:selected, {})).job
+        job = River::Job.new(@client, @selected.job_claim(id: row.id, attempted_by: "test"))
+
+        SelectedRiverConnection.transaction do
+          @client.job_complete_tx(job)
+          raise ActiveRecord::Rollback
+        end
+
+        expect(@client.job_get(row.id)).to have_attributes(finalized_at: nil, state: "running")
+
+        SelectedRiverConnection.transaction { @client.job_complete_tx(job) }
+
+        expect(@client.job_get(row.id)).to have_attributes(finalized_at: be_a(Time), state: "completed")
+        expect(@primary.job_list).to be_empty
+      end
+
+      it "rejects job completion in an unrelated Base transaction" do
+        row = @client.insert(River::JobArgsHash.new(:selected, {})).job
+        job = River::Job.new(@client, @selected.job_claim(id: row.id, attempted_by: "test"))
+
+        ActiveRecord::Base.transaction do
+          expect { @client.job_complete_tx(job) }.to raise_error(River::Error, /requires an active transaction/)
+        end
+
+        expect(@client.job_get(row.id)).to have_attributes(finalized_at: nil, state: "running")
+      end
+
       if backend == :postgres
         it "detects capabilities again when the selected class replaces its pool" do
           capabilities = @selected.postgres_capabilities

ruby/driver/riverqueue-sequel/Gemfile.lock

@@ -12,7 +12,7 @@ PATH
   specs:
     riverqueue-sequel (0.13.0)
       riverqueue (= 0.13.0)
-      sequel (> 0, < 1000)
+      sequel (>= 5.20, < 1000)
 
 GEM
   remote: https://rubygems.org/

ruby/driver/riverqueue-sequel/lib/driver.rb

@@ -31,6 +31,16 @@ def initialize(db)
       end
     end
 
+    # Runs the callback only if this transaction and its savepoints commit.
+    def after_commit(&block)
+      @db.after_commit(savepoint: true, &block)
+    end
+
+    # Reports whether this thread has a transaction on the configured database.
+    def in_transaction?
+      @db.in_transaction?
+    end
+
     def job_get_by_id(id)
       if @is_sqlite
         row = sqlite_job_rows("WHERE id = ? LIMIT 1", id).first

ruby/driver/riverqueue-sequel/riverqueue-sequel.gemspec

@@ -12,6 +12,6 @@ Gem::Specification.new do |s|
   s.license = "MPL-2.0"
   s.required_ruby_version = ">= 3.2"
   # The stupid version bounds are used to silence Ruby's extremely obnoxious warnings.
-  s.add_dependency "sequel", "> 0", "< 1000"
+  s.add_dependency "sequel", ">= 5.20", "< 1000" # transaction callbacks that respect savepoints
   s.add_dependency "riverqueue", "= #{s.version}"
 end

ruby/driver/riverqueue-sequel/spec/client_spec.rb

@@ -35,6 +35,21 @@
       it_behaves_like "Postgres finalized job list plans" if adapter == :postgres
       it_behaves_like "Postgres rescue concurrency" if adapter == :postgres
       it_behaves_like "dedicated worker process"
+
+      it "rejects job completion in an unrelated database's transaction" do
+        other_database = Sequel.sqlite
+        client = River::Client.new(@driver)
+        row = client.insert(River::JobArgsHash.new(:transactional, {})).job
+        job = River::Job.new(client, @driver.job_claim(id: row.id, attempted_by: "test"))
+
+        other_database.transaction do
+          expect { client.job_complete_tx(job) }.to raise_error(River::Error, /requires an active transaction/)
+        end
+
+        expect(client.job_get(row.id)).to have_attributes(finalized_at: nil, state: "running")
+      ensure
+        other_database&.disconnect
+      end
     end
   end
 end

ruby/lib/client.rb

@@ -182,6 +182,44 @@ def job_cancel(id)
       @driver.job_cancel(id) || raise(NotFoundError, "job not found: #{id}")
     end
 
+    # Completes a running Job and returns its updated JobRow, including pending
+    # metadata and output. The original JobRow is not modified.
+    #
+    # Requires and joins the caller's transaction on the same Active Record
+    # connection or Sequel database. Commit before returning from work;
+    # propagate rollback errors so the runtime can retry the job. A committed
+    # completion takes precedence over any subsequent worker error.
+    #
+    # Raises NotFoundError for a missing job, JobCancelError for a pending
+    # cancellation, and Error if there is no active transaction or the job is
+    # no longer running or completed.
+    def job_complete_tx(job)
+      raise ArgumentError, "job must be a River::Job" unless job.is_a?(Job)
+      raise ArgumentError, "job must belong to this client" unless job.client.equal?(self)
+      raise Error, "job must be running" unless job.row.state == JOB_STATE_RUNNING
+      raise Error, "job_complete_tx requires an active transaction on the driver's connection" unless @driver.in_transaction?
+
+      # Mark the attempt before the write: an interrupt can arrive after the
+      # database commits but before the driver returns. The runtime checks the
+      # persisted state because the caller may still roll back this transaction.
+      job.__completion_attempted = true
+      now = @time_now_utc.call
+      completed = @driver.job_complete(id: job.row.id, finalized_at: now, expected_job: job.row, metadata: job.metadata_updates, now: now)
+      raise JobCancelError if completed == :cancelled
+
+      completed ||= job_get(job.row.id)
+      unless completed.attempt == job.row.attempt && completed.attempted_at == job.row.attempted_at
+        raise Error, "job attempt is no longer current"
+      end
+      raise Error, "job must be running" unless completed.state == JOB_STATE_COMPLETED
+
+      # Retention or application code may delete the row before work returns.
+      # Only retain a snapshot after the caller's outer transaction commits;
+      # rolled-back transactions and savepoints must not report completion.
+      @driver.after_commit { job.__completion_snapshot = completed }
+      completed
+    end
+
     # Deletes a job by ID and returns its former JobRow.
     #
     # Raises NotFoundError if the job does not exist and JobRunningError if it

ruby/lib/client_runtime.rb

@@ -303,6 +303,8 @@ def wake
           finish_work(row.id)
         end
 
+        return [:completed, nil] if finish_transactional_completion(job, timing)
+
         finalize_hooks = @config.plugins.any? { |plugin| plugin.respond_to?(:job_finalize) }
         if finalize_hooks
           raise JobCancelError if @driver.job_get_cancelled_ids([row.id]).include?(row.id)
@@ -322,19 +324,22 @@ def wake
         end
 
         [:completed, nil]
-      rescue JobSnoozeError => error
-        job.__capture_resumable_metadata!
-        [finish_snoozed(row, job, error, timing), error]
-      rescue JobCancelError => error
-        job.__capture_resumable_metadata!
-        finish_failed(row, job, error, timing, cancelled: true)
-        [:cancelled, error]
-      rescue Interrupted => error
-        job.__capture_resumable_metadata!
-        [finish_interrupted(row, job, timing), error]
       rescue => error
+        return [:completed, nil] if finish_transactional_completion(job, timing)
+
         job.__capture_resumable_metadata!
-        [finish_failed(row, job, error, timing, worker: worker), error]
+        outcome = case error
+        when JobSnoozeError
+          finish_snoozed(row, job, error, timing)
+        when JobCancelError
+          finish_failed(row, job, error, timing, cancelled: true)
+          :cancelled
+        when Interrupted
+          finish_interrupted(row, job, timing)
+        else
+          finish_failed(row, job, error, timing, worker: worker)
+        end
+        [outcome, error]
       ensure
         @mutex.synchronize do
           @running.delete(row.id)
@@ -410,6 +415,35 @@ def wake
       (updated&.state == JOB_STATE_CANCELLED) ? :cancelled : :snoozed
     end
 
+    private def finish_transactional_completion(job, timing)
+      return false unless job.__completion_attempted
+
+      completed = @driver.job_get_by_id(job.row.id)
+      unless completed
+        snapshot = job.__completion_snapshot
+        return false unless snapshot
+
+        publish(EVENT_JOB_COMPLETED, snapshot, timing)
+        return true
+      end
+
+      # A retry can replace a committed attempt before its worker or middleware
+      # returns. Only a still-running original attempt may have rolled back and
+      # need normal finalization; never finalize or publish a replacement.
+      return true unless completed.attempt == job.row.attempt && completed.attempted_at == job.row.attempted_at
+      return false if completed.state == JOB_STATE_RUNNING
+      return true unless completed.state == JOB_STATE_COMPLETED
+
+      # Work middleware (including persisted logging) can add metadata after the
+      # worker commits. Save it without changing the committed state or timestamp.
+      # Guard the write too: a retry can happen after the read above.
+      metadata = job.metadata_updates
+      completed = @driver.job_metadata_merge(job.row.id, metadata, completed) || completed unless metadata.empty?
+
+      publish(EVENT_JOB_COMPLETED, completed, timing)
+      true
+    end
+
     private def finish_work(id)
       @mutex.synchronize do
         entry = @running[id]