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/docs/README.md

@@ -206,6 +206,45 @@ 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, use it inside a `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.
+
+`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.
+Metadata added afterward, including persisted logs, is saved separately when
+the worker finishes.
+
+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.
+
 ### [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/lib/driver.rb

@@ -48,6 +48,11 @@ def errors = {}
       end
     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/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/lib/driver.rb

@@ -31,6 +31,11 @@ def initialize(db)
       end
     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/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,37 @@ 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, metadata: job.metadata_updates, now: now)
+      raise JobCancelError if completed == :cancelled
+
+      completed ||= job_get(job.row.id)
+      raise Error, "job must be running" unless completed.state == JOB_STATE_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,21 @@ 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)
+      return false unless completed && 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.
+      metadata = job.metadata_updates
+      completed = @driver.job_metadata_merge(job.row.id, metadata) || completed unless metadata.empty?
+
+      publish(EVENT_JOB_COMPLETED, completed, timing)
+      true
+    end
+
     private def finish_work(id)
       @mutex.synchronize do
         entry = @running[id]

ruby/lib/worker.rb

@@ -81,6 +81,9 @@ def work(job)
   # A job being worked, with access to its persisted row and attempt-local
   # metadata changes.
   class Job
+    # Internal runtime hint; persisted state determines whether completion committed.
+    attr_accessor :__completion_attempted
+
     # Client working this job.
     attr_reader :client
 
@@ -89,6 +92,7 @@ class Job
 
     def initialize(client, row)
       @client = client
+      @__completion_attempted = false
       @logger = nil
       @metadata_updates = {}
       @row = row

ruby/sig/client.rbs

@@ -24,6 +24,7 @@ module River
       ?scheduled_at: Time?, ?state: jobStateInitialInput?, ?tags: Array[String]?, ?unique_opts: UniqueOpts?) -> JobInsertResult
     def insert_many: (Array[jobArgs | InsertManyParams]) -> Array[JobInsertResult]
     def job_cancel: (Integer) -> JobRow
+    def job_complete_tx: (Job) -> JobRow
     def job_delete: (Integer) -> JobRow
     def job_delete_many: (?JobListParams?, ?after: untyped, ?after_id: untyped,
       ?ids: untyped, ?kinds: untyped, ?limit: untyped, ?metadata: untyped, ?priorities: untyped,

ruby/sig/driver.rbs

@@ -1,6 +1,7 @@
 module River
   interface _Driver
     def respond_to?: (Symbol, ?bool) -> bool
+    def in_transaction?: () -> bool
     def init_driver: () -> void
     def migration_backend: () -> Symbol
     def migration_connection: [A] () { (untyped) -> A } -> A

ruby/sig/runtime.rbs

@@ -94,11 +94,13 @@ module River
 
   class Job
     RESUMABLE_CURSOR_UNSET: Object
+    @__completion_attempted: bool
     @client: Client
     @logger: untyped
     @metadata_updates: Hash[String, untyped]
     @resumable_state: ResumableState
     @row: JobRow
+    attr_accessor __completion_attempted: bool
     attr_reader client: Client
     attr_reader row: JobRow
     def initialize: (Client, JobRow) -> void
@@ -380,6 +382,7 @@ module River
     def finish_failed: (JobRow, Job, Exception, JobTiming, ?cancelled: bool, ?worker: untyped) -> untyped
     def finish_interrupted: (JobRow, Job, JobTiming) -> untyped
     def finish_snoozed: (JobRow, Job, JobSnoozeError, JobTiming) -> untyped
+    def finish_transactional_completion: (Job, JobTiming) -> bool
     def finish_work: (Integer) -> untyped
     def invoke_plugins: (Symbol, *untyped) -> Array[untyped]
     def invoke_worker: (untyped, Job) -> untyped

ruby/spec/client_driver_shared_examples.rb

@@ -4,6 +4,7 @@
 require "riverqueue/testing"
 require_relative "runtime_draining_shared_examples"
 require_relative "runtime_finishing_shared_examples"
+require_relative "transactional_completion_shared_examples"
 
 RSpec.shared_examples "Postgres state update races" do
   it "skips a concurrent retry when selecting jobs for bulk deletion" do
@@ -239,6 +240,7 @@
 RSpec.shared_examples "client driver end to end" do
   it_behaves_like "cancellation while draining"
   it_behaves_like "externally claimed job finalization"
+  it_behaves_like "transactional job completion"
   [false, true].product([false, true]).each do |with_cursor, rollback|
     it "resumes after a #{rollback ? "rolled-back" : "committed"} #{with_cursor ? "cursor" : "step"} checkpoint" do
       received = []