diff --git a/ruby/CHANGELOG.md b/ruby/CHANGELOG.md
index ff57759aa..51691aa46 100644
--- a/ruby/CHANGELOG.md
+++ b/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
 
diff --git a/ruby/Gemfile.lock b/ruby/Gemfile.lock
index 77e61c8fd..d23a3092b 100644
--- a/ruby/Gemfile.lock
+++ b/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/
diff --git a/ruby/docs/README.md b/ruby/docs/README.md
index 1c9c7dece..be138e4e2 100644
--- a/ruby/docs/README.md
+++ b/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:
diff --git a/ruby/driver/riverqueue-activerecord/Gemfile.lock b/ruby/driver/riverqueue-activerecord/Gemfile.lock
index 000139b90..32ea4d1a8 100644
--- a/ruby/driver/riverqueue-activerecord/Gemfile.lock
+++ b/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)
 
diff --git a/ruby/driver/riverqueue-activerecord/lib/driver.rb b/ruby/driver/riverqueue-activerecord/lib/driver.rb
index 2f45e0d9e..c997f4946 100644
--- a/ruby/driver/riverqueue-activerecord/lib/driver.rb
+++ b/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
diff --git a/ruby/driver/riverqueue-activerecord/riverqueue-activerecord.gemspec b/ruby/driver/riverqueue-activerecord/riverqueue-activerecord.gemspec
index e85a28c6d..b16a38c91 100644
--- a/ruby/driver/riverqueue-activerecord/riverqueue-activerecord.gemspec
+++ b/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
diff --git a/ruby/driver/riverqueue-activerecord/spec/connection_class_spec.rb b/ruby/driver/riverqueue-activerecord/spec/connection_class_spec.rb
index db59a5655..a075cfe3f 100644
--- a/ruby/driver/riverqueue-activerecord/spec/connection_class_spec.rb
+++ b/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
diff --git a/ruby/driver/riverqueue-sequel/Gemfile.lock b/ruby/driver/riverqueue-sequel/Gemfile.lock
index 95ff36b51..99d44b245 100644
--- a/ruby/driver/riverqueue-sequel/Gemfile.lock
+++ b/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/
diff --git a/ruby/driver/riverqueue-sequel/lib/driver.rb b/ruby/driver/riverqueue-sequel/lib/driver.rb
index a2c576e50..8bcdc7d0b 100644
--- a/ruby/driver/riverqueue-sequel/lib/driver.rb
+++ b/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
diff --git a/ruby/driver/riverqueue-sequel/riverqueue-sequel.gemspec b/ruby/driver/riverqueue-sequel/riverqueue-sequel.gemspec
index 2a242b5a4..125c7cb83 100644
--- a/ruby/driver/riverqueue-sequel/riverqueue-sequel.gemspec
+++ b/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
diff --git a/ruby/driver/riverqueue-sequel/spec/client_spec.rb b/ruby/driver/riverqueue-sequel/spec/client_spec.rb
index 17cf40f4a..4195d33b2 100644
--- a/ruby/driver/riverqueue-sequel/spec/client_spec.rb
+++ b/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
diff --git a/ruby/lib/client.rb b/ruby/lib/client.rb
index 02c3dae8b..703aafdcb 100644
--- a/ruby/lib/client.rb
+++ b/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
diff --git a/ruby/lib/client_runtime.rb b/ruby/lib/client_runtime.rb
index dbf82328b..0c1565fe1 100644
--- a/ruby/lib/client_runtime.rb
+++ b/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]
diff --git a/ruby/lib/driver/runtime.rb b/ruby/lib/driver/runtime.rb
index 93d942981..3e84b1dc2 100644
--- a/ruby/lib/driver/runtime.rb
+++ b/ruby/lib/driver/runtime.rb
@@ -68,19 +68,23 @@ def job_claim(id:, attempted_by:, allow_scheduled: false, now: Time.now.utc)
     # The cancellation predicate must be part of the completion update so it is
     # checked against the row we lock, including concurrent cancellations.
     # :cancelled asks the runtime to apply its normal cancellation/error hooks.
-    def job_complete(id:, finalized_at:, metadata: nil, now: Time.now.utc)
+    def job_complete(id:, finalized_at:, expected_job: nil, metadata: nil, now: Time.now.utc)
       id = Integer(id)
+      predicate = "id = #{id}"
+      predicate += " AND #{runtime_job_attempt_predicate(expected_job)}" if expected_job
       assignments = ["state = 'completed'", "finalized_at = #{runtime_time(finalized_at)}"]
       assignments << "metadata = #{runtime_merge_metadata(metadata)}" unless metadata.nil? || metadata.empty?
 
-      updated_id = runtime_returning_ids(<<~SQL).first
-        UPDATE river_job SET #{assignments.join(", ")}
-        WHERE id = #{id} AND state = 'running' AND NOT (#{runtime_cancel_attempted})
-        RETURNING id
-      SQL
-      return runtime_read_job(updated_id) if updated_id
+      transaction do
+        updated_id = runtime_returning_ids(<<~SQL).first
+          UPDATE river_job SET #{assignments.join(", ")}
+          WHERE #{predicate} AND state = 'running' AND NOT (#{runtime_cancel_attempted})
+          RETURNING id
+        SQL
+        next runtime_read_job(updated_id) if updated_id
 
-      :cancelled if job_get_cancelled_ids([id]).include?(id)
+        :cancelled if runtime_returning_ids("SELECT id FROM river_job WHERE #{predicate} AND (#{runtime_cancel_attempted})").any?
+      end
     end
 
     def job_delete(id)
@@ -165,14 +169,24 @@ def job_list(params = nil)
       runtime_job_list(params)
     end
 
-    def job_metadata_merge(id, metadata)
-      updated_id = runtime_returning_ids(<<~SQL).first
-        UPDATE river_job
-        SET metadata = #{runtime_merge_metadata(metadata)}
-        WHERE id = #{Integer(id)}
-        RETURNING id
-      SQL
-      updated_id ? job_get_by_id(updated_id) : nil
+    # When supplied, expected_job prevents metadata from crossing into another
+    # attempt or state. Keep the update and returned snapshot in one transaction.
+    def job_metadata_merge(id, metadata, expected_job = nil)
+      predicate = "id = #{Integer(id)}"
+      if expected_job
+        predicate += " AND #{runtime_job_attempt_predicate(expected_job)}"
+        predicate += " AND state = #{runtime_quote(expected_job.state)}"
+      end
+
+      transaction do
+        updated_id = runtime_returning_ids(<<~SQL).first
+          UPDATE river_job
+          SET metadata = #{runtime_merge_metadata(metadata)}
+          WHERE #{predicate}
+          RETURNING id
+        SQL
+        updated_id ? job_get_by_id(updated_id) : nil
+      end
     end
 
     def job_rescue_stuck(horizon:, retry_policy:, now: Time.now.utc, max: 1_000, logger: nil, rescue_if: nil)
@@ -580,6 +594,10 @@ def queue_upsert(name, metadata: {}, now: Time.now.utc)
       "(#{column} IS NULL OR #{column} #{comparison} #{value} OR (#{column} = #{value} AND #{id_clause}))"
     end
 
+    private def runtime_job_attempt_predicate(job)
+      "attempt = #{Integer(job.attempt)} AND attempted_at IS NOT DISTINCT FROM #{runtime_nullable_time(job.attempted_at)}"
+    end
+
     private def runtime_job_list(params, for_delete: false)
       clauses = [] #: Array[String]
       clauses << "state != 'running'" if for_delete
diff --git a/ruby/lib/worker.rb b/ruby/lib/worker.rb
index 082595537..a96b28005 100644
--- a/ruby/lib/worker.rb
+++ b/ruby/lib/worker.rb
@@ -81,6 +81,12 @@ 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
+
+    # Internal snapshot retained only after transactional completion commits.
+    attr_accessor :__completion_snapshot
+
     # Client working this job.
     attr_reader :client
 
@@ -89,6 +95,8 @@ class Job
 
     def initialize(client, row)
       @client = client
+      @__completion_attempted = false
+      @__completion_snapshot = nil
       @logger = nil
       @metadata_updates = {}
       @row = row
diff --git a/ruby/rails/riverqueue-rails/Gemfile.lock b/ruby/rails/riverqueue-rails/Gemfile.lock
index 4fd1be8f3..61f251102 100644
--- a/ruby/rails/riverqueue-rails/Gemfile.lock
+++ b/ruby/rails/riverqueue-rails/Gemfile.lock
@@ -2,7 +2,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)
 
diff --git a/ruby/sig/client.rbs b/ruby/sig/client.rbs
index b0652d648..f729e810f 100644
--- a/ruby/sig/client.rbs
+++ b/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,
diff --git a/ruby/sig/driver.rbs b/ruby/sig/driver.rbs
index 00853c97a..1579a43f1 100644
--- a/ruby/sig/driver.rbs
+++ b/ruby/sig/driver.rbs
@@ -1,6 +1,8 @@
 module River
   interface _Driver
+    def after_commit: () { () -> void } -> void
     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
@@ -13,11 +15,11 @@ module River
     def job_get_available: (queue: String, max: Integer, attempted_by: String, ?kinds: Array[String]?, ?now: Time) -> Array[JobRow]
     def job_claim: (id: Integer, attempted_by: String, ?allow_scheduled: bool, ?now: Time) -> JobRow?
     def job_get_cancelled_ids: (Array[Integer]) -> Array[Integer]
-    def job_complete: (id: Integer, finalized_at: Time, ?metadata: Hash[untyped, untyped]?, ?now: Time) -> (JobRow | :cancelled)?
+    def job_complete: (id: Integer, finalized_at: Time, ?expected_job: JobRow?, ?metadata: Hash[untyped, untyped]?, ?now: Time) -> (JobRow | :cancelled)?
     def job_get_by_kind_and_unique_properties: (Driver::JobGetByKindAndUniquePropertiesParam) -> JobRow?
     def job_insert: (Driver::JobInsertParams) -> [JobRow, bool]
     def job_insert_many: (Array[Driver::JobInsertParams]) -> Array[[JobRow, bool]]
-    def job_metadata_merge: (Integer, Hash[untyped, untyped]) -> JobRow?
+    def job_metadata_merge: (Integer, Hash[untyped, untyped], ?JobRow?) -> JobRow?
     def job_rescue_stuck: (**untyped) -> Integer
     def job_retry: (Integer, ?now: Time) -> JobRow?
     def job_schedule: (?now: Time, ?max: Integer) -> Integer
@@ -106,7 +108,7 @@ module River
       def postgres_capabilities: () -> PostgresCapabilities
       def postgres_notify_insert: (Array[JobInsertParams]) -> void
       def job_get_cancelled_ids: (Array[Integer]) -> Array[Integer]
-      def job_complete: (id: untyped, finalized_at: untyped, ?metadata: untyped, ?now: untyped) -> untyped
+      def job_complete: (id: untyped, finalized_at: untyped, ?expected_job: untyped, ?metadata: untyped, ?now: untyped) -> untyped
       def job_cancel: (untyped, ?now: untyped) -> untyped
       def job_delete: (untyped) -> untyped
       def job_delete_finalized: (retention: untyped, ?now: untyped, ?max: untyped) -> untyped
@@ -116,7 +118,7 @@ module River
       def job_claim: (id: untyped, attempted_by: untyped, ?allow_scheduled: untyped, ?now: untyped) -> untyped
       private def runtime_claim_jobs: (untyped, untyped, untyped, untyped) -> untyped
       def job_list: (?untyped) -> untyped
-      def job_metadata_merge: (untyped, untyped) -> untyped
+      def job_metadata_merge: (untyped, untyped, ?untyped) -> untyped
       def job_rescue_stuck: (horizon: untyped, retry_policy: untyped, ?now: untyped, ?max: untyped, ?logger: untyped, ?rescue_if: untyped) -> untyped
       def job_retry: (untyped, ?now: untyped) -> untyped
       def job_schedule: (?now: untyped, ?max: untyped) -> untyped
@@ -142,6 +144,7 @@ module River
       def runtime_cursor_clause: (untyped) -> String
       def runtime_execute: (String) -> untyped
       def runtime_in_clause: (untyped, untyped) -> untyped
+      def runtime_job_attempt_predicate: (untyped) -> String
       def runtime_job_list_without_params: () -> untyped
       def runtime_job_list: (JobListParams, ?for_delete: bool) -> Array[JobRow]
       def runtime_job_rows: (String) -> untyped
diff --git a/ruby/sig/runtime.rbs b/ruby/sig/runtime.rbs
index ed40bf87d..43e66420d 100644
--- a/ruby/sig/runtime.rbs
+++ b/ruby/sig/runtime.rbs
@@ -94,11 +94,15 @@ module River
 
   class Job
     RESUMABLE_CURSOR_UNSET: Object
+    @__completion_attempted: bool
+    @__completion_snapshot: JobRow?
     @client: Client
     @logger: untyped
     @metadata_updates: Hash[String, untyped]
     @resumable_state: ResumableState
     @row: JobRow
+    attr_accessor __completion_attempted: bool
+    attr_accessor __completion_snapshot: JobRow?
     attr_reader client: Client
     attr_reader row: JobRow
     def initialize: (Client, JobRow) -> void
@@ -380,6 +384,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
diff --git a/ruby/spec/client_driver_shared_examples.rb b/ruby/spec/client_driver_shared_examples.rb
index b94077329..af7c01987 100644
--- a/ruby/spec/client_driver_shared_examples.rb
+++ b/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 = []
diff --git a/ruby/spec/transactional_completion_shared_examples.rb b/ruby/spec/transactional_completion_shared_examples.rb
new file mode 100644
index 000000000..f29be04df
--- /dev/null
+++ b/ruby/spec/transactional_completion_shared_examples.rb
@@ -0,0 +1,486 @@
+# frozen_string_literal: true
+
+require "riverqueue/testing"
+
+RSpec.shared_examples "transactional job completion" do
+  let(:completion_plugins) { [] }
+  let(:completion_worker) { Object.new }
+  let(:completion_client) do
+    River::Client.new(@driver, config: River::Config.new(
+      plugins: completion_plugins, workers: River::Workers.new.add(:transactional, completion_worker)
+    ))
+  end
+
+  def claim_completion_job
+    row = insert_completion_job
+    River::Job.new(completion_client, @driver.job_claim(id: row.id, attempted_by: "test"))
+  end
+
+  # Open an application transaction directly, rather than using River's driver
+  # transaction wrapper, to exercise the public Active Record and Sequel APIs.
+  def completion_transaction(&block)
+    if @driver.respond_to?(:connection_class)
+      @driver.connection_class.transaction(requires_new: true, &block)
+    else
+      @driver.instance_variable_get(:@db).transaction(savepoint: true, &block)
+    end
+  end
+
+  def insert_completion_job
+    completion_client.insert(River::JobArgsHash.new(:transactional, {}), metadata: {"existing" => true}).job
+  end
+
+  it "commits completion, pending metadata, and other writes in the caller's transaction" do
+    events = completion_client.subscribe(:job_completed, :job_failed)
+    transaction = method(:completion_transaction)
+    check_events = -> { expect { events.pop(true) }.to raise_error(ThreadError) }
+    snapshots = []
+    completion_worker.define_singleton_method(:work) do |job|
+      job.output = {"result" => 42}
+      transaction.call do
+        job.client.insert(River::JobArgsHash.new(:followup, {}))
+        snapshots << job.client.job_complete_tx(job)
+        snapshots << job.row
+        check_events.call
+      end
+    end
+    row = insert_completion_job
+
+    result = River::Testing.perform_job(completion_client, row.id)
+
+    expect(result).to have_attributes(error: nil, outcome: :completed)
+    expect(result.job).to have_attributes(
+      finalized_at: be_a(Time), metadata: include("existing" => true, "output" => {"result" => 42}), state: "completed"
+    )
+    expect(snapshots[0]).to have_attributes(finalized_at: result.job.finalized_at, state: "completed")
+    expect(snapshots[1]).to have_attributes(finalized_at: nil, state: "running")
+    expect(completion_client.job_list(kinds: [:followup]).jobs.length).to eq(1)
+    expect(events.pop(true)).to have_attributes(kind: :job_completed, job: have_attributes(id: row.id, state: "completed"))
+    expect { events.pop(true) }.to raise_error(ThreadError)
+  ensure
+    events&.close
+  end
+
+  it "rolls back completion and other writes and retries a failed attempt" do
+    events = completion_client.subscribe(:job_completed, :job_failed)
+    transaction = method(:completion_transaction)
+    completion_worker.define_singleton_method(:work) do |job|
+      transaction.call do
+        job.client.insert(River::JobArgsHash.new(:followup, {}))
+        job.client.job_complete_tx(job)
+        raise "roll back the work"
+      end
+    end
+    row = insert_completion_job
+
+    result = River::Testing.perform_job(completion_client, row.id)
+
+    expect(result).to have_attributes(error: have_attributes(message: "roll back the work"), outcome: :retried)
+    expect(result.job).to have_attributes(finalized_at: nil, errors: contain_exactly(have_attributes(error: "roll back the work")))
+    expect(result.job.state).not_to eq("completed")
+    expect(completion_client.job_list(kinds: [:followup]).jobs).to be_empty
+    expect(events.pop(true).kind).to eq(:job_failed)
+    expect { events.pop(true) }.to raise_error(ThreadError)
+  ensure
+    events&.close
+  end
+
+  [RuntimeError.new("after commit"), River::JobCancelError.new, River::JobSnoozeError.new(60), River::ClientRuntime::Interrupted.new].each do |error|
+    it "keeps a committed completion when the worker raises #{error.class}" do
+      events = completion_client.subscribe(:job_completed, :job_failed, :job_cancelled, :job_snoozed, :job_interrupted)
+      transaction = method(:completion_transaction)
+      completion_worker.define_singleton_method(:work) do |job|
+        transaction.call { job.client.job_complete_tx(job) }
+        raise error
+      end
+      row = insert_completion_job
+
+      result = River::Testing.perform_job(completion_client, row.id)
+
+      expect(result).to have_attributes(error: nil, outcome: :completed)
+      expect(result.job).to have_attributes(attempt: 1, state: "completed")
+      expect(Array(result.job.errors)).to be_empty
+      expect(events.pop(true)).to have_attributes(kind: :job_completed, job: have_attributes(id: row.id, state: "completed"))
+      expect { events.pop(true) }.to raise_error(ThreadError)
+    ensure
+      events&.close
+    end
+  end
+
+  it "allows normal runtime completion after the caller deliberately rolls back" do
+    transaction = method(:completion_transaction)
+    completion_worker.define_singleton_method(:work) do |job|
+      transaction.call do
+        job.client.job_complete_tx(job)
+        raise job.client.driver.rollback_exception
+      end
+      job.output = "after rollback"
+    end
+
+    result = River::Testing.perform_job(completion_client, insert_completion_job.id)
+
+    expect(result).to have_attributes(error: nil, outcome: :completed)
+    expect(result.job).to have_attributes(metadata: include("output" => "after rollback"), state: "completed")
+  end
+
+  it "honors cancellation and rolls back the caller's other writes" do
+    transaction = method(:completion_transaction)
+    completion_worker.define_singleton_method(:work) do |job|
+      job.client.job_cancel(job.id)
+      transaction.call do
+        job.client.insert(River::JobArgsHash.new(:followup, {}))
+        job.client.job_complete_tx(job)
+      end
+    end
+
+    result = River::Testing.perform_job(completion_client, insert_completion_job.id)
+
+    expect(result).to have_attributes(error: be_a(River::JobCancelError), outcome: :cancelled)
+    expect(result.job.state).to eq("cancelled")
+    expect(completion_client.job_list(kinds: [:followup]).jobs).to be_empty
+  end
+
+  it "returns an already committed completion without changing its timestamp or metadata" do
+    job = claim_completion_job
+    first = completion_transaction { completion_client.job_complete_tx(job) }
+    job.output = "too late"
+
+    repeated = completion_transaction { completion_client.job_complete_tx(job) }
+
+    expect(repeated).to have_attributes(finalized_at: first.finalized_at, metadata: first.metadata, state: "completed")
+  end
+
+  it "rejects jobs that are not running before writing" do
+    job = River::Job.new(completion_client, insert_completion_job)
+
+    expect { completion_client.job_complete_tx(job) }.to raise_error(River::Error, "job must be running")
+    expect(completion_client.job_get(job.id).state).to eq("available")
+  end
+
+  it "rejects stale running jobs whose persisted state changed" do
+    job = claim_completion_job
+    completion_client.job_update(job.id, state: :scheduled)
+
+    expect { completion_transaction { completion_client.job_complete_tx(job) } }.to raise_error(River::Error, "job must be running")
+    expect(completion_client.job_get(job.id).state).to eq("scheduled")
+  end
+
+  %w[running completed cancelled].product([false, true]).each do |state, reuse_attempt|
+    it "rejects a stale attempt when its replacement is #{state} with reuse_attempt=#{reuse_attempt}" do
+      job = claim_completion_job
+      @driver.job_set_state_if_running(id: job.id, state: "available")
+      @driver.job_claim(id: job.id, attempted_by: "replacement", now: job.attempted_at + 1)
+      completion_client.job_update(job.id, attempt: job.attempt) if reuse_attempt
+      @driver.job_complete(id: job.id, finalized_at: Time.now.utc) if state == "completed"
+      completion_client.job_cancel(job.id) if state == "cancelled"
+      replacement = completion_client.job_get(job.id)
+      job.output = "stale output"
+
+      expect do
+        completion_transaction do
+          completion_client.insert(River::JobArgsHash.new(:followup, {}))
+          completion_client.job_complete_tx(job)
+        end
+      end.to raise_error(River::Error, "job attempt is no longer current")
+
+      expect(completion_client.job_get(job.id)).to have_attributes(
+        attempt: replacement.attempt, attempted_at: replacement.attempted_at,
+        finalized_at: replacement.finalized_at, metadata: replacement.metadata, state: replacement.state
+      )
+      expect(completion_client.job_list(kinds: [:followup]).jobs).to be_empty
+    end
+  end
+
+  it "checks attempt ownership in the completion write itself" do
+    job = claim_completion_job
+    replacement = nil
+    @driver.define_singleton_method(:job_complete) do |**options|
+      job_set_state_if_running(id: job.id, state: "available")
+      replacement = job_claim(id: job.id, attempted_by: "replacement", now: job.attempted_at + 1)
+      super(**options)
+    end
+
+    completion_transaction do
+      expect { completion_client.job_complete_tx(job) }.to raise_error(River::Error, "job attempt is no longer current")
+    end
+
+    expect(completion_client.job_get(job.id)).to have_attributes(
+      attempt: replacement.attempt, attempted_at: replacement.attempted_at, finalized_at: nil, state: "running"
+    )
+  end
+
+  it "rejects jobs belonging to another client" do
+    job = claim_completion_job
+
+    expect { River::Client.new(@driver).job_complete_tx(job) }.to raise_error(ArgumentError, "job must belong to this client")
+    expect(completion_client.job_get(job.id).state).to eq("running")
+  end
+
+  it "rejects job rows and IDs instead of a running Job" do
+    row = insert_completion_job
+
+    [row, row.id].each do |argument|
+      expect { completion_client.job_complete_tx(argument) }.to raise_error(ArgumentError, "job must be a River::Job")
+    end
+  end
+
+  it "rejects completion outside a transaction without changing the job" do
+    job = claim_completion_job
+    job.output = "not saved"
+
+    expect { completion_client.job_complete_tx(job) }.to raise_error(
+      River::Error, "job_complete_tx requires an active transaction on the driver's connection"
+    )
+
+    expect(completion_client.job_get(job.id)).to have_attributes(
+      finalized_at: nil, metadata: job.row.metadata, state: "running"
+    )
+  end
+
+  it "reports a missing job without publishing a completion event" do
+    transaction = method(:completion_transaction)
+    completion_worker.define_singleton_method(:work) do |job|
+      job.client.driver.job_delete_if_running(job.id)
+      transaction.call { job.client.job_complete_tx(job) }
+    end
+    events = completion_client.subscribe(:job_completed)
+    row = insert_completion_job
+
+    result = River::Testing.perform_job(completion_client, row.id)
+
+    expect(result.error).to be_a(River::NotFoundError)
+    expect { events.pop(true) }.to raise_error(ThreadError)
+  ensure
+    events&.close
+  end
+
+  it "reports the committed snapshot if the job is deleted before saving later metadata" do
+    transaction = method(:completion_transaction)
+    completion_worker.define_singleton_method(:work) do |job|
+      job.output = 42
+      transaction.call { job.client.job_complete_tx(job) }
+    end
+    @driver.define_singleton_method(:job_metadata_merge) do |id, metadata, expected_job = nil|
+      job_delete(id)
+      super(id, metadata, expected_job)
+    end
+    events = completion_client.subscribe(:job_completed)
+    row = insert_completion_job
+
+    result = River::Testing.perform_job(completion_client, row.id)
+
+    expect(result).to have_attributes(error: nil, job: nil, outcome: :completed)
+    expect(events.pop(true).job).to have_attributes(id: row.id, metadata: include("output" => 42), state: "completed")
+  ensure
+    events&.close
+  end
+
+  context "when the completed job is retried" do
+    let(:finalize_calls) { [] }
+    let(:finalize_result) { nil }
+    let(:completion_plugins) do
+      calls = finalize_calls
+      result = finalize_result
+      finalizer = Object.new
+      finalizer.define_singleton_method(:job_finalize) do |*args|
+        calls << args
+        result
+      end
+      [finalizer]
+    end
+
+    def retry_completed_job(id, state)
+      completion_client.job_retry(id)
+      unless state == "available"
+        @driver.job_claim(id: id, attempted_by: "replacement", now: Time.now.utc + 1)
+        @driver.job_complete(id: id, finalized_at: Time.now.utc) if state == "completed"
+      end
+      @driver.job_metadata_merge(id, {"output" => "replacement"})
+    end
+
+    [nil, :delete].each do |finalize_result|
+      context "with finalizer returning #{finalize_result.inspect}" do
+        let(:finalize_result) { finalize_result }
+
+        [nil, RuntimeError.new("after retry"), River::JobCancelError.new, River::JobSnoozeError.new(60), River::ClientRuntime::Interrupted.new].product(%w[available running completed]).each do |error, state|
+          it "preserves the #{state} replacement when the old worker finishes with #{error.class}" do
+            transaction = method(:completion_transaction)
+            retry_job = method(:retry_completed_job)
+            replacement = nil
+            completion_worker.define_singleton_method(:work) do |job|
+              transaction.call { job.client.job_complete_tx(job) }
+              replacement = retry_job.call(job.id, state)
+              job.output = "old attempt"
+              raise error if error
+            end
+            events = completion_client.subscribe(:job_completed, :job_failed, :job_cancelled, :job_snoozed, :job_interrupted)
+
+            result = River::Testing.perform_job(completion_client, insert_completion_job.id)
+
+            expect(result).to have_attributes(error: nil, outcome: :completed)
+            expect(result.job).to have_attributes(
+              attempt: replacement.attempt, attempted_at: replacement.attempted_at, errors: replacement.errors,
+              finalized_at: replacement.finalized_at, metadata: replacement.metadata,
+              scheduled_at: replacement.scheduled_at, state: replacement.state
+            )
+            expect(finalize_calls).to be_empty
+            expect { events.pop(true) }.to raise_error(ThreadError)
+          ensure
+            events&.close
+          end
+        end
+      end
+    end
+
+    it "recognizes a replacement even when its attempt counter matches" do
+      transaction = method(:completion_transaction)
+      retry_job = method(:retry_completed_job)
+      replacement = nil
+      completion_worker.define_singleton_method(:work) do |job|
+        transaction.call { job.client.job_complete_tx(job) }
+        retry_job.call(job.id, "running")
+        replacement = job.client.job_update(job.id, attempt: job.attempt)
+        job.output = "old attempt"
+      end
+
+      result = River::Testing.perform_job(completion_client, insert_completion_job.id)
+
+      expect(result).to have_attributes(error: nil, outcome: :completed)
+      expect(result.job).to have_attributes(
+        attempt: replacement.attempt, attempted_at: replacement.attempted_at, errors: replacement.errors,
+        finalized_at: replacement.finalized_at, metadata: replacement.metadata,
+        scheduled_at: replacement.scheduled_at, state: replacement.state
+      )
+      expect(finalize_calls).to be_empty
+    end
+
+    %w[available running completed].each do |state|
+      it "preserves a #{state} replacement created before the metadata write" do
+        transaction = method(:completion_transaction)
+        committed = nil
+        completion_worker.define_singleton_method(:work) do |job|
+          job.output = "committed"
+          committed = transaction.call { job.client.job_complete_tx(job) }
+          job.output = "too late"
+        end
+        retry_job = method(:retry_completed_job)
+        replacement = nil
+        @driver.define_singleton_method(:job_metadata_merge) do |id, metadata, expected_job = nil|
+          replacement = retry_job.call(id, state) if expected_job
+          super(id, metadata, expected_job)
+        end
+        events = completion_client.subscribe(:job_completed)
+
+        result = River::Testing.perform_job(completion_client, insert_completion_job.id)
+
+        expect(result).to have_attributes(error: nil, outcome: :completed)
+        expect(result.job).to have_attributes(
+          attempt: replacement.attempt, attempted_at: replacement.attempted_at, errors: replacement.errors,
+          finalized_at: replacement.finalized_at, metadata: replacement.metadata,
+          scheduled_at: replacement.scheduled_at, state: replacement.state
+        )
+        expect(events.pop(true).job).to have_attributes(
+          attempt: committed.attempt, finalized_at: committed.finalized_at,
+          metadata: committed.metadata, state: "completed"
+        )
+        expect { events.pop(true) }.to raise_error(ThreadError)
+        expect(finalize_calls).to be_empty
+      ensure
+        events&.close
+      end
+    end
+  end
+
+  context "with work middleware and finalization hooks" do
+    let(:finalize_calls) { [] }
+    let(:completion_plugins) do
+      calls = finalize_calls
+      finalizer = Object.new
+      finalizer.define_singleton_method(:job_finalize) do |*args|
+        calls << args
+        :delete
+      end
+      [River::JobPersistedLogging::Plugin.new, finalizer]
+    end
+
+    [false, true].product([:delete, :retention]).each do |raise_after_commit, deletion|
+      it "reports committed completion after #{deletion} with raise_after_commit=#{raise_after_commit}" do
+        transaction = method(:completion_transaction)
+        committed = nil
+        completion_worker.define_singleton_method(:work) do |job|
+          job.output = "committed"
+          committed = transaction.call { job.client.job_complete_tx(job) }
+          if deletion == :delete
+            job.client.job_delete(job.id)
+          else
+            job.client.driver.job_delete_finalized(retention: {"completed" => 0}, now: Time.now.utc + 1)
+          end
+          job.output = "after deletion"
+          raise "after commit" if raise_after_commit
+        end
+        events = completion_client.subscribe(:job_completed, :job_failed)
+
+        result = River::Testing.perform_job(completion_client, insert_completion_job.id)
+
+        expect(result).to have_attributes(error: nil, job: nil, outcome: :completed)
+        expect(events.pop(true)).to have_attributes(
+          kind: :job_completed, job: have_attributes(
+            attempt: committed.attempt, finalized_at: committed.finalized_at,
+            metadata: committed.metadata, state: "completed"
+          )
+        )
+        expect { events.pop(true) }.to raise_error(ThreadError)
+        expect(finalize_calls).to be_empty
+      ensure
+        events&.close
+      end
+    end
+
+    [:outer, :savepoint].each do |rollback|
+      it "does not retain a snapshot after a #{rollback} rollback followed by deletion" do
+        transaction = method(:completion_transaction)
+        completion_worker.define_singleton_method(:work) do |job|
+          transaction.call do
+            transaction.call do
+              job.client.job_complete_tx(job)
+              raise job.client.driver.rollback_exception if rollback == :savepoint
+            end
+            raise job.client.driver.rollback_exception if rollback == :outer
+          end
+          job.client.driver.job_delete_if_running(job.id)
+          raise "after rollback"
+        end
+        events = completion_client.subscribe(:job_completed)
+
+        result = River::Testing.perform_job(completion_client, insert_completion_job.id)
+
+        expect(result).to have_attributes(error: have_attributes(message: "after rollback"), job: nil, outcome: :retried)
+        expect { events.pop(true) }.to raise_error(ThreadError)
+      ensure
+        events&.close
+      end
+    end
+
+    [false, true].each do |raise_after_commit|
+      it "preserves logs and skips normal finalization hooks with raise_after_commit=#{raise_after_commit}" do
+        transaction = method(:completion_transaction)
+        completion_worker.define_singleton_method(:work) do |job|
+          job.logger << "before commit\n"
+          transaction.call { job.client.job_complete_tx(job) }
+          job.logger << "after commit\n"
+          raise "after commit" if raise_after_commit
+        end
+        events = completion_client.subscribe(:job_completed)
+
+        result = River::Testing.perform_job(completion_client, insert_completion_job.id)
+
+        expect(result).to have_attributes(error: nil, outcome: :completed)
+        expect(result.job.metadata.fetch("river:log")).to eq([{"attempt" => 1, "log" => "before commit\nafter commit\n"}])
+        expect(events.pop(true).job.metadata).to eq(result.job.metadata)
+        expect(finalize_calls).to be_empty
+      ensure
+        events&.close
+      end
+    end
+  end
+end
diff --git a/ruby/spec/upstream_runtime_spec.rb b/ruby/spec/upstream_runtime_spec.rb
index 0e1909f47..222bc17fb 100644
--- a/ruby/spec/upstream_runtime_spec.rb
+++ b/ruby/spec/upstream_runtime_spec.rb
@@ -7,11 +7,13 @@
 require_relative "row_decoding_shared_examples"
 require_relative "runtime_draining_shared_examples"
 require_relative "runtime_finishing_shared_examples"
+require_relative "transactional_completion_shared_examples"
 
 RSpec.describe "upstream runtime parity", database: :sqlite do
   it_behaves_like "SQLite corrupt job runtime"
   it_behaves_like "cancellation while draining"
   it_behaves_like "externally claimed job finalization"
+  it_behaves_like "transactional job completion"
 
   before do
     @database = Sequel.sqlite